kevin/operating-room-monitor-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

将 Demo 录制收敛为三条独立链路，并重做联调台 UI。

Browse Source 
移除 demo_orch 统一编排，改为 recording_demo 与 live/simulated 服务；客户端拆分为静态资源，以模式卡片与 chip 耗材覆盖三链路联调，并同步测试与文档。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Kevin
2026-05-21 16:50:23 +08:00
parent 09885b4184
commit 153c91f8ff
16 changed files with 2030 additions and 1364 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/.env.example
View File

@@ -86,11 +86,12 @@ POSTGRES_PORT=35432
# DEMO_CORS_ENABLED=true
# 跨主机语音页访问 API 时，可先用 * 联调；生产建议改成具体语音页来源，如 http://192.168.1.100:8080
# DEMO_CORS_ORIGINS=*
# 链路 2（模拟实时）与链路 3（离线 batch）需开启；链路 1 真 RTSP 开录不依赖此项
# DEMO_ORCHESTRATOR_ENABLED=false
# DEMO_ORCHESTRATOR_RTSP_PORT=18554
# Docker 内 API 访问宿主机假流时写入站点 JSON 的主机名（默认 host.docker.internal）
# DOCKER_DEMO_ORCHESTRATOR_RTSP_JSON_HOST=host.docker.internal
# 一键联调 / fake_rtsp_from_file 起 MediaMTX 容器用；默认 DaoCloud 前缀，海外可改回 bluenviron/mediamtx:latest
# 链路 2 simulated-start / fake_rtsp_from_file 起 MediaMTX 容器用
# MEDIAMTX_DOCKER_IMAGE=m.daocloud.io/docker.io/bluenviron/mediamtx:latest
# --- 语音确认静态页（clients/voice-confirmation/start.sh）---

61
backend/app/api.py
View File

