Merge branch 'master' of ssh://188.166.251.33:2222/colden/cv_state_ana

reproject/analyzer.py

@@ -2,10 +2,8 @@ import cv2
 import mediapipe as mp
 import time
 import numpy as np
-import threading
 import queue
 import multiprocessing as mp_proc
-from multiprocessing import shared_memory
 from collections import deque
 from geometry_utils import (
     calculate_ear,
@@ -65,46 +63,20 @@ class MonitorSystem:
         self.current_emotion = "Neutral"
 
         self.frame_shape = (720, 1280, 3)
-        frame_size = int(np.prod(self.frame_shape))
 
-        # 必须先解除可能存在的残留 (Windows上有时不需要，但保持好习惯)
-        # 最好是随机生成一个名字，确保每次运行都是新的
-        import secrets
-        auth_key = secrets.token_hex(4)
-        shm_unique_name = f"monitor_shm_{auth_key}"
+        # 使用 spawn 避免 fork 复制 OpenCV/MediaPipe/ONNXRuntime 等 C++ 运行时状态。
+        self.mp_ctx = mp_proc.get_context("spawn")
+        self.task_queue = self.mp_ctx.Queue(maxsize=2)
+        self.result_queue = self.mp_ctx.Queue(maxsize=2)
 
-        try:
-            self.shm = shared_memory.SharedMemory(create=True, size=frame_size, name=shm_unique_name)
-        except FileExistsError:
-            # 如果真的点背碰上了，就 connect 这一块
-            self.shm = shared_memory.SharedMemory(name=shm_unique_name)
-        
-        print(f"[Main] 共享内存已创建: {self.shm.name} (Size: {frame_size} bytes)")
-
-        # 本地 numpy 包装器
-        self.shared_frame_array = np.ndarray(
-            self.frame_shape, dtype=np.uint8, buffer=self.shm.buf
-        )
-        # 初始化为全黑，避免噪音
-        self.shared_frame_array.fill(0)
-
-        # 跨进程队列
-        self.task_queue = mp_proc.Queue(maxsize=2) 
-        self.result_queue = mp_proc.Queue(maxsize=2) # 1就够了，最新的覆盖
-
-        # 3. 启动进程
-        # Windows下传参，只传名字字符串是安全的
-        self.worker_proc = mp_proc.Process(
+        self.worker_proc = self.mp_ctx.Process(
             target=background_worker_process,
             args=(
-                self.shm.name,
-                self.frame_shape,
                 self.task_queue,
                 self.result_queue,
                 face_db,
             ),
         )
-        self.worker_proc.daemon = True
         self.worker_proc.start()
 
     def _get_smoothed_value(self, history, current_val):
@@ -125,12 +97,6 @@ class MonitorSystem:
             frame = cv2.resize(frame, (target_w, target_h))
 
         h, w = frame.shape[:2]
-        # 现在肯定匹配了，放心写入
-        try:
-            self.shared_frame_array[:] = frame[:]
-        except Exception:
-            # 极端情况：数组形状不匹配 (比如通道数变了)
-            pass
 
         rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
 
@@ -257,9 +223,7 @@ class MonitorSystem:
                 max(0, sface_loc[3] - spad),
             )
 
-            if self.task_queue.full():
-                self.task_queue.get()
-            self.task_queue.put((sface_loc, 0))
+            self._put_latest_task((0, frame.copy(), sface_loc))
 
             self.last_identity_check_time = now
 
@@ -283,20 +247,14 @@ class MonitorSystem:
             y_max = min(h, y_max + pad_y)
 
             face_loc = (y_min, x_max, y_max, x_min)
+            face_crop = frame[y_min:y_max, x_min:x_max].copy()
 
-            if self.task_queue.full():
-                self.task_queue.get()
-            self.task_queue.put((face_loc, 1))
+            if face_crop.size > 0:
+                self._put_latest_task((1, face_crop, None))
 
             self.last_emotion_check_time = now
 
-        while not self.result_queue.empty():
-            type_, data = self.result_queue.get()
-            if type_ == "identity":
-                self.current_user = data
-            elif type_ == "emotion":
-                self.cached_emotion["label"] = data.get("emotion", "unknown")
-                self.cached_emotion["va"] = data.get("vaVal", (0.0, 0.0))
+        self._drain_results()
         
         analysis_data["identity"] = self.current_user
         analysis_data["emotion_label"] = self.cached_emotion["label"]
@@ -304,6 +262,52 @@ class MonitorSystem:
 
         return analysis_data
 
+    def _put_latest_task(self, task):
+        try:
+            if self.task_queue.full():
+                self.task_queue.get_nowait()
+            self.task_queue.put_nowait(task)
+        except queue.Full:
+            try:
+                self.task_queue.get_nowait()
+                self.task_queue.put_nowait(task)
+            except (queue.Empty, queue.Full):
+                pass
+        except queue.Empty:
+            pass
+
+    def _drain_results(self):
+        while True:
+            try:
+                type_, data = self.result_queue.get_nowait()
+            except queue.Empty:
+                break
+
+            if type_ == "identity":
+                self.current_user = data
+            elif type_ == "emotion":
+                self.cached_emotion["label"] = data.get("emotion", "unknown")
+                self.cached_emotion["va"] = data.get("vaVal", (0.0, 0.0))
+
+    def close(self):
+        try:
+            self._put_latest_task(None)
+        except Exception:
+            pass
+
+        if self.worker_proc.is_alive():
+            self.worker_proc.join(timeout=3)
+
+        if self.worker_proc.is_alive():
+            print("[Worker] 未正常退出，强制结束")
+            self.worker_proc.terminate()
+            self.worker_proc.join(timeout=2)
+
+        try:
+            self.face_mesh.close()
+        except Exception:
+            pass
+
     # def _id_emo_loop(self):
     #     while True:
     #         try:
@@ -337,16 +341,10 @@ class MonitorSystem:
 
 
 def background_worker_process(
-    shm_name,  # 共享内存的名字
-    frame_shape,  # 图像大小 (h, w, 3)
     task_queue,  # 任务队列 (主 -> 从)
     result_queue,  # 结果队列 (从 -> 主)
     face_db_data,  # 把人脸库数据传过去初始化
 ):
-    existing_shm = shared_memory.SharedMemory(name=shm_name)
-    # 创建 numpy 数组视图，无需复制数据
-    shared_frame = np.ndarray(frame_shape, dtype=np.uint8, buffer=existing_shm.buf)
-
     print("[Worker] 正在加载模型...")
     from face_library import FaceLibrary
 
@@ -362,39 +360,38 @@ def background_worker_process(
 
     while True:
         try:
-            # 阻塞等待任务
-            # task_info = (task_type, face_loc)
-            face_loc, task_type = task_queue.get()
+            task = task_queue.get()
+            if task is None:
+                break
 
-            # 注意：这里读取的是共享内存里的图，不需要传图！
-            # 切片操作也是零拷贝
-            # 为了安全，这里 copy 一份出来处理，避免主进程修改
-            # 但实际上如果主进程只写新帧，这里读旧帧也问题不大
-            # 为了绝对安全和解耦，我们假定主进程已经写入了对应的帧
-
-            # (实战技巧：通常我们会用一个信号量或多块共享内存来实现乒乓缓存)
-            # 简化版：我们直接从 shared_frame 读。
-            # 由于主进程跑得快，可能SharedMemory里已经是下一帧了。
-            # 但对于识别身份来说，差一两帧根本没区别！这才是优化的精髓。
-
-            current_frame_view = shared_frame.copy()  # .copy() 如果你怕读写冲突
+            task_type, frame_data, face_loc = task
 
             if task_type == 0:  # Identity
-                # RGB转换
-                rgb = cv2.cvtColor(current_frame_view, cv2.COLOR_BGR2RGB)
+                rgb = cv2.cvtColor(frame_data, cv2.COLOR_BGR2RGB)
                 res = face_lib.identify(rgb, face_location=face_loc)
                 if res:
-                    result_queue.put(("identity", res["info"]))
+                    _put_latest_result(result_queue, ("identity", res["info"]))
 
             elif task_type == 1 and has_emo:  # Emotion
-                # BGR 直接切
-                roi = current_frame_view[
-                    face_loc[0] : face_loc[2], face_loc[3] : face_loc[1]
-                ]
-                if roi.size > 0:
-                    emo_res = analyze_emotion_with_hsemotion(roi)
+                if frame_data.size > 0:
+                    emo_res = analyze_emotion_with_hsemotion(frame_data)
                     if emo_res:
-                        result_queue.put(("emotion", emo_res[0]))
+                        _put_latest_result(result_queue, ("emotion", emo_res[0]))
 
         except Exception as e:
             print(f"[Worker Error] {e}")
+
+
+def _put_latest_result(result_queue, result):
+    try:
+        if result_queue.full():
+            result_queue.get_nowait()
+        result_queue.put_nowait(result)
+    except queue.Full:
+        try:
+            result_queue.get_nowait()
+            result_queue.put_nowait(result)
+        except (queue.Empty, queue.Full):
+            pass
+    except queue.Empty:
+        pass

reproject/main.py

@@ -16,8 +16,8 @@ from webrtc_server import WebRTCServer
 from HeartRateMonitor import HeartRateMonitor
 from hook_mocker import HookMocker
 
-API_URL = "http://10.128.48.48:5000/api/states"
-CAMERA_ID = 5
+API_URL = "http://10.128.48.204:5000/api/states"
+CAMERA_ID = 3
 
 BASIC_FACE_DB = {
     "Zhihang": {"name": "Zhihang Deng", "age": 20, "image-path": "zhihang.png"},
@@ -43,6 +43,21 @@ ana_data_queue = queue.Queue(maxsize=10)
 stop_event = threading.Event()
 
 
+def put_latest(q, item):
+    try:
+        if q.full():
+            q.get_nowait()
+        q.put_nowait(item)
+    except queue.Full:
+        try:
+            q.get_nowait()
+            q.put_nowait(item)
+        except (queue.Empty, queue.Full):
+            pass
+    except queue.Empty:
+        pass
+
+
 def capture_thread():
     """
     采集线程：优化了分发逻辑，对视频流进行降频处理
@@ -98,148 +113,116 @@ def analysis_thread():
     2. 队列满时丢弃旧数据，保证数据实时性。
     """
     monitor = MonitorSystem(BASIC_FACE_DB)
-    print("[Analysis] 分析系统启动...")
-    freq = 0
-    gap = 60
-    status = 0  # 0:open 1:close
-    last_time = time.time()
-    last_freq = 0
-    heart_monitor = HeartRateMonitor()
-    while not stop_event.is_set():
-        try:
-            frame = frame_queue.get(timeout=1)
-        except queue.Empty:
-            continue
-
-        # 核心分析
-        result = monitor.process_frame(frame)
-        result["eye_close_freq"] = 0
-        result["heart_rate"] = 0
-
-        if video_queue.full():
-            video_queue.get_nowait()
-        video_queue.put(result["frame"])
-        if ana_video_queue.full():
-            ana_video_queue.get_nowait()
-        ana_video_queue.put(result["frame"])
-        # print(f"[Analysis] {time.strftime('%Y-%m-%d %H:%M:%S')} - Frame processed")
-
-        payload = {
-            "seat_id": CAMERA_ID,
-            "timestamp": time.time(),
-            "heart_rate": 0,
-            "emo_v": 0,
-            "emo_a": 0,
-            "pose_0": 0,  #pitch
-            "pose_1": 0,  #yaw
-            "pose_2": 0,  #roll
-            "ear": 0,
-            "mar": 0,
-            "label": "",
-            "eye_close_freq": 0,
-            "iris_ratio_x": 0,
-            "iris_ratio_y": 0,
-        }
-
-        front_data = {
-            "seat_id": CAMERA_ID,
-            "timestamp": time.time(),
-            "label": "",
-            "eye_close_freq": 0.0,
-            "iris_ratio_x": 0,
-            "iris_ratio_y": 0,
-            "pose_0": 0,  #pitch
-            "pose_1": 0,  #yaw
-            "pose_2": 0,  #roll
-            "heart_rate": 0,
-        }
-
-        if result["has_face"]: # and result["identity"]:
-            payload.update(
-                {
-                    "ear": result["ear"],
-                    "mar": result["mar"],
-                    "iris_ratio_x": result["iris_ratio"][0],
-                    "iris_ratio_y": result["iris_ratio"][1],
-                    "pose_0": result["pose"][0],
-                    "pose_1": result["pose"][1],
-                    "pose_2": result["pose"][2],
-                    "label": result["emotion_label"],
-                    "emo_v": result["emotion_va"][0],
-                    "emo_a": result["emotion_va"][1],
-                } 
-            )
-            front_data.update(
-                {
-                    "label": result["emotion_label"],
-                    "iris_ratio_x": result["iris_ratio"][0],
-                    "iris_ratio_y": result["iris_ratio"][1],
-                    "pose_0": result["pose"][0],
-                    "pose_1": result["pose"][1],
-                    "pose_2": result["pose"][2],
-                }
-            )
-        # elif result["has_face"]:
-        #     payload.update(
-        #         {
-        #             "name": "Unknown",
-        #             "ear": result["ear"],
-        #             "mar": result["mar"],
-        #             "iris_ratio": result["iris_ratio"],
-        #             "pose": result["pose"],
-        #             "emo_label": result["emotion_label"],
-        #             "emo_va": result["emotion_va"],
-        #         }
-        #     )
-
-        if result["has_face"] and result["ear"] < 0.2:
-            if status == 0:
-                freq += 1
-            status = 1
-        elif result["has_face"] and result["ear"] >= 0.2:
-            if status == 1:
-                freq += 1
-            status = 0
-
-        if time.time() - last_time >= gap:
-            last_freq = freq / 2
-            freq = 0
-            last_time = time.time()
-        result["eye_close_freq"] = last_freq
-        payload["eye_close_freq"] = last_freq
-        front_data["eye_close_freq"] = last_freq
-        bpm = heart_monitor.process_frame(frame, result["landmark"])
-        if bpm != None and result["has_face"]:
-            result["heart_rate"] = bpm
-            payload["heart_rate"] = bpm
-            front_data["heart_rate"] = bpm
-        if data_queue.full():
+    try:
+        print("[Analysis] 分析系统启动...")
+        freq = 0
+        gap = 60
+        status = 0  # 0:open 1:close
+        last_time = time.time()
+        last_freq = 0
+        heart_monitor = HeartRateMonitor()
+        while not stop_event.is_set():
             try:
-                _ = data_queue.get_nowait()
+                frame = frame_queue.get(timeout=1)
             except queue.Empty:
-                pass
-        if front_data_queue.full():
-            try:
-                _ = front_data_queue.get_nowait()
-            except queue.Empty:
-                pass
-        if ana_data_queue.full():
-            try:
-                _ = ana_data_queue.get_nowait()
-            except queue.Empty:
-                pass
+                continue
 
-        data_queue.put(payload)
+            # 核心分析
+            result = monitor.process_frame(frame)
+            result["eye_close_freq"] = 0
+            result["heart_rate"] = 0
 
-        ana_data_queue.put(payload)
+            put_latest(video_queue, result["frame"])
+            put_latest(ana_video_queue, result["frame"])
+            # print(f"[Analysis] {time.strftime('%Y-%m-%d %H:%M:%S')} - Frame processed")
 
-        front_data_queue.put(front_data)
+            payload = {
+                "seat_id": CAMERA_ID,
+                "timestamp": time.time(),
+                "heart_rate": 0,
+                "emo_v": 0,
+                "emo_a": 0,
+                "pose_0": 0,  #pitch
+                "pose_1": 0,  #yaw
+                "pose_2": 0,  #roll
+                "ear": 0,
+                "mar": 0,
+                "label": "",
+                "eye_close_freq": 0,
+                "iris_ratio_x": 0,
+                "iris_ratio_y": 0,
+            }
 
-        show_queue.put((result["frame"], result))
-        # draw_debug_info(frame, result)
-        # cv2.imshow("Monitor Client", frame)
+            front_data = {
+                "seat_id": CAMERA_ID,
+                "timestamp": time.time(),
+                "label": "",
+                "eye_close_freq": 0.0,
+                "iris_ratio_x": 0,
+                "iris_ratio_y": 0,
+                "pose_0": 0,  #pitch
+                "pose_1": 0,  #yaw
+                "pose_2": 0,  #roll
+                "heart_rate": 0,
+            }
 
-    print("[Analysis] 分析线程结束")
+            if result["has_face"]: # and result["identity"]:
+                payload.update(
+                    {
+                        "ear": result["ear"],
+                        "mar": result["mar"],
+                        "iris_ratio_x": result["iris_ratio"][0],
+                        "iris_ratio_y": result["iris_ratio"][1],
+                        "pose_0": result["pose"][0],
+                        "pose_1": result["pose"][1],
+                        "pose_2": result["pose"][2],
+                        "label": result["emotion_label"],
+                        "emo_v": result["emotion_va"][0],
+                        "emo_a": result["emotion_va"][1],
+                    } 
+                )
+                front_data.update(
+                    {
+                        "label": result["emotion_label"],
+                        "iris_ratio_x": result["iris_ratio"][0],
+                        "iris_ratio_y": result["iris_ratio"][1],
+                        "pose_0": result["pose"][0],
+                        "pose_1": result["pose"][1],
+                        "pose_2": result["pose"][2],
+                    }
+                )
+
+            if result["has_face"] and result["ear"] < 0.2:
+                if status == 0:
+                    freq += 1
+                status = 1
+            elif result["has_face"] and result["ear"] >= 0.2:
+                if status == 1:
+                    freq += 1
+                status = 0
+
+            if time.time() - last_time >= gap:
+                last_freq = freq / 2
+                freq = 0
+                last_time = time.time()
+            result["eye_close_freq"] = last_freq
+            payload["eye_close_freq"] = last_freq
+            front_data["eye_close_freq"] = last_freq
+            bpm = heart_monitor.process_frame(frame, result["landmark"])
+            if bpm != None:
+                result["heart_rate"] = bpm
+                payload["heart_rate"] = bpm
+                front_data["heart_rate"] = bpm
+
+            put_latest(data_queue, payload)
+            put_latest(ana_data_queue, payload)
+            put_latest(front_data_queue, front_data)
+            put_latest(show_queue, (result["frame"], result))
+            # draw_debug_info(frame, result)
+            # cv2.imshow("Monitor Client", frame)
+    finally:
+        monitor.close()
+        print("[Analysis] 分析线程结束")
 
 
 def video_stream_thread(server):
@@ -289,6 +272,13 @@ def video_stream_thread(server):
 #         print("✅ [Video] H.264 软编码启动成功！视频将保存为 MP4。")
     # -----------------------------------------------------------
     out1 = cv2.VideoWriter('output1.mp4', fourcc, 30.0, (1280, 720))
+    if not out1.isOpened():
+        print("[Video] avc1 编码器不可用，回退到 mp4v")
+        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+        out1 = cv2.VideoWriter('output1.mp4', fourcc, 30.0, (1280, 720))
+    if not out1.isOpened():
+        print("[Video] 视频写入初始化失败，跳过本地录像")
+        out1 = None
     # out2 = cv2.VideoWriter('output2.mp4', fourcc, 30.0, (1280, 720))
 
     while not stop_event.is_set():
@@ -297,7 +287,8 @@ def video_stream_thread(server):
             server.provide_frame(frame)
             data = front_data_queue.get(timeout=1)
             server.send_data(json.dumps(data))
-            out1.write(frame)
+            if out1 is not None:
+                out1.write(frame)
             # out2.write(frame)
         except queue.Empty:
             continue
@@ -348,7 +339,8 @@ def video_stream_thread(server):
         # except Exception as e:
         #     print(f"[Video] 重连中... {e}")
         #     time.sleep(3)
-    out1.release()
+    if out1 is not None:
+        out1.release()
     # out2.release()
     print("[Video] 线程结束")
 
@@ -552,7 +544,7 @@ def alert_thread(server):
                 if alert_status:
                     print(f"警报: {alert_st}")
                     alert = server.alert(int(time.time()), alert_st, info_level)
-                    alert = HookMocker(alert, "http://10.128.48.48:5000/api/osshook")
+                    alert = HookMocker(alert, "http://10.128.48.204:5000/api/osshook")
                     alert.start(width=1280, height=720, fps=30)
                     print(f"upload buffered frames... {buffered_frame.__len__()}")
                     frames_to_upload = buffered_frame
@@ -675,7 +667,7 @@ def draw_debug_info(frame, result):
 
 
 if __name__ == "__main__":
-    server = WebRTCServer(60, 5, "ws://10.128.48.48:5000")
+    server = WebRTCServer(60, CAMERA_ID, "ws://10.128.48.204:5000")
     server.start()
     t1 = threading.Thread(target=capture_thread, daemon=True)
     t2 = threading.Thread(target=analysis_thread, daemon=True)
@@ -690,23 +682,26 @@ if __name__ == "__main__":
     t5.start()
 
     try:
-        while not stop_event.is_set():
-            try:
-                frame, result = show_queue.get(timeout=1)
-            except queue.Empty:
-                continue
-            # frame = apply_soft_roi(frame)
-            display_frame = frame.copy()
-            draw_debug_info(display_frame, result)
-            cv2.imshow("Monitor Client", display_frame)
-            if cv2.waitKey(1) & 0xFF == ord("q"):
-                stop_event.set()
-            # time.sleep(1)
-        
-        cv2.destroyAllWindows()
-    except KeyboardInterrupt:
-        print("停止程序...")
+        try:
+            while not stop_event.is_set():
+                try:
+                    frame, result = show_queue.get(timeout=1)
+                except queue.Empty:
+                    continue
+                # frame = apply_soft_roi(frame)
+                display_frame = frame.copy()
+                draw_debug_info(display_frame, result)
+                cv2.imshow("Monitor Client", display_frame)
+                if cv2.waitKey(1) & 0xFF == ord("q"):
+                    stop_event.set()
+                # time.sleep(1)
+        except KeyboardInterrupt:
+            print("停止程序...")
+            stop_event.set()
+    finally:
         stop_event.set()
+        cv2.destroyAllWindows()
+        server.stop()
 
     t1.join()
     t2.join()

reproject/webrtc_server.py

@@ -93,6 +93,7 @@ class WebRTCServer:
         self.pcs = set()
         self.fps = fps
         self.frameContainer = [None]
+        self.frame_lock = threading.Lock()
         self.hub = MonitoringHub()
         self.sio = None
 
@@ -161,7 +162,7 @@ class WebRTCServer:
         await pc.setRemoteDescription(
             RTCSessionDescription(offer["sdp"], offer.get("type", "offer"))
         )
-        pc.addTrack(VideoFrameTrack(self.fps, self.frameContainer))
+        pc.addTrack(VideoFrameTrack(self.fps, self.frameContainer, self.frame_lock))
 
         dc = pc.createDataChannel("monitoring")
 
@@ -179,10 +180,12 @@ class WebRTCServer:
         asyncio.run_coroutine_threadsafe(self._websocket_start(), self.background_loop)
 
     def provide_frame(self, frame):
-        self.frameContainer[0] = frame
+        with self.frame_lock:
+            self.frameContainer[0] = frame.copy()
 
     def send_data(self, data):
-        self.hub.send_data(data)
+        if self.background_loop.is_running():
+            self.background_loop.call_soon_threadsafe(self.hub.send_data, data)
 
     def alert(self, timestamp, summary, level):
         payload = {
@@ -207,10 +210,11 @@ class WebRTCServer:
 
 
 class VideoFrameTrack(VideoStreamTrack):
-    def __init__(self, fps, fc):
+    def __init__(self, fps, fc, lock):
         super().__init__()
         self.fps = fps
         self.frameContainer = fc
+        self.frame_lock = lock
 
     async def next_timestamp(self):
         """
@@ -230,11 +234,13 @@ class VideoFrameTrack(VideoStreamTrack):
 
     async def recv(self):
         pts, time_base = await self.next_timestamp()
-        frame = self.frameContainer[0]
+        with self.frame_lock:
+            frame = self.frameContainer[0]
+            if frame is not None:
+                frame = frame.copy()
         if frame is None:
             frame = np.zeros((480, 640, 3), dtype=np.uint8)
         else:
-            frame = self.frameContainer[0]
             frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
         video_frame = av.VideoFrame.from_ndarray(frame, format="rgb24")
         video_frame.pts = pts