operating-room-monitor-server/app/routers/demo_orch.py

"""Dev-only: upload 1–4 videos, start synthetic RTSP, write RTSP URL file, then start surgery."""

from __future__ import annotations

import json
import shutil
import tempfile
from pathlib import Path
from typing import Annotated

import anyio
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile, status
from loguru import logger

from app.config import settings
from app.dependencies import get_surgery_pipeline
from app.schemas import SurgeryApiResponse, SurgeryStartRequest
from app.services.synthetic_rtsp import StreamSpec, SyntheticRtspManager, write_rtsp_url_json_file
from app.services.surgery_pipeline import SurgeryPipeline
from app.surgery_errors import SurgeryPipelineError

router = APIRouter(prefix="/internal/demo", tags=["demo"])


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(
    "/orchestrate-and-start",
    response_model=SurgeryApiResponse,
    summary="一键联调：上传 1–4 路视频并开录",
    description=(
        "仅当 DEMO_ORCHESTRATOR_ENABLED=true。保存一路或多路视频、启动 MediaMTX+ffmpeg、"
        "将 RTSP 映射写入 VIDEO_RTSP_URLS_JSON_FILE，再执行与 /client/surgeries/start 相同的开录逻辑。"
    ),
)
async def orchestrate_and_start(
    surgery_id: Annotated[str, Form()],
    video1: Annotated[UploadFile, File(description="第 1 路视频（必填，至少一路）")],
    video2: Annotated[UploadFile | None, File(description="第 2 路视频（可选）")] = None,
    video3: Annotated[UploadFile | None, File(description="第 3 路视频（可选）")] = None,
    video4: Annotated[UploadFile | None, File(description="第 4 路视频（可选）")] = None,
    camera_1: Annotated[str, Form()] = "or-cam-01",
    camera_2: Annotated[str, Form()] = "or-cam-02",
    camera_3: Annotated[str, Form()] = "or-cam-03",
    camera_4: Annotated[str, Form()] = "or-cam-04",
    rtsp_path_1: Annotated[str, Form()] = "demo1",
    rtsp_path_2: Annotated[str, Form()] = "demo2",
    rtsp_path_3: Annotated[str, Form()] = "demo3",
    rtsp_path_4: Annotated[str, Form()] = "demo4",
    candidate_consumables_json: Annotated[str, Form()] = "[]",
    pipeline: SurgeryPipeline = Depends(get_surgery_pipeline),
) -> SurgeryApiResponse:
    logger.info(
        "demo orchestrate-and-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.video_rtsp_urls_json_file or "").strip()
    if not path_raw:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=(
                "VIDEO_RTSP_URLS_JSON_FILE must be set to a writable path; "
                "in Docker, bind-mount a host file to this path."
            ),
        )
    json_path = Path(path_raw).expanduser()

    try:
        candidates = json.loads(candidate_consumables_json)
    except json.JSONDecodeError as exc:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=f"invalid candidate_consumables_json: {exc}",
        ) from exc
    if not isinstance(candidates, list) or not all(isinstance(x, str) for x in candidates):
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="candidate_consumables_json must be a JSON array of strings",
        )

    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(),
    )
    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],
            candidate_consumables=[str(x) for x in candidates],
        )
    except Exception as exc:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            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],
        )
        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)
        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 = _orchestrate_write_rtsp_host()
    try:

        def _write() -> None:
            write_rtsp_url_json_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),
        )
    except SurgeryPipelineError as exc:
        await anyio.to_thread.run_sync(SyntheticRtspManager.stop_active)
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail={"code": exc.code, "message": exc.message, "surgery_id": body.surgery_id},
        ) from exc

    return SurgeryApiResponse(
        surgery_id=body.surgery_id,
        status="accepted",
        message="假 RTSP 已起；映射已写入；摄像头录制已开始。",
    )