6b3adb4ad8
- 用 OR_SITE_CONFIG_JSON_FILE 统一术间配置（video_rtsp_urls + voice_or_room_bindings） - VoiceTerminalHub：assignment、WS 推送与 HTTP 查询；开录/停录后 notify - 一键联调 orchestrate-and-start 与 /client/surgeries/start 共用指派逻辑，修复 demo 路径不发 WS - 语音桌面端：SIGINT 退出、shutdown 清理、仅 WS 指派、固定 pending 轮询间隔、界面仅保留录音时长 - 新增/调整契约与绑定测试，文档与示例配置同步 Made-with: Cursor
105 lines
3.9 KiB
Python
105 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from loguru import logger
|
|
|
|
from app.config import Settings
|
|
from app.services.video.hikvision_runtime import HikvisionRuntime
|
|
from app.services.video.types import VideoBackendKind
|
|
|
|
|
|
class BackendResolver:
|
|
"""Resolve per-camera backend (RTSP vs Hikvision SDK) and RTSP URL."""
|
|
|
|
def __init__(
|
|
self,
|
|
settings: Settings,
|
|
*,
|
|
hikvision_runtime: HikvisionRuntime | None,
|
|
) -> None:
|
|
self._s = settings
|
|
self._hik = hikvision_runtime
|
|
|
|
def _parse_json_object(self, raw: str) -> dict[str, Any]:
|
|
raw = (raw or "").strip()
|
|
if not raw:
|
|
return {}
|
|
try:
|
|
data = json.loads(raw)
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError(f"Invalid JSON mapping: {exc}") from exc
|
|
if not isinstance(data, dict):
|
|
raise ValueError("JSON mapping must be an object")
|
|
return {str(k): v for k, v in data.items()}
|
|
|
|
def backend_for_camera(self, camera_id: str) -> VideoBackendKind:
|
|
overrides = self._parse_json_object(self._s.video_camera_backend_overrides_json)
|
|
if camera_id in overrides:
|
|
v = str(overrides[camera_id]).lower()
|
|
if v in ("rtsp", "hikvision_sdk", "sdk"):
|
|
return (
|
|
VideoBackendKind.HIKVISION_SDK
|
|
if v in ("hikvision_sdk", "sdk")
|
|
else VideoBackendKind.RTSP
|
|
)
|
|
default = self._s.video_default_backend.strip().lower()
|
|
if default in ("auto", ""):
|
|
if self._hik is not None and self._s.hikvision_sdk_enabled:
|
|
return VideoBackendKind.HIKVISION_SDK
|
|
return VideoBackendKind.RTSP
|
|
if default in ("hikvision_sdk", "sdk"):
|
|
return VideoBackendKind.HIKVISION_SDK
|
|
return VideoBackendKind.RTSP
|
|
|
|
def rtsp_url_for_camera(self, camera_id: str) -> str:
|
|
# Re-read on each use so OR_SITE_CONFIG_JSON_FILE can be hot-updated (e.g. dev orchestrator).
|
|
m = self._s.video_rtsp_url_map()
|
|
if camera_id in m:
|
|
return m[camera_id]
|
|
tpl = (self._s.video_rtsp_url_template or "").strip()
|
|
if tpl:
|
|
try:
|
|
return tpl.format(camera_id=camera_id)
|
|
except KeyError as exc:
|
|
raise ValueError(
|
|
f"video_rtsp_url_template missing placeholder: {exc}"
|
|
) from exc
|
|
raise ValueError(
|
|
f"No RTSP URL for camera_id={camera_id!r}: set OR_SITE_CONFIG_JSON_FILE "
|
|
f"(video_rtsp_urls) or VIDEO_RTSP_URL_TEMPLATE"
|
|
)
|
|
|
|
def rtsp_url_after_hikvision_login(self, camera_id: str) -> str:
|
|
"""RTSP URL used after SDK login (often same as device preview URL)."""
|
|
urls = self._parse_json_object(self._s.hikvision_camera_rtsp_urls_json)
|
|
if camera_id in urls:
|
|
return str(urls[camera_id])
|
|
tpl = (self._s.hikvision_preview_rtsp_template or "").strip()
|
|
if not tpl:
|
|
logger.warning(
|
|
"Hikvision backend without HIKVISION_PREVIEW_RTSP_TEMPLATE / "
|
|
"HIKVISION_CAMERA_RTSP_URLS_JSON — falling back to generic RTSP map"
|
|
)
|
|
return self.rtsp_url_for_camera(camera_id)
|
|
return self._format_hikvision_rtsp(tpl, camera_id)
|
|
|
|
def _format_hikvision_rtsp(self, template: str, camera_id: str) -> str:
|
|
ip = self._s.hikvision_device_ip.strip()
|
|
user = self._s.hikvision_user.strip()
|
|
password = self._s.hikvision_password.strip()
|
|
channel = self._s.hikvision_channel
|
|
try:
|
|
return template.format(
|
|
camera_id=camera_id,
|
|
ip=ip,
|
|
user=user,
|
|
password=password,
|
|
channel=channel,
|
|
)
|
|
except KeyError as exc:
|
|
raise ValueError(
|
|
f"hikvision_preview_rtsp_template missing key: {exc}"
|
|
) from exc
|