@@ -33,11 +33,9 @@ from app.schemas import (
VoiceTerminalAssignmentResponse,
build_consumption_summary,
)
from app.services.recording_live import accept_live_recording
from app.services.surgery_pipeline import SurgeryPipeline
from app.services.voice_terminal_hub import (
VoiceTerminalHub,
assign_voice_terminal_after_recording_started,
)
from app.services.voice_terminal_hub import VoiceTerminalHub
from app.surgery_errors import SurgeryPipelineError
router = APIRouter()
@@ -135,19 +133,21 @@ async def health() -> HealthResponse | JSONResponse:
@router.get(
"/internal/demo/orchestrator-status",
"/internal/demo/recording-modes-status",
tags=["demo"],
summary="一键联调接口是否可用",
description="供 demo 页探测：是否启用 orchestrator、RTSP 文件配置等；此路由始终存在，不依赖 DEMO_ORCHESTRATOR_ENABLED。",
summary="Demo 录制模式（链路 2/3）是否可用",
description="供 demo 页探测；始终注册，不依赖 DEMO_ORCHESTRATOR_ENABLED。",
)
async def demo_orchestrator_status() -> dict:
async def recording_modes_status() -> dict:
f = (settings.or_site_config_json_file or "").strip()
enabled = bool(settings.demo_orchestrator_enabled)
return {
"orchestrator_enabled": bool(settings.demo_orchestrator_enabled),
"orchestrate_method": "POST",
"orchestrate_path": "/internal/demo/orchestrate-and-start",
"video_batch_method": "POST",
"video_batch_path": "/internal/demo/video-batch-surgery",
"demo_recording_modes_enabled": enabled,
"orchestrator_enabled": enabled,
"simulated_start_method": "POST",
"simulated_start_path": "/internal/demo/simulated-start",
"offline_batch_method": "POST",
"offline_batch_path": "/internal/demo/offline-batch",
"or_site_config_json_file_set": bool(f),
"or_site_config_json_file": f or None,
"orchestrator_rtsp_port": settings.demo_orchestrator_rtsp_port,
@@ -181,15 +181,20 @@ async def start_surgery(
payload.camera_ids,
payload.candidate_consumables,
)
accepted: SurgeryApiResponse | None = None
async def _start() -> None:
nonlocal accepted
accepted = await accept_live_recording(
pipeline,
voice_hub,
surgery_id=payload.surgery_id,
camera_ids=list(payload.camera_ids),
candidate_consumables=list(payload.candidate_consumables),
message="摄像头录制已开始，手术已启动。",
)
try:
async def _start() -> None:
await pipeline.start_recording(
payload.surgery_id,
payload.camera_ids,
payload.candidate_consumables,
)
await _call_recording_with_retries(
_start,
max_attempts=bp.SURGERY_RECORDING_MAX_ATTEMPTS,
@@ -199,18 +204,8 @@ async def start_surgery(
except SurgeryPipelineError as exc:
_raise_surgery_pipeline_http(exc, payload.surgery_id)
await assign_voice_terminal_after_recording_started(
voice_hub,
surgery_id=payload.surgery_id,
camera_ids=list(payload.camera_ids),
set_voice_terminal_id=pipeline.set_voice_terminal_id,
)
return SurgeryApiResponse(
surgery_id=payload.surgery_id,
status="accepted",
message="摄像头录制已开始，手术已启动。",
)
assert accepted is not None
return accepted
@router.post(

281
backend/app/routers/demo_orch.py → backend/app/routers/recording_demo.py Executable file → Normal file
View File

@@ -1,10 +1,8 @@
"""Dev-only: upload 1–4 videos, start synthetic RTSP, write RTSP URL file, then start surgery."""
"""Demo 录制模式：链路 2 模拟实时、链路 3 离线 batch（需 DEMO_ORCHESTRATOR_ENABLED）。"""
from __future__ import annotations
import json
import shutil
import tempfile
from pathlib import Path
from typing import Annotated
@@ -18,10 +16,14 @@ from app.config import settings
from app.consumable_catalog import normalize_candidate_consumables_raw
from app.dependencies import get_surgery_pipeline, get_voice_terminal_hub
from app.schemas import SurgeryApiResponse, SurgeryStartRequest
from app.or_site_config import merge_video_rtsp_urls_into_file
from app.services.synthetic_rtsp import StreamSpec, SyntheticRtspManager
from app.services.recording_live import accept_live_recording
from app.services.simulated_rtsp_setup import (
prepare_simulated_rtsp_streams,
read_simulated_stream_uploads,
)
from app.services.surgery_pipeline import SurgeryPipeline
from app.baked import pipeline as bp
from app.services.synthetic_rtsp import SyntheticRtspManager
from app.services.video_batch_cleanup import (
purge_batch_artifacts,
purge_expired_visualizations,
@@ -29,16 +31,35 @@ from app.services.video_batch_cleanup import (
stage_visualization_pending,
)
from app.services.video_batch_runner import VideoBatchRunner
from app.services.voice_terminal_hub import (
VoiceTerminalHub,
assign_voice_terminal_after_recording_started,
)
from app.services.voice_terminal_hub import VoiceTerminalHub
from app.surgery_errors import SurgeryPipelineError
router = APIRouter(prefix="/internal/demo", tags=["demo"])
# Bumped when video-batch flow changes; grep this string in logs after restart to confirm new code.
VIDEO_BATCH_FLOW_MARKER = "purge-all+opt-in-vis-v4"
# Grep in logs after restart to confirm new offline-batch code is loaded.
OFFLINE_BATCH_FLOW_MARKER = "offline-batch-v5"
def _require_demo_orchestrator() -> None:
if not settings.demo_orchestrator_enabled:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Demo recording modes disabled (set DEMO_ORCHESTRATOR_ENABLED=true).",
)
def _require_site_config_path() -> Path:
path_raw = (settings.or_site_config_json_file or "").strip()
if not path_raw:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=(
"OR_SITE_CONFIG_JSON_FILE must be set to a writable path "
"(strict site JSON with video_rtsp_urls + voice_or_room_bindings); "
"in Docker, bind-mount a host file to this path."
),
)
return Path(path_raw).expanduser()
def _background_finalize_visualization(
@@ -48,7 +69,7 @@ def _background_finalize_visualization(
try:
runner.finalize_visualization(surgery_id=surgery_id)
except Exception:
logger.exception("video batch background visualization failed surgery_id={}", surgery_id)
logger.exception("offline batch visualization failed surgery_id={}", surgery_id)
finally:
purge_expired_visualizations(
runner.root_dir,
@@ -56,7 +77,7 @@ def _background_finalize_visualization(
)
class VideoBatchSurgeryResponse(BaseModel):
class OfflineBatchResponse(BaseModel):
surgery_id: str
status: str
message: str
@@ -66,44 +87,29 @@ class VideoBatchSurgeryResponse(BaseModel):
doctor_display: str | None = None
def _orchestrate_write_rtsp_host() -> str:
"""Write JSON 里用于 RTSP 的主机名。
一键在本进程起 MediaMTX（端口映射在**本机网络命名空间**的 127.0.0.1）并拉流，OpenCV
必须连 ``rtsp://127.0.0.1:port/...``。若改写成 ``host.docker.internal``，会指到
宿主机上的同端口，通常没有这路流，故 DESCRIBE 返回 404。
`DEMO_ORCHESTRATOR_RTSP_JSON_HOST` 对此路由无效；手填假流+仅改 JSON 的拓扑仍可用该配置。
"""
return "127.0.0.1"
@router.post(
"/video-batch-surgery",
response_model=VideoBatchSurgeryResponse,
summary="非实时精确模式：上传单路 MP4 并跑配置引用包 batch",
"/offline-batch",
response_model=OfflineBatchResponse,
summary="链路 3：非实时精确模式（上传 MP4 + 可选标注视频）",
description=(
"仅当 DEMO_ORCHESTRATOR_ENABLED=true。保存上传视频，调用配置算法子进程包 main.py（默认 algorithm_subprocesses/5.15），"
"解析 TSV 后写入最终结果；可选 include_visualization 生成临时标注视频。"
"仅当 DEMO_ORCHESTRATOR_ENABLED=true。不启动 RTSP 实时会话、不触发语音终端；"
"调用 algorithm_subprocesses/5.15 main.py，解析 TSV 后写入最终结果。"
),
)
async def video_batch_surgery(
async def offline_batch(
background_tasks: BackgroundTasks,
surgery_id: Annotated[str, Form()],
video1: Annotated[UploadFile, File(description="单路完整 MP4")],
candidate_consumables_json: Annotated[str, Form()] = "[]",
include_visualization: Annotated[bool, Form()] = False,
pipeline: SurgeryPipeline = Depends(get_surgery_pipeline),
) -> SurgeryApiResponse:
) -> OfflineBatchResponse:
_require_demo_orchestrator()
if len(surgery_id) != 6 or not surgery_id.isdigit():
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail="surgery_id must be exactly 6 digits",
)
if not settings.demo_orchestrator_enabled:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Demo orchestrator disabled (set DEMO_ORCHESTRATOR_ENABLED=true).",
)
try:
candidates = json.loads(candidate_consumables_json)
except json.JSONDecodeError as exc:
@@ -125,9 +131,9 @@ async def video_batch_surgery(
detail="video1 is empty",
)
logger.info(
"video batch request surgery_id={} flow={} include_visualization={}",
"offline batch request surgery_id={} flow={} include_visualization={}",
surgery_id,
VIDEO_BATCH_FLOW_MARKER,
OFFLINE_BATCH_FLOW_MARKER,
include_visualization,
)
runner = VideoBatchRunner()
@@ -154,18 +160,17 @@ async def video_batch_surgery(
)
)
except (FileNotFoundError, RuntimeError, OSError, ValueError) as exc:
logger.exception("video batch failed surgery_id={}: {}", surgery_id, exc)
logger.exception("offline batch failed surgery_id={}: {}", surgery_id, exc)
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=f"video batch failed: {exc}",
detail=f"offline batch failed: {exc}",
) from exc
await pipeline.save_video_batch_result(surgery_id, result.details)
logger.info(
"video batch result saved to database surgery_id={} rows={} (query GET /client/surgeries/{}/result now)",
"offline batch result saved surgery_id={} rows={}",
surgery_id,
len(result.details),
surgery_id,
)
cache_input = result.output_path.parent.parent / "input" / "input.mp4"
@@ -188,7 +193,7 @@ async def video_batch_surgery(
visualization_url: str | None = None
if include_visualization:
visualization_url = f"/internal/demo/video-batch-surgery/{surgery_id}/visualization"
visualization_url = f"/internal/demo/offline-batch/{surgery_id}/visualization"
doctor = result.doctor
doctor_suffix = ""
if doctor is not None and doctor.display:
@@ -196,7 +201,7 @@ async def video_batch_surgery(
vis_suffix = ""
if include_visualization:
vis_suffix = "；标注视频后台生成中（完成后刷新 visualization URL，24 小时内有效）"
return VideoBatchSurgeryResponse(
return OfflineBatchResponse(
surgery_id=surgery_id,
status="accepted",
message=(
@@ -212,26 +217,22 @@ async def video_batch_surgery(
@router.get(
"/video-batch-surgery/{surgery_id}/visualization",
summary="获取非实时精确模式生成的带标签视频",
"/offline-batch/{surgery_id}/visualization",
summary="链路 3：获取离线 batch 生成的标注视频",
)
async def video_batch_visualization(surgery_id: str) -> FileResponse:
async def offline_batch_visualization(surgery_id: str) -> FileResponse:
_require_demo_orchestrator()
if len(surgery_id) != 6 or not surgery_id.isdigit():
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail="surgery_id must be exactly 6 digits",
)
if not settings.demo_orchestrator_enabled:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Demo orchestrator disabled (set DEMO_ORCHESTRATOR_ENABLED=true).",
)
runner = VideoBatchRunner()
path = runner.latest_visualization_path(surgery_id)
if path is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="video batch visualization not found; run non-realtime batch first.",
detail="offline batch visualization not found; run offline-batch first.",
)
return FileResponse(
path,
@@ -242,16 +243,15 @@ async def video_batch_visualization(surgery_id: str) -> FileResponse:
@router.post(
"/orchestrate-and-start",
"/simulated-start",
response_model=SurgeryApiResponse,
summary="一键联调：上传 1–4 路视频并开录",
summary="链路 2：模拟实时（上传 1–4 路视频并开录 + 语音）",
description=(
"仅当 DEMO_ORCHESTRATOR_ENABLED=true。保存一路或多路视频、启动 MediaMTX+ffmpeg、"
"将 RTSP 映射合并写入 OR_SITE_CONFIG_JSON_FILE 的 video_rtsp_urls，再执行与 /client/surgeries/start 相同的开录逻辑"
"（含按 voice_or_room_bindings 解析并 WebSocket 推送语音终端指派）。"
"仅当 DEMO_ORCHESTRATOR_ENABLED=true。合成假 RTSP 并写入 OR_SITE_CONFIG_JSON_FILE，"
"再执行与 POST /client/surgeries/start 相同的实时开录与语音终端指派。"
),
)
async def orchestrate_and_start(
async def simulated_start(
surgery_id: Annotated[str, Form()],
video1: Annotated[UploadFile, File(description="第 1 路视频（必填，至少一路）")],
video2: Annotated[UploadFile | None, File(description="第 2 路视频（可选）")] = None,
@@ -269,28 +269,14 @@ async def orchestrate_and_start(
pipeline: SurgeryPipeline = Depends(get_surgery_pipeline),
voice_hub: VoiceTerminalHub = Depends(get_voice_terminal_hub),
) -> SurgeryApiResponse:
_require_demo_orchestrator()
json_path = _require_site_config_path()
logger.info(
"demo orchestrate-and-start: surgery_id={} cameras={} rpaths={}",
"simulated-start: surgery_id={} cameras={} rpaths={}",
surgery_id,
(camera_1, camera_2, camera_3, camera_4),
(rtsp_path_1, rtsp_path_2, rtsp_path_3, rtsp_path_4),
)
if not settings.demo_orchestrator_enabled:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Demo orchestrator disabled (set DEMO_ORCHESTRATOR_ENABLED=true).",
)
path_raw = (settings.or_site_config_json_file or "").strip()
if not path_raw:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=(
"OR_SITE_CONFIG_JSON_FILE must be set to a writable path "
"(strict site JSON with video_rtsp_urls + voice_or_room_bindings); "
"in Docker, bind-mount a host file to this path."
),
)
json_path = Path(path_raw).expanduser()
try:
candidates = json.loads(candidate_consumables_json)
@@ -304,56 +290,27 @@ async def orchestrate_and_start(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail="candidate_consumables_json must be a JSON array",
)
candidates = normalize_candidate_consumables_raw(candidates)
default_rtsp = ("demo1", "demo2", "demo3", "demo4")
async def _bytes_and_suffix(u: UploadFile) -> tuple[bytes, str]:
raw = await u.read()
ext = Path(u.filename or "clip.mp4").suffix or ".mp4"
return raw, ext
slot_uploads = (video1, video2, video3, video4)
slot_cameras = (
camera_1.strip(),
camera_2.strip(),
camera_3.strip(),
camera_4.strip(),
uploads = await read_simulated_stream_uploads(
video1=video1,
video2=video2,
video3=video3,
video4=video4,
camera_1=camera_1,
camera_2=camera_2,
camera_3=camera_3,
camera_4=camera_4,
rtsp_path_1=rtsp_path_1,
rtsp_path_2=rtsp_path_2,
rtsp_path_3=rtsp_path_3,
rtsp_path_4=rtsp_path_4,
)
slot_rpaths = (
rtsp_path_1.strip(),
rtsp_path_2.strip(),
rtsp_path_3.strip(),
rtsp_path_4.strip(),
)
gathered: list[tuple[bytes, str, str, str]] = []
for idx, u in enumerate(slot_uploads):
if u is None:
break
raw, ext = await _bytes_and_suffix(u)
if not raw:
break
cam = slot_cameras[idx] or f"or-cam-0{idx + 1}"
rp = slot_rpaths[idx] or default_rtsp[idx]
gathered.append((raw, ext, cam, rp))
if not gathered:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail="至少需要一路非空视频（video1）",
)
if len(gathered) > 4:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail="最多 4 路视频",
)
try:
body = SurgeryStartRequest(
surgery_id=surgery_id,
camera_ids=[g[2] for g in gathered],
camera_ids=[u.camera_id for u in uploads],
candidate_consumables=candidates,
)
except Exception as exc:
@@ -362,73 +319,28 @@ async def orchestrate_and_start(
detail=str(exc),
) from exc
work_root = Path(tempfile.mkdtemp(prefix="orm-orch-"))
try:
def _save_files() -> None:
for i, (raw, ext, _cam, _rp) in enumerate(gathered):
fp = work_root / f"v{i + 1}{ext}"
fp.write_bytes(raw)
await anyio.to_thread.run_sync(_save_files)
except OSError as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"failed to save uploads: {exc}",
) from exc
streams = [
StreamSpec(
camera_id=g[2],
file_path=work_root / f"v{i + 1}{g[1]}",
rtsp_path=g[3],
await prepare_simulated_rtsp_streams(
site_config_json_path=json_path,
uploads=uploads,
)
for i, g in enumerate(gathered)
]
port = int(settings.demo_orchestrator_rtsp_port)
try:
def _start_synth() -> dict[str, str]:
mgr = SyntheticRtspManager.get()
_run, url_map = mgr.start(streams, host_port=port, work_dir=work_root)
return url_map
url_map_host = await anyio.to_thread.run_sync(_start_synth)
except (FileNotFoundError, OSError, ValueError, RuntimeError) as exc:
logger.exception("synthetic RTSP start failed: {}", exc)
except HTTPException:
raise
except Exception as exc:
await anyio.to_thread.run_sync(SyntheticRtspManager.stop_active)
shutil.rmtree(work_root, ignore_errors=True)
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=f"synthetic RTSP failed: {exc}",
detail=f"simulated RTSP setup failed: {exc}",
) from exc
host_for_json = _orchestrate_write_rtsp_host()
try:
def _write() -> None:
merge_video_rtsp_urls_into_file(
json_path,
url_map_host,
replace_host=host_for_json,
)
await anyio.to_thread.run_sync(_write)
except OSError as exc:
await anyio.to_thread.run_sync(SyntheticRtspManager.stop_active)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"failed to write RTSP JSON file: {exc}",
) from exc
await anyio.sleep(0.2)
try:
await pipeline.start_recording(
body.surgery_id,
list(body.camera_ids),
list(body.candidate_consumables),
return await accept_live_recording(
pipeline,
voice_hub,
surgery_id=body.surgery_id,
camera_ids=list(body.camera_ids),
candidate_consumables=list(body.candidate_consumables),
message="假 RTSP 已起；映射已写入；摄像头录制已开始。",
)
except SurgeryPipelineError as exc:
await anyio.to_thread.run_sync(SyntheticRtspManager.stop_active)
@@ -436,16 +348,3 @@ async def orchestrate_and_start(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail={"code": exc.code, "message": exc.message, "surgery_id": body.surgery_id},
) from exc
await assign_voice_terminal_after_recording_started(
voice_hub,
surgery_id=body.surgery_id,
camera_ids=list(body.camera_ids),
set_voice_terminal_id=pipeline.set_voice_terminal_id,
)
return SurgeryApiResponse(
surgery_id=body.surgery_id,
status="accepted",
message="假 RTSP 已起；映射已写入；摄像头录制已开始。",
)

37
backend/app/services/recording_live.py Normal file
View File

@@ -0,0 +1,37 @@
"""共享实时开录：CameraSessionManager + 语音终端指派（链路 1 真 RTSP、链路 2 模拟 RTSP）。"""
from __future__ import annotations
from app.schemas import SurgeryApiResponse
from app.services.surgery_pipeline import SurgeryPipeline
from app.services.voice_terminal_hub import (
VoiceTerminalHub,
assign_voice_terminal_after_recording_started,
)
async def accept_live_recording(
pipeline: SurgeryPipeline,
voice_hub: VoiceTerminalHub,
*,
surgery_id: str,
camera_ids: list[str],
candidate_consumables: list[str],
message: str,
) -> SurgeryApiResponse:
await pipeline.start_recording(
surgery_id,
list(camera_ids),
list(candidate_consumables),
)
await assign_voice_terminal_after_recording_started(
voice_hub,
surgery_id=surgery_id,
camera_ids=list(camera_ids),
set_voice_terminal_id=pipeline.set_voice_terminal_id,
)
return SurgeryApiResponse(
surgery_id=surgery_id,
status="accepted",
message=message,
)

160
backend/app/services/simulated_rtsp_setup.py Normal file
View File

@@ -0,0 +1,160 @@
"""链路 2：上传视频 → 合成假 RTSP → 写入站点 JSON 的 video_rtsp_urls。"""
from __future__ import annotations
import shutil
import tempfile
from dataclasses import dataclass
from pathlib import Path
import anyio
from fastapi import HTTPException, UploadFile, status
from loguru import logger
from app.config import settings
from app.or_site_config import merge_video_rtsp_urls_into_file
from app.services.synthetic_rtsp import StreamSpec, SyntheticRtspManager
def simulated_rtsp_json_host() -> str:
"""本进程内 MediaMTX 监听 127.0.0.1；OpenCV 须连同命名空间内的地址。"""
return "127.0.0.1"
@dataclass(frozen=True)
class SimulatedStreamUpload:
raw: bytes
ext: str
camera_id: str
rtsp_path: str
async def read_simulated_stream_uploads(
*,
video1: UploadFile,
video2: UploadFile | None,
video3: UploadFile | None,
video4: UploadFile | None,
camera_1: str,
camera_2: str,
camera_3: str,
camera_4: str,
rtsp_path_1: str,
rtsp_path_2: str,
rtsp_path_3: str,
rtsp_path_4: str,
) -> list[SimulatedStreamUpload]:
default_rtsp = ("demo1", "demo2", "demo3", "demo4")
slot_uploads = (video1, video2, video3, video4)
slot_cameras = (
camera_1.strip(),
camera_2.strip(),
camera_3.strip(),
camera_4.strip(),
)
slot_rpaths = (
rtsp_path_1.strip(),
rtsp_path_2.strip(),
rtsp_path_3.strip(),
rtsp_path_4.strip(),
)
async def _bytes_and_suffix(u: UploadFile) -> tuple[bytes, str]:
raw = await u.read()
ext = Path(u.filename or "clip.mp4").suffix or ".mp4"
return raw, ext
gathered: list[SimulatedStreamUpload] = []
for idx, u in enumerate(slot_uploads):
if u is None:
break
raw, ext = await _bytes_and_suffix(u)
if not raw:
break
cam = slot_cameras[idx] or f"or-cam-0{idx + 1}"
rp = slot_rpaths[idx] or default_rtsp[idx]
gathered.append(
SimulatedStreamUpload(raw=raw, ext=ext, camera_id=cam, rtsp_path=rp)
)
if not gathered:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail="至少需要一路非空视频（video1）",
)
if len(gathered) > 4:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail="最多 4 路视频",
)
return gathered
async def prepare_simulated_rtsp_streams(
*,
site_config_json_path: Path,
uploads: list[SimulatedStreamUpload],
) -> list[str]:
"""保存上传、启动假 RTSP、合并写入站点 JSON；返回 camera_id 列表。"""
work_root = Path(tempfile.mkdtemp(prefix="orm-sim-"))
try:
def _save_files() -> None:
for i, item in enumerate(uploads):
fp = work_root / f"v{i + 1}{item.ext}"
fp.write_bytes(item.raw)
await anyio.to_thread.run_sync(_save_files)
except OSError as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"failed to save uploads: {exc}",
) from exc
streams = [
StreamSpec(
camera_id=item.camera_id,
file_path=work_root / f"v{i + 1}{item.ext}",
rtsp_path=item.rtsp_path,
)
for i, item in enumerate(uploads)
]
port = int(settings.demo_orchestrator_rtsp_port)
try:
def _start_synth() -> dict[str, str]:
mgr = SyntheticRtspManager.get()
_run, url_map = mgr.start(streams, host_port=port, work_dir=work_root)
return url_map
url_map_host = await anyio.to_thread.run_sync(_start_synth)
except (FileNotFoundError, OSError, ValueError, RuntimeError) as exc:
logger.exception("synthetic RTSP start failed: {}", exc)
await anyio.to_thread.run_sync(SyntheticRtspManager.stop_active)
shutil.rmtree(work_root, ignore_errors=True)
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=f"synthetic RTSP failed: {exc}",
) from exc
host_for_json = simulated_rtsp_json_host()
try:
def _write() -> None:
merge_video_rtsp_urls_into_file(
site_config_json_path,
url_map_host,
replace_host=host_for_json,
)
await anyio.to_thread.run_sync(_write)
except OSError as exc:
await anyio.to_thread.run_sync(SyntheticRtspManager.stop_active)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"failed to write RTSP JSON file: {exc}",
) from exc
await anyio.sleep(0.2)
return [item.camera_id for item in uploads]

