From 61eb00b05be07a6ca0d3a49f4e071ea91974462b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=B2=9C=EA=B8=B0=EC=A0=95?= Date: Sun, 1 Dec 2024 16:49:42 +0900 Subject: [PATCH] =?UTF-8?q?AI:=20[feature]=20=EB=B6=80=EC=A0=95=ED=96=89?= =?UTF-8?q?=EC=9C=84=20=EC=84=A0=ED=83=9D=EA=B8=B0=EB=8A=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/ai/__pycache__/main.cpython-310.pyc | Bin 10501 -> 11169 bytes src/ai/main.py | 157 +++++++++++++++++------- 2 files changed, 113 insertions(+), 44 deletions(-) diff --git a/src/ai/__pycache__/main.cpython-310.pyc b/src/ai/__pycache__/main.cpython-310.pyc index 5c632413cae1090a8d5e20e3d85fec8c97ed6d26..c043b874f764024e1fd24173b88ed0b677677067 100644 GIT binary patch delta 5823 zcma)AYit|GwceS1aQRwM)XS13n|{e+Ecq3`>L`vKKVql0>{w1@Gi@1~vy>@IBt5%~ zEoNytk3qa=}T-HJQ*64Xs zV}q=JQCc9W?Pko}0r#Da`-6@9uEu@$V-g!;YZfK8#?%+&)E?-sWeMmf7HDd(xpzTI z?K9(S9b5l8Nqx**RUc&=mI;x}!TNr-ahb49rox8JhuCKGp`6A>SaMNWsFBpT5O|<1 zkW<(eF$iNgAxGIL8(UOT51TQz)n9Jr0Ncj4FG{IL8WKC$PLMdr9%Q?Kdx-63d*J>k z+spRB{V*ewQu1T}iI$j1Dqi4N#qEQZ`QQ|<@%0GESgNbw^skyt9 z-5aCaESVYGWa;Tl#q_k|#TheezbWy4nAjlhP36>X1NxM-|JwU}HL$X9-SZ1V51;&A zFA#T~+BK@f^W=$RZ|b}USiBF(Dj;5GE|WFWnTwWLfctE*Fzva|sQoq1xMXHn+Ai~g z&yBVF{&H8Y4%Lap4qyw{G!EB$xbgOC{g(FM|L1b~AjrcqIgz%2njz?Y^h1uJ#CrFR za`bd-M|=%3*CN5)^8}K0KvEHN-b~LGpEc*qg8k8o&$vHT$J_(jTU%DV0#qdSHw-P@ zgrx%sNjlLTYAstT`YXwvZTx(#CiLXJOm_3d0XXu+CuG_mivR?cb+x($2ynYUsb%aRL^Td|w#37EfOizG^9+!w@ zQ->}PzgLwVnQw4pTXyKUWGi)*5r@`pN{l-4P3e|`p{URa0~!Mn;85f^Qa!LNH5D>x zQmTvxk*hbkRmBe0Lk`TZs&=>@S(fTirb17%je3Wp+Qv+5nN|aKXFcu!o#}#}?&wS^ zOT3l?RozTufjn7mtdvMscH2F$6YA)WwxBd0er1m$*}d~M0y~4fN%cMl<`C%u>~>Ou zu~qZSM5+dfROu7j&>aHX3p&B(_S$o%#H?ztc0Gub8BDt#yh^T0$RpTa{m z4Ll>jz^lMx^m-7y4B@FbN(-K$dKeruG!p?2QSE5m0Pl&;bU;sYG~q8#FqLU}QVoFH z#GF9eWk6%6&=>Ii_Da}}g9~YnvLO5GI1YXl0WayA&$m3N`#ulqnST{LNEaRi3R%;G z^wtu=TiqW=*6N{&bNn&bz5A!gpeg|6-ie&4@e{z~j{`~e@{eO%gvry`I)mg1B#1VC z7Re+M5&a*<);T0kB6$kQ6q55uE+9cp!X7M!3&e&z^KL7PMt?@nXlkqLvJ%yWA>BF|;bwOrqhY1iX&)UCnzaM{?*idc^ z*7yc^h+nF}sy5=%+6HO2L@vW}$N-r0P0+|fqxHj=5r#|TqX%Hb&qvGBu150sTCGYd z=NwX}Hl!QKJrbs#TS{%2v_$q*Az{rZOukI{xI^n|GsjFx6k@aj|0bG=GzI?;awjEQ z120pAdKI-bp*!^f4TbOy>4f{e4(;UmtB^oP=FKn_G# zx#3GwzVag(hSRXpsp_wM{d_kOHw_!EZLzY0u>UJGD{esRX?Em zn(EPIfsjVUVPGt+DgqznW`5ZhhQA&^P?fPQk_QLb;F1(Sr{pA<``ug{8-+JfW zSAOJf>3O7DpVhX{s#>{X5 zT9GT@fPfhROoL$>4JdZROTeu0D>%6g<*V35Q+aFujYT%28MlHYy`d|HOsRfHQ4J_7 zG{C$@w0}{}pm7?JV_5G~AM?szcon9!3i&SRNN7iVA0l%8m*hsfYLi2D(2*P!(suxA z;GhHL^#WvNTdn(1NN!*<VCHSaavyQ z&h{kKuP*-eo3FVuJ-bQRt@j++*|K5VcwR@j1O=Gcf}L1@GUC7WXWWV2O0sReHqcVA zpHpZdKZWXG6NR`&-=Gnr-7RzQl>6t^L!|!p|Ezw1kYBqaLsP?XQ5tnPaR}S~0ffN* zRj1R5JFy&oYlxMfl-!>VO}f8b^G?K5OL-`}J$XKJ)qQ#GPWN|fyE_Iw)w+}^nfy6e z2A^{Owsz04Ph+#SLoX=46#%doINmN#MVnW5w8|hNf%8OFAUlbSoMR(1*aq^P;^twHJ@PVG@dlLT&bW$=Xg z+(hy^l2?%|A$bMK%kHn%tsVOeHb0Bxb4c#X!7l(WhqlGwtSN5S`XjX%<;c!rJ=*+r zKm+^eD%L1W@Xv>NI3HHwOjIWfPEWiW&WIAkN!eCx)zKF;IP&2b*xp(rS`f(ihz+=d z(lj9CbX#2rfSke?9fc|VQmqOAWokcwPE65Y${_*akP~i9@~5C`s9LR$3ppL+p0WYN z0Pv<<7?cCl9SxMDbP!|~?FiJrp?(QYjbVtD(bdvw0Dd(h_~HH<4#2Q&GzOwh#EIr; z)v!B2xnakei*Q0hIVz5T;xE{pFdk@*;_m%G4LFNVh|fwZKEkx$i}>=C0K9UK@5iAYxw-4g?QNy>eGpbO|12R07!F=PiHJA2|t`t0ec z$wm_nRFf(9F9REDscvT4@FgXUU`8$?z;LPxxrT zO>Z6<{Q~q;k-1X3<*fYk$m(pee4D^ol7G>Cd-I{1a9(N``Ik`BLrCgKzK%qgA76>5 z%;w>Q`%PpnBl!jrd@+0&$qpnU%RG;*y+|;QiGzWkgGA0bi9C$w{&D05@ud&9*0fr) zH2^$MABO-_o;433LaZ0uS4Pf{>5)(*q({(hI*rLjKZg|wt$`lGMupZ>^9|8zT<&b0l}LGq8Z)6O*0mY#DZ zOERUM$yA#CcJI08oO}1&^OENJPaW<`)zr8J_%zlW7>Yh12!Fsq^~V8&-EbM7i0^hi zDQ5LelVW(Y?&tR{rTb?}_j?}}v?i^2O3<2hWzrVD55}vtH85T?DTTM{TPKC^{d!PS zwR>I?!`t-6T$k1|CyIjJlzTvHofEY-UDn$5o!*b94y{Yu0Qa3*x3&@PyR=QEvDhbsHeX9ie-zg$uaK=)}q#mP#?B}jgaSPk!uHTA!$h`>n0feL~6?aIxwYX7IFwD-lzy0)>)T4&% zgzv;Qq*KcUC!|G=8mT1LXxTM=ATtz_t8{L|mF~7TFP9zB4I?_Fr`gM%ZIL$QZAVy( zun}P$f(>CS!u<##gzX5G9rGt{;`J~w0GENMsDO!sHlMx0PS&%v-rmNVjmM3RU(Fag ziNv+>;?B^1Y{b7(46^TdmrG*@*_^+5*ghwW%hZu`YLY3wE@(DWeqFdEW3*8v=Q3?n zIqII1#_iOT^O}H1YhdIs9h#64$RSPE>~V42X$m=?>6{byL-gzuK-Q6~<@y|#B-Hhc zhx*yZKz(mubWntO;1F^_Q^dKaK&DTS1zI;cEehj`D2y#fdCkc+D5g^05y-Gpfi*VQ zp~u;Lpvh_3bwW~v{VMRenlX33KG2tn4eM0Ru!}y`zsz!uM3Z26!p$0Hfm-!ZOQ!J= zz33Ycf=R+sG97jglGMmRW^hm^VS6T?q}wc8?AVsD5+Bj@SW45w-u(v;@9o_adFZhN zj~ymGAV_wxleLWlcvhBc?}?b6rtwrVTPlC>y}hCT5S*h6GsldaR$(Z9UmB*+QwV{1vL%a5P`@Wv`gFWqiVF8>22$Fzu$pbef9kGTo*;sf>F~FL{HybDlXN=hbX;;&_cIP`FkIW)1b>U6W_! zao<=Atu=+oisSlo0nq20lgh5lxpToebXt4PVG0l?_M%(&3t#D9^j_!aW$<1PcyAEc z6wc-_J>a{ryeRZ@KOgs;^b{pD;63R%E*^)OlK(qLe_YZKC~I)CovT|#g$=BJq=k6l zy&>=LO!#?t=H6r}CATp$_p;xv>Qt^z-@0*i;o>W-scDy#*~q@TT4jG-T_@Ht*P1;? zc~V5j=Lf`(%Bszz!Bh3}DG(>jTef8CIQar{;#yXqd`64w5j-HW8v*kaxLiVyl2C1N zgZo%^%~%#sJX{ggbPAbT*T_>S6-HR<4rDLTK7h+O1RzZ);Oa7Wcu8#h=Q6ubTn4VM zOi2Hd?0%fPElWO0k{s~t5FP)JoIWuK)us`++C(KgA>DR->A>>`ibNfp=A`nZQxr@a88&SiNR2CS zm*DQA3h;q9(rl5HGg4QHSfsf@99E#h=3ft!xI zIUOiKhgs??(LLb!otPvaoXnmoXjW7!AObqr4ElPw;u+d2vwE?3!ar znHKiI60k^(Sp!XvGxIaBid%)xt*bMIcP8>PvuYuCK7Z|c{_UG8I9dMcg~II3{M@XX z|G@;zEWG+cVRi~93$LBZzjU>5DSLh4%#DTf7xGuGb}eC27ba09FwTQ4ih#0(ix(Eo zU&)`H&Yyj&+s#)#tYgz-^zZ?0WgwMG^ej*_>FTYlz3;kD`fnODl1n>Kw&05P!Qb(fD+zjakm>J5`VGs-XitiJLXL1x_7$Jc$f{;Wwf{;Q; z16axsiE9y(I$nB;r4m*Rr=?Ry94n*cEZRWBhzHXWl;!4%!PD>IGO&&ED$vQ40vj90 zWnwYRA;@{KH$rGD5Zp2!c{yjJEQ=(@4yeuIkFFh6*NNh<+4+0=ch_MHqt%*qkc4>~ zLYvHb@b4g}`_14Y-4vm6*v)`hX9i9GAh!I0j8A02mcA_apI$Ol<%7IvtDVWrNP4t8 zsYMBij)r>F!sS!>+}mp5()IlJZh#B^Hy5UvbVBb@JBwX_aU`17yVT-wfSrskmH!L? z|AOX`KG3D|ULVW|TzJMYpz5Sc9g2?Wk#q{W6=1Vg#0?r^TFZOx$`P^l z)-63juyB5QI{)&;!sQvY@YcD)+4&wdgN3zlcD68;Qz3v0?_YoxxY@Gvtq}tz$sRV= zy4Ay7qF6!L2d&NGY4)4eCb5J4t#xd91Nd_+lfXKHyIeeWmTGHo@ShAv*x9!2;!*a? zwyjxIWI2k(p3J}?Y#T_OAm?x?hVVQBcQo#N-0#o>!#-}D@~ehj<;Tjal4+E_qp)DN z3@J}A4q&=>xY*Lb0|cxK55Z;p34jXF6^K-%61A0TfeaLh+BpqU zdZ`xZ|4|E|Ip(#X$jSLAv6KcrSR*R6 z09qHQ1+|O%%B6r%Kg>92c5xvvjArlSSICiBEB(KDAJcv@fiI1m(oYh#)1U+*CR=!J zzVMAJ=!^UJ6f-sE_)09Oo!pgssw&W}tCtGjzh1a>LCt@2zHsI(Sg9ge9XV{_`*FFLrg8a`QpO4S+(;7sHqaqR${m!za$X3j=KxrCFaZKDx98C z^OtAyY&QSS^un|Ace53wpDg?e;Xz)t`eOFNl@~6YUzmh#=TCnfvKmzIspT?`#gDxZC=-#Jq@##L{sEzs0shY zDB7)TP)o+pC|ZUalUtBmSw(LAhk)!xh$3u8ScmWk0&m^1*Ctqfc>=7m%Sr+MPU6Mc zQnrtzv`j+Z2?i336Rc_dldUeVNAY5DRwTd8=XBJ&ZEjJqL1(TiO-hH-=H2^m63}>7 diff --git a/src/ai/main.py b/src/ai/main.py index d281d74..68eb72c 100644 --- a/src/ai/main.py +++ b/src/ai/main.py @@ -54,7 +54,6 @@ 'face_absence': None, 'head_turn': None, 'hand_gesture': None, - 'eye_movement': None, 'repeated_gaze': None, 'object': None }) @@ -66,8 +65,7 @@ 'face_absence_repeat': False, 'hand_gesture': False, 'head_turn_long': False, - 'head_turn_repeat': False, - 'eye_movement': False + 'head_turn_repeat': False }) cheating_counts = defaultdict(lambda: { 'look_around': 0, @@ -77,8 +75,18 @@ 'face_absence_repeat': 0, 'hand_gesture': 0, 'head_turn_long': 0, - 'head_turn_repeat': 0, - 'eye_movement': 0 + 'head_turn_repeat': 0 +}) +cheating_settings_cache = defaultdict(lambda: { + 'look_around': False, + 'repeated_gaze': False, + 'object': False, + 'face_absence_long': False, + 'face_absence_repeat': False, + 'hand_gesture': False, + 'head_turn_long': False, + 'head_turn_repeat': False, + 'eye_movement': False }) gaze_history = defaultdict(list) @@ -122,6 +130,40 @@ async def send_message(self, user_id: str, message: dict): executor = ProcessPoolExecutor(max_workers=4) logging.info("ProcessPoolExecutor 초기화 완료") +@app.websocket("/ws/{user_id}/{exam_id}") +async def websocket_endpoint(websocket: WebSocket, user_id: str, exam_id: str): + await manager.connect(user_id, websocket) + + # Fetch cheating settings from backend + cheating_settings = await fetch_cheating_settings(exam_id) + if cheating_settings: + cheating_settings_cache[exam_id] = cheating_settings + logging.info(f"Cached cheating settings for examId {exam_id}: {cheating_settings}") + else: + logging.error(f"Could not fetch cheating settings for examId {exam_id}") + await websocket.close() + return + + try: + while True: + # WebSocket 데이터 처리 로직 + data = await websocket.receive_text() + image_bytes = base64.b64decode(data) + timestamp = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ') + + image_np = np.frombuffer(image_bytes, np.uint8) + image = cv2.imdecode(image_np, cv2.IMREAD_COLOR) + + if image is None: + await websocket.send_json({"error": "이미지 디코딩 실패"}) + continue + + await process_frame(user_id, exam_id, image, timestamp) + except WebSocketDisconnect: + manager.disconnect(user_id) + except Exception as e: + logging.error(f"WebSocket {user_id} 연결 중 오류 발생: {e}") + manager.disconnect(user_id) @app.websocket("/ws/{user_id}") async def websocket_endpoint(websocket: WebSocket, user_id: str): @@ -237,6 +279,18 @@ async def get_cheating_result(user_id: str): 'eye_movement': 0 }) +async def fetch_cheating_settings(exam_id): + async with aiohttp.ClientSession() as session: + try: + async with session.get(f"{BACKEND_API_URL}/exams/{exam_id}/cheatingTypes") as resp: + if resp.status == 200: + return await resp.json() + else: + logging.error(f"Failed to fetch cheating settings for examId {exam_id}. Status: {resp.status}") + except Exception as e: + logging.error(f"Error fetching cheating settings for examId {exam_id}: {e}") + return None + # 부정행위 메시지 전송 async def send_cheating_result_if_changed(user_id: str): global previous_cheating_counts @@ -304,12 +358,6 @@ async def process_frame(user_id, image, frame_timestamp): def process_image(image): """ 이미지를 처리하여 부정행위 탐지에 필요한 정보를 반환 - - Args: - image (numpy.ndarray): 입력 이미지. - - Returns: - tuple: (image_shape, detections, face_present, head_pose, eye_center, gaze_point, hand_landmarks_list) """ image_shape = image.shape image_for_detection = image.copy() @@ -326,7 +374,7 @@ def process_image(image): # 손 랜드마크 추출 hands_results = hands.process(image_rgb) - # 객체 탐지 (YOLO11 사용) + # 객체 탐지 (YOLO 사용) object_results = model(image_for_detection, device=device) logging.debug(f"객체 탐지 완료: {len(object_results)} 결과") @@ -359,49 +407,70 @@ def process_image(image): head_pose = {'pitch': pitch, 'yaw': yaw, 'roll': roll} logging.debug(f"머리 자세: Pitch={pitch}, Yaw={yaw}, Roll={roll}") - # 눈동자 위치 계산 - eye_center = calculate_eye_position(landmarks) - logging.debug(f"눈동자 중심: {eye_center}") - # 시선 위치 추정 gaze_point = get_gaze_position(landmarks) logging.debug(f"시선 위치: {gaze_point}") - return image_shape, detections, face_present, head_pose, eye_center, gaze_point, hand_landmarks_list - - -def update_cheating(user_id, detections, face_present, head_pose, eye_center, gaze_point, hand_landmarks_list, - image_shape): - """ - 감지된 정보를 기반으로 부정행위를 업데이트 - - Args: - user_id (str): 사용자 ID. - detections (list): 감지된 객체의 리스트. - face_present (bool): 얼굴 존재 여부. - head_pose (dict): 머리 자세 (pitch, yaw, roll). - eye_center (tuple): 눈동자의 중심 좌표 (x, y). - gaze_point (tuple): 시선 위치의 좌표 (x, y). - hand_landmarks_list (list): 손 랜드마크 리스트. - image_shape (tuple): 이미지의 형태 (높이, 너비, 채널). - """ - detect_object_presence(user_id, detections, cheating_flags, cheating_counts, cheating_messages, image_shape) - detect_face_absence(user_id, face_present, start_times, cheating_flags, cheating_counts, face_absence_history, - cheating_messages) + return image_shape, detections, face_present, head_pose, gaze_point, hand_landmarks_list +# def update_cheating(user_id, detections, face_present, head_pose, eye_center, gaze_point, hand_landmarks_list, +# image_shape): +# """ +# 감지된 정보를 기반으로 부정행위를 업데이트 +# +# Args: +# user_id (str): 사용자 ID. +# detections (list): 감지된 객체의 리스트. +# face_present (bool): 얼굴 존재 여부. +# head_pose (dict): 머리 자세 (pitch, yaw, roll). +# eye_center (tuple): 눈동자의 중심 좌표 (x, y). +# gaze_point (tuple): 시선 위치의 좌표 (x, y). +# hand_landmarks_list (list): 손 랜드마크 리스트. +# image_shape (tuple): 이미지의 형태 (높이, 너비, 채널). +# """ +# detect_object_presence(user_id, detections, cheating_flags, cheating_counts, cheating_messages, image_shape) +# detect_face_absence(user_id, face_present, start_times, cheating_flags, cheating_counts, face_absence_history, +# cheating_messages) +# if head_pose: +# pitch = head_pose['pitch'] +# yaw = head_pose['yaw'] +# detect_look_around(user_id, pitch, yaw, start_times, cheating_flags, cheating_counts, cheating_messages) +# detect_head_turn(user_id, pitch, yaw, start_times, cheating_flags, cheating_counts, head_turn_history, +# cheating_messages) +# detect_eye_movement(user_id, eye_center, image_shape, start_times, cheating_flags, cheating_counts, +# cheating_messages) +# if gaze_point: +# grid_row = int(gaze_point[1] / (image_shape[0] / GRID_ROWS)) +# grid_col = int(gaze_point[0] / (image_shape[1] / GRID_COLS)) +# grid_position = (grid_row, grid_col) +# detect_repeated_gaze(user_id, grid_position, gaze_history, start_times, cheating_flags, cheating_counts, +# cheating_messages, pitch) +# if hand_landmarks_list: +# detect_hand_gestures(user_id, hand_landmarks_list, start_times, cheating_flags, cheating_counts, +# cheating_messages) +def update_cheating(user_id, exam_id, detections, face_present, head_pose, eye_center, gaze_point, hand_landmarks_list, image_shape): + # Fetch cheating settings for the current exam + cheating_settings = cheating_settings_cache.get(exam_id, {}) + + # 동적으로 탐지 로직 활성화 + if cheating_settings.get('object'): + detect_object_presence(user_id, detections, cheating_flags, cheating_counts, cheating_messages, image_shape) + if cheating_settings.get('face_absence_long') or cheating_settings.get('face_absence_repeat'): + detect_face_absence(user_id, face_present, start_times, cheating_flags, cheating_counts, face_absence_history, + cheating_messages) if head_pose: pitch = head_pose['pitch'] yaw = head_pose['yaw'] - detect_look_around(user_id, pitch, yaw, start_times, cheating_flags, cheating_counts, cheating_messages) - detect_head_turn(user_id, pitch, yaw, start_times, cheating_flags, cheating_counts, head_turn_history, - cheating_messages) - detect_eye_movement(user_id, eye_center, image_shape, start_times, cheating_flags, cheating_counts, - cheating_messages) - if gaze_point: + if cheating_settings.get('look_around'): + detect_look_around(user_id, pitch, yaw, start_times, cheating_flags, cheating_counts, cheating_messages) + if cheating_settings.get('head_turn_long') or cheating_settings.get('head_turn_repeat'): + detect_head_turn(user_id, pitch, yaw, start_times, cheating_flags, cheating_counts, head_turn_history, + cheating_messages) + if gaze_point and cheating_settings.get('repeated_gaze'): grid_row = int(gaze_point[1] / (image_shape[0] / GRID_ROWS)) grid_col = int(gaze_point[0] / (image_shape[1] / GRID_COLS)) grid_position = (grid_row, grid_col) detect_repeated_gaze(user_id, grid_position, gaze_history, start_times, cheating_flags, cheating_counts, cheating_messages, pitch) - if hand_landmarks_list: + if cheating_settings.get('hand_gesture'): detect_hand_gestures(user_id, hand_landmarks_list, start_times, cheating_flags, cheating_counts, cheating_messages) \ No newline at end of file