operating-room-monitor-server/scripts/demo_client/fake_rtsp_from_file.py

#!/usr/bin/env python3
"""Publish local video file(s) to RTSP once per file (fake camera) for local dev.

The Operation Room server only opens RTSP URLs (OpenCV); there is no video-upload API.
This script does NOT change the application backend: it runs ffmpeg + a small
RTSP server (MediaMTX); put the printed ``video_rtsp_urls`` into ``OR_SITE_CONFIG_JSON_FILE``.

Requires:
  - ffmpeg in PATH
  - Docker; 默认拉取 ``MEDIAMTX_DOCKER_IMAGE``（DaoCloud 前缀的 bluenviron/mediamtx），
    或本地已拉取的镜像；也可用 PATH 中的 ``mediamtx`` 二进制（高级）。

Single stream (legacy)::
  python3 scripts/demo_client/fake_rtsp_from_file.py /path/to/video.mp4
  python3 scripts/demo_client/fake_rtsp_from_file.py video.mp4 --port 18554 --path demo

Multiple streams (one MediaMTX, one ffmpeg per camera; different RTSP path per stream)::

  python3 scripts/demo_client/fake_rtsp_from_file.py --port 18554 \\
    --stream 'or-cam-01|./a.mp4|demo1' \\
    --stream 'or-cam-02|./b.mp4|demo2'

--stream format: ``CAMERA_ID|FILE|RTSP_PATH`` (use quotes in shell; RTSP path is
the last segment, e.g. ``demo1`` -> ``rtsp://127.0.0.1:<port>/demo1``).
"""

from __future__ import annotations

import argparse
import atexit
import json
import os
import signal
import shutil
import subprocess
import sys
import time
from pathlib import Path

# 默认 DaoCloud 镜像前缀；可设 MEDIAMTX_DOCKER_IMAGE=bluenviron/mediamtx:latest 直连 Docker Hub
MEDIAMTX_IMAGE = os.environ.get(
    "MEDIAMTX_DOCKER_IMAGE",
    "m.daocloud.io/docker.io/bluenviron/mediamtx:latest",
)
CONTAINER_NAME = "orm-fake-rtsp-mediamtx"


def _has_docker() -> bool:
    return shutil.which("docker") is not None


def _has_ffmpeg() -> bool:
    return shutil.which("ffmpeg") is not None


def _stop_mediamtx_container() -> None:
    if not _has_docker():
        return
    try:
        subprocess.run(
            ["docker", "rm", "-f", CONTAINER_NAME],
            capture_output=True,
            check=False,
            timeout=30,
        )
    except (OSError, subprocess.SubprocessError):
        pass


def _start_mediamtx_docker(host_port: int) -> bool:
    _stop_mediamtx_container()
    cmd = [
        "docker",
        "run",
        "-d",
        "--name",
        CONTAINER_NAME,
        "-p",
        f"127.0.0.1:{host_port}:8554",
        MEDIAMTX_IMAGE,
    ]
    print("[fake-rtsp] Starting MediaMTX:", " ".join(cmd), file=sys.stderr)
    try:
        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    except (OSError, subprocess.SubprocessError) as exc:
        print(f"[fake-rtsp] docker run failed: {exc}", file=sys.stderr)
        return False
    if proc.returncode != 0:
        err = (proc.stderr or proc.stdout or "").strip()
        print(f"[fake-rtsp] docker run exit {proc.returncode}: {err}", file=sys.stderr)
        return False
    atexit.register(_stop_mediamtx_container)
    return True


