将model的解析移到model文件内

common/camera_config.py

@@ -1,4 +1,11 @@
 from dataclasses import dataclass
+import json
+import base64
+import yaml
+from utils.logger import get_logger
+
+logger = get_logger(__name__)
+
 
 @dataclass
 class CameraConfig:
@@ -6,3 +13,37 @@ class CameraConfig:
     name: str
     index: str
     params: dict = None  # 额外参数字典，可选
+
+
+def parse_cameras_from_json(json_str: str) -> list[CameraConfig]:
+    """从 JSON 字符串解析摄像头配置（支持 base64 编码）"""
+    try:
+        # 尝试 base64 解码
+        try:
+            decoded = base64.b64decode(json_str).decode('utf-8')
+            cameras_data = json.loads(decoded)
+        except Exception:
+            # 如果不是 base64，直接解析 JSON
+            cameras_data = json.loads(json_str)
+
+        return [CameraConfig(
+            id=c["id"],
+            name=c.get("name", f"cam_{c['id']}"),
+            index=c.get("index"),
+            params=c.get("params")
+        ) for c in cameras_data]
+    except Exception as e:
+        logger.error(f"[ERROR] Failed to parse cameras JSON: {e}")
+        return []
+
+
+def parse_cameras_from_yaml(yaml_path: str) -> list[CameraConfig]:
+    """从 YAML 文件解析摄像头配置"""
+    with open(yaml_path, "r", encoding="utf-8") as f:
+        cfg = yaml.safe_load(f)
+    return [CameraConfig(
+        id=c["id"],
+        name=c.get("name", f"cam_{c['id']}"),
+        index=c.get("index"),
+        params=c.get("params")
+    ) for c in cfg.get("cameras", [])]

rtsp_service_ws_kadian.py

@@ -6,19 +6,13 @@ import cv2
 
 import os
 import sys
-import json
 import time
 import argparse
 import threading
 import queue
-import yaml
-import base64
-
-from dataclasses import dataclass
-
 
 from biz.checkpoint.checkpoint_biz import FrameProcessorWorker
-from common.camera_config import CameraConfig
+from common.camera_config import CameraConfig, parse_cameras_from_json, parse_cameras_from_yaml
 from test_cam import get_camera_preview_url
 from utils.web_socket_sender import WebSocketSender
 
@@ -206,40 +200,6 @@ class RTSPService:
         logger.info("[INFO] Service stopped")
 
 
-def parse_cameras_from_json(json_str: str) -> list[CameraConfig]:
-    """从 JSON 字符串解析摄像头配置（支持 base64 编码）"""
-    try:
-        # 尝试 base64 解码
-        try:
-            decoded = base64.b64decode(json_str).decode('utf-8')
-            cameras_data = json.loads(decoded)
-        except Exception:
-            # 如果不是 base64，直接解析 JSON
-            cameras_data = json.loads(json_str)
-
-        return [CameraConfig(
-            id=c["id"],
-            name=c.get("name", f"cam_{c['id']}"),
-            index=c.get("index"),
-            params=c.get("params")
-        ) for c in cameras_data]
-    except Exception as e:
-        logger.error(f"[ERROR] Failed to parse cameras JSON: {e}")
-        return []
-
-
-def parse_cameras_from_yaml(yaml_path: str) -> list[CameraConfig]:
-    """从 YAML 文件解析摄像头配置"""
-    with open(yaml_path, "r", encoding="utf-8") as f:
-        cfg = yaml.safe_load(f)
-    return [CameraConfig(
-        id=c["id"],
-        name=c.get("name", f"cam_{c['id']}"),
-        index=c.get("index"),
-        params=c.get("params")
-    ) for c in cfg.get("cameras", [])]
-
-
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="RTSP Service for Detection")
     parser.add_argument("--cameras", type=str, help="Cameras config in JSON format (or base64 encoded JSON)")