16
backend/main.py
View File

@@ -89,19 +89,19 @@ def create_app() -> FastAPI:
logger.info("CORS enabled for demo client; origins={}", origins)
application.include_router(api_router)
if settings.demo_orchestrator_enabled:
from app.routers import demo_orch
from app.routers import recording_demo
application.include_router(demo_orch.router)
application.include_router(recording_demo.router)
logger.info(
"Demo orchestrator enabled: POST /internal/demo/orchestrate-and-start; "
"video-batch flow={}",
demo_orch.VIDEO_BATCH_FLOW_MARKER,
"Demo recording modes enabled: POST /internal/demo/simulated-start; "
"POST /internal/demo/offline-batch; flow={}",
recording_demo.OFFLINE_BATCH_FLOW_MARKER,
)
else:
logger.info(
"Demo orchestrator disabled (DEMO_ORCHESTRATOR_ENABLED=false): "
"GET /internal/demo/orchestrator-status for status; "
"POST /internal/demo/orchestrate-and-start is not registered",
"Demo recording modes disabled (DEMO_ORCHESTRATOR_ENABLED=false): "
"GET /internal/demo/recording-modes-status for status; "
"simulated-start and offline-batch are not registered",
)
return application

1
backend/tests/test_app_integration.py
View File

@@ -162,6 +162,7 @@ def integration_client(
return None
monkeypatch.setattr(main_module, "check_database", _noop)
monkeypatch.setattr("app.api.check_database", _noop)
class _FakeEngine:
async def dispose(self) -> None:

19
backend/tests/test_fastapi_algorithm_subprocess.py
View File

@@ -1,7 +1,7 @@
"""FastAPI → 算法子进程调用链单元测试。
覆盖两条生产路径：
1. ``POST /internal/demo/video-batch-surgery`` → ``VideoBatchRunner`` → ``subprocess.run``（reference bundle ``main.py``）
1. ``POST /internal/demo/offline-batch`` → ``VideoBatchRunner`` → ``subprocess.run``（reference bundle ``main.py``）
2. ``POST /client/surgeries/start`` → ``CameraSessionManager`` → ``asyncio.create_subprocess_exec``（``python -m app.algorithm_runner``）
"""
@@ -22,7 +22,7 @@ from httpx import ASGITransport, AsyncClient
from app.api import router as api_router
from app.config import Settings
from app.dependencies import build_container, get_surgery_pipeline, get_voice_terminal_hub
from app.routers import demo_orch
from app.routers import recording_demo
from app.services.video.session_manager import CameraSessionManager
from app.services.video_batch_runner import VideoBatchRunner, build_reference_command
from tests.test_video_batch_runner import _complete_result_tsv_body, _write_minimal_reference_bundle
@@ -65,7 +65,7 @@ def test_video_batch_endpoint_invokes_reference_bundle_subprocess(
reference_bundle: Path,
sqlite_session_factory,
) -> None:
monkeypatch.setattr(demo_orch.settings, "demo_orchestrator_enabled", True)
monkeypatch.setattr(recording_demo.settings, "demo_orchestrator_enabled", True)
monkeypatch.setattr(
"app.services.video_batch_runner.resolve_reference_bundle_dir",
lambda _override=None: reference_bundle.resolve(),
@@ -75,7 +75,7 @@ def test_video_batch_endpoint_invokes_reference_bundle_subprocess(
lambda _bundle: True,
)
monkeypatch.setattr(
demo_orch,
recording_demo,
"VideoBatchRunner",
lambda: VideoBatchRunner(
bundle_dir=reference_bundle,
@@ -89,14 +89,14 @@ def test_video_batch_endpoint_invokes_reference_bundle_subprocess(
_fake_reference_subprocess_run(captured),
)
container = build_container(demo_orch.settings, session_factory=sqlite_session_factory)
container = build_container(recording_demo.settings, session_factory=sqlite_session_factory)
app = FastAPI()
app.include_router(demo_orch.router)
app.include_router(recording_demo.router)
app.dependency_overrides[get_surgery_pipeline] = lambda: container.surgery_pipeline
client = TestClient(app)
res = client.post(
"/internal/demo/video-batch-surgery",
"/internal/demo/offline-batch",
data={
"surgery_id": "100001",
"candidate_consumables_json": '["耗材1"]',
@@ -198,7 +198,10 @@ async def test_start_surgery_endpoint_spawns_algorithm_runner_subprocess(
async def _noop_voice_assign(*args: Any, **kwargs: Any) -> None:
return None
monkeypatch.setattr("app.api.assign_voice_terminal_after_recording_started", _noop_voice_assign)
monkeypatch.setattr(
"app.services.recording_live.assign_voice_terminal_after_recording_started",
_noop_voice_assign,
)
app = FastAPI()
app.include_router(api_router)

18
backend/tests/test_recording_modes_status.py Normal file
View File

@@ -0,0 +1,18 @@
"""GET /internal/demo/recording-modes-status 契约。"""
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.api import router as api_router
def test_recording_modes_status_paths() -> None:
app = FastAPI()
app.include_router(api_router)
client = TestClient(app)
res = client.get("/internal/demo/recording-modes-status")
assert res.status_code == 200
body = res.json()
assert body["simulated_start_path"] == "/internal/demo/simulated-start"
assert body["offline_batch_path"] == "/internal/demo/offline-batch"
assert "demo_recording_modes_enabled" in body

20
backend/tests/test_video_batch_runner.py
View File

@@ -16,7 +16,7 @@ from app.algorithm_runner import reference_bundle_runtime
from app.api import router as api_router
from app.dependencies import get_surgery_pipeline
from app.domain.consumption import SurgeryConsumptionStored
from app.routers import demo_orch
from app.routers import recording_demo
from app.schemas import SurgeryConsumptionDetail
from app.services.video_batch_runner import (
VideoBatchRunResult,
@@ -558,7 +558,7 @@ def test_demo_video_batch_endpoint_writes_queryable_result(
tmp_path: Path,
monkeypatch,
) -> None:
monkeypatch.setattr(demo_orch.settings, "demo_orchestrator_enabled", True)
monkeypatch.setattr(recording_demo.settings, "demo_orchestrator_enabled", True)
detail = SurgeryConsumptionStored(
item_id="P1",
@@ -629,17 +629,17 @@ def test_demo_video_batch_endpoint_writes_queryable_result(
for r in rows
]
monkeypatch.setattr(demo_orch, "VideoBatchRunner", _FakeRunner)
monkeypatch.setattr(recording_demo, "VideoBatchRunner", _FakeRunner)
pipeline = _FakePipeline()
app = FastAPI()
app.include_router(api_router)
app.include_router(demo_orch.router)
app.include_router(recording_demo.router)
app.dependency_overrides[get_surgery_pipeline] = lambda: pipeline
client = TestClient(app)
res = client.post(
"/internal/demo/video-batch-surgery",
"/internal/demo/offline-batch",
data={"surgery_id": "100001", "candidate_consumables_json": '["耗材1"]'},
files={"video1": ("case.mp4", b"video-bytes", "video/mp4")},
)
@@ -662,7 +662,7 @@ def test_demo_video_batch_endpoint_stages_vis_and_purges_cache_when_requested(
tmp_path: Path,
monkeypatch,
) -> None:
monkeypatch.setattr(demo_orch.settings, "demo_orchestrator_enabled", True)
monkeypatch.setattr(recording_demo.settings, "demo_orchestrator_enabled", True)
detail = SurgeryConsumptionStored(
item_id="P1",
@@ -708,14 +708,14 @@ def test_demo_video_batch_endpoint_stages_vis_and_purges_cache_when_requested(
) -> None:
return None
monkeypatch.setattr(demo_orch, "VideoBatchRunner", _FakeRunner)
monkeypatch.setattr(recording_demo, "VideoBatchRunner", _FakeRunner)
app = FastAPI()
app.include_router(demo_orch.router)
app.include_router(recording_demo.router)
app.dependency_overrides[get_surgery_pipeline] = lambda: _FakePipeline()
client = TestClient(app)
res = client.post(
"/internal/demo/video-batch-surgery",
"/internal/demo/offline-batch",
data={
"surgery_id": "100001",
"candidate_consumables_json": '["耗材1"]',
@@ -725,7 +725,7 @@ def test_demo_video_batch_endpoint_stages_vis_and_purges_cache_when_requested(
)
assert res.status_code == 200, res.text
body = res.json()
assert body["visualization_url"] == "/internal/demo/video-batch-surgery/100001/visualization"
assert body["visualization_url"] == "/internal/demo/offline-batch/100001/visualization"
assert vis_calls == ["100001"]
assert not (root_dir / "cache" / "100001").exists()
pending_input = root_dir / "vis_pending" / "100001" / "input.mp4"