diff --git a/common/camera_config.py b/common/camera_config.py index 928010c..7166eec 100644 --- a/common/camera_config.py +++ b/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", [])] diff --git a/rtsp_service_ws_kadian.py b/rtsp_service_ws_kadian.py index f258b47..88965dd 100644 --- a/rtsp_service_ws_kadian.py +++ b/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)")