def _parse_stream_arg(spec: str) -> tuple[str, Path, str]:
    parts = spec.split("|", 2)
    if len(parts) != 3:
        raise ValueError(f"Invalid --stream {spec!r}; expected CAM|FILE|RTSP_PATH (three fields separated by |)")
    cam = parts[0].strip()
    fpath = Path(parts[1].strip()).expanduser()
    rpath = parts[2].strip().strip("/")
    if not cam:
        raise ValueError("empty camera id in --stream")
    if not rpath:
        rpath = "demo"
    return cam, fpath, rpath


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Play each video file once to an RTSP URL (dev fake camera; no backend code change).",
    )
    parser.add_argument(
        "video",
        nargs="?",
        type=Path,
        default=None,
        help="(single-stream mode) Path to a video file",
    )
    parser.add_argument(
        "--path",
        default="demo",
        help="(single-stream mode) RTSP path segment (rtsp://host:port/<path>)",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=18554,
        help="Host port mapped to MediaMTX RTSP (container internal 8554). Default: 18554",
    )
    parser.add_argument(
        "--stream",
        action="append",
        default=None,
        help=("Multi-stream mode. Repeat for each camera. Format: CAM|FILE|RTSP_PATH e.g. or-cam-01|./a.mp4|demo1"),
    )
    parser.add_argument(
        "--no-docker",
        action="store_true",
        help="Do not start Docker; run MediaMTX yourself on the host port mapping.",
    )
    args = parser.parse_args()

    if not _has_ffmpeg():
        print("ffmpeg not found in PATH. Install ffmpeg and retry.", file=sys.stderr)
        return 1

    streams: list[tuple[str, Path, str]] = []
    if args.stream:
        for s in args.stream:
            try:
                streams.append(_parse_stream_arg(s))
            except ValueError as exc:
                print(f"[fake-rtsp] {exc}", file=sys.stderr)
                return 1
    elif args.video is not None:
        fpath = args.video.resolve()
        sp = (args.path or "demo").strip().strip("/") or "demo"
        streams = [("or-cam-01", fpath, sp)]
    else:
        parser.error("Provide a video file (single mode) or one or more --stream CAM|FILE|RTSP_PATH")

    for cam, fpath, rpath in streams:
        rp_file = fpath.resolve()
        if not rp_file.is_file():
            print(f"File not found: {rp_file} (camera {cam!r})", file=sys.stderr)
            return 1
        for ch in rpath:
            if ch not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.-":
                print(
                    f"[fake-rtsp] RTSP path segment {rpath!r} for {cam!r} should be "
                    r"[a-zA-Z0-9_.-] only; adjust --path/--stream",
                    file=sys.stderr,
                )
                return 1

    host_port: int = args.port
    if not args.no_docker:
        if not _has_docker():
            print("Docker not found. Use --no-docker and start MediaMTX manually.", file=sys.stderr)
            return 1
        if not _start_mediamtx_docker(host_port):
            return 1
        print("[fake-rtsp] MediaMTX container started. Waiting for RTSP…", file=sys.stderr)
        time.sleep(1.0)
    else:
        print(
            f"[fake-rtsp] --no-docker: ensure an RTSP server is listening for publish on port {host_port}.",
            file=sys.stderr,
        )

    procs: list[subprocess.Popen] = []
    url_map: dict[str, str] = {}

    for cam, fpath, stream_path in streams:
        fp = fpath.resolve()
        dest_url = f"rtsp://127.0.0.1:{host_port}/{stream_path}"
        url_map[cam] = dest_url
        publish_cmd: list[str] = [
            "ffmpeg",
            "-hide_banner",
            "-loglevel",
            "info",
            "-re",
            "-i",
            str(fp),
            "-c",
            "copy",
            "-f",
            "rtsp",
            "-rtsp_transport",
            "tcp",
            dest_url,
        ]
        print("---", file=sys.stderr)
        print(f"Publish {cam} -> {dest_url}", file=sys.stderr)
        print("  " + " ".join(publish_cmd), file=sys.stderr)
        p = subprocess.Popen(publish_cmd)  # noqa: S603
        procs.append(p)

    site_doc = {"video_rtsp_urls": url_map, "voice_or_room_bindings": []}
    print("---", file=sys.stderr)
    print("RTSP mapping (per camera):", file=sys.stderr)
    for k, u in url_map.items():
        print(f"  {k}: {u}", file=sys.stderr)
    print("", file=sys.stderr)
    print(
        "OR site config (merge video_rtsp_urls into OR_SITE_CONFIG_JSON_FILE; add voice_or_room_bindings as needed):",
        file=sys.stderr,
    )
    print(json.dumps(site_doc, ensure_ascii=False, indent=2), file=sys.stderr)
    print("", file=sys.stderr)
    print("If the server runs in Docker on Mac/Win, use host.docker.internal, e.g.:", file=sys.stderr)
    for cam, u in url_map.items():
        h = u.replace("127.0.0.1", "host.docker.internal", 1)
        print(f"  {cam}: {h}", file=sys.stderr)
    print("---", file=sys.stderr)
    print(
        "Fake RTSP running: each file plays once; script exits when ffmpeg ends "
        "(Ctrl+C to stop early; MediaMTX container removed on exit).",
        file=sys.stderr,
    )

    def on_sigint(_sig: int, _frame) -> None:
        for p in procs:
            if p.poll() is None:
                p.terminate()
        _stop_mediamtx_container()
        raise SystemExit(130)

    signal.signal(signal.SIGINT, on_sigint)
    signal.signal(signal.SIGTERM, on_sigint)

    try:
        while True:
            time.sleep(0.5)
            for p in procs:
                if p.poll() is not None:
                    print(
                        f"[fake-rtsp] ffmpeg ended (code {p.returncode}), stopping all.",
                        file=sys.stderr,
                    )
                    raise KeyboardInterrupt
    except KeyboardInterrupt:
        pass
    finally:
        for p in procs:
            if p.poll() is None:
                p.terminate()
                try:
                    p.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    p.kill()
        _stop_mediamtx_container()

    return 0


if __name__ == "__main__":
    raise SystemExit(main())