diff --git a/pyro_camera_api/pyro_camera_api/api/routes_stream.py b/pyro_camera_api/pyro_camera_api/api/routes_stream.py index 1494ec02..8a414557 100644 --- a/pyro_camera_api/pyro_camera_api/api/routes_stream.py +++ b/pyro_camera_api/pyro_camera_api/api/routes_stream.py @@ -13,10 +13,11 @@ from fastapi import APIRouter, HTTPException, Request from pyro_camera_api.camera.focus_manager import cancel_focus_and_wait -from pyro_camera_api.camera.registry import CAMERA_REGISTRY, FOCUS_CANCEL_EVENTS +from pyro_camera_api.camera.registry import FOCUS_CANCEL_EVENTS from pyro_camera_api.core.config import RAW_CONFIG, STREAMS from pyro_camera_api.services.anonymizer_rtsp import EncoderWorker, RTSPDecoderWorker from pyro_camera_api.services.stream import ( + STREAM_LOCK, Pipeline, build_ffmpeg_restream_cmd, get_processes, @@ -25,6 +26,7 @@ is_pipeline_running, is_process_running, log_ffmpeg_output, + save_pre_stream_zoom, stop_any_running_stream, ) from pyro_camera_api.utils.time_utils import update_command_time @@ -32,28 +34,6 @@ router = APIRouter() logger = logging.getLogger(__name__) -# Zoom each camera had when its stream started, restored when the stream stops. -# Static varifocal cameras carry a zoom set by hand as their detection framing, -# so resetting them to 0 would lose it. -_pre_stream_zoom: dict[str, int] = {} - - -def _read_zoom(camera_ip: str) -> int: - """Current zoom position of a camera, or 0 when it cannot tell.""" - try: - zoom = (CAMERA_REGISTRY[camera_ip].get_focus_level() or {}).get("zoom") - except Exception as exc: - logger.warning("[%s] Failed to read zoom before stream, %s", camera_ip, exc) - return 0 - return zoom if isinstance(zoom, int) else 0 - - -# Serializes stream startup: without it, two concurrent start_stream calls -# can both pass the idempotent check and spawn two pipelines for the same -# camera. The untracked duplicate keeps streaming invisibly, so stream-aware -# guards (focus, patrol) believe no stream is running. -_START_STREAM_LOCK = threading.Lock() - @router.post("/start_stream/{camera_ip}") def start_stream(camera_ip: str, request: Request): @@ -75,7 +55,7 @@ def start_stream(camera_ip: str, request: Request): if camera_ip not in STREAMS: raise HTTPException(status_code=404, detail=f"No stream config for camera {camera_ip}") - with _START_STREAM_LOCK: + with STREAM_LOCK: app = request.app workers = get_workers(app) procs = get_processes(app) @@ -101,7 +81,7 @@ def start_stream(camera_ip: str, request: Request): # startup failed), otherwise autofocus would be blocked forever. try: stopped_cam = stop_any_running_stream(app) - _pre_stream_zoom[camera_ip] = _read_zoom(camera_ip) + save_pre_stream_zoom(camera_ip) cfg_stream = STREAMS[camera_ip] input_url: str = cfg_stream["input_url"] output_url: str = cfg_stream["output_url"] @@ -181,22 +161,10 @@ def stop_stream(request: Request): or a message indicating that no stream was active. """ update_command_time() - app = request.app - # Same lock as startup: a concurrent start_stream for this camera must not - # have its freshly saved zoom popped by this stop. - with _START_STREAM_LOCK: - stopped_cam = stop_any_running_stream(app) - if not stopped_cam: - return {"message": "No active stream was running"} - cam = CAMERA_REGISTRY.get(stopped_cam) - if cam: - zoom = _pre_stream_zoom.pop(stopped_cam, 0) - try: - cam.start_zoom_focus(position=zoom) - logger.info("[%s] Zoom reset to position %s after stream stop", stopped_cam, zoom) - except Exception as exc: - logger.warning("[%s] Failed to reset zoom, %s", stopped_cam, exc) - return {"message": f"Stream for {stopped_cam} stopped. Zoom reset if supported.", "camera_ip": stopped_cam} + stopped_cam = stop_any_running_stream(request.app) + if stopped_cam: + return {"message": f"Stream for {stopped_cam} stopped. Zoom reset if supported.", "camera_ip": stopped_cam} + return {"message": "No active stream was running"} @router.get("/status") diff --git a/pyro_camera_api/pyro_camera_api/services/stream.py b/pyro_camera_api/pyro_camera_api/services/stream.py index b31284ba..d49ac8b3 100644 --- a/pyro_camera_api/pyro_camera_api/services/stream.py +++ b/pyro_camera_api/pyro_camera_api/services/stream.py @@ -12,6 +12,7 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Optional, cast +from pyro_camera_api.camera.registry import CAMERA_REGISTRY from pyro_camera_api.core.config import FFMPEG_PARAMS from pyro_camera_api.services.anonymizer_rtsp import ( AnonymizerWorker, @@ -39,6 +40,42 @@ class Pipeline: # Optional global app reference for the idle stopper _APP: Optional["FastAPI"] = None +# Serializes stream startup and every stop path. Without it, two concurrent +# start_stream calls can both pass the idempotent check and spawn two +# pipelines for the same camera; the untracked duplicate keeps streaming +# invisibly, so stream-aware guards (focus, patrol) believe no stream is +# running. Reentrant because start_stream stops the previous stream while +# holding it. +STREAM_LOCK = threading.RLock() + +# Zoom each camera had when its stream started, restored when the stream stops. +# Static varifocal cameras carry a zoom set by hand as their detection framing, +# so resetting them to 0 would lose it. +_pre_stream_zoom: dict[str, int] = {} + + +def save_pre_stream_zoom(camera_id: str) -> None: + """Remember the camera's current zoom, or 0 when it cannot tell.""" + try: + zoom = (CAMERA_REGISTRY[camera_id].get_focus_level() or {}).get("zoom") + except Exception as exc: + logger.warning("[%s] Failed to read zoom before stream, %s", camera_id, exc) + zoom = None + _pre_stream_zoom[camera_id] = zoom if isinstance(zoom, int) else 0 + + +def _restore_zoom(camera_id: str) -> None: + """Best-effort return to the pre-stream zoom; failures are logged only.""" + zoom = _pre_stream_zoom.pop(camera_id, 0) + cam = CAMERA_REGISTRY.get(camera_id) + if cam is None or not hasattr(cam, "start_zoom_focus"): + return + try: + cam.start_zoom_focus(position=zoom) + logger.info("[%s] Zoom reset to position %s after stream stop", camera_id, zoom) + except Exception as exc: + logger.warning("[%s] Failed to reset zoom, %s", camera_id, exc) + def set_app_for_stream(app: "FastAPI") -> None: """ @@ -179,11 +216,20 @@ def get(key: str, default): def stop_any_running_stream(app: Optional["FastAPI"]) -> Optional[str]: """ Stop one active stream, whether decoder plus encoder pipeline - or a plain ffmpeg restream process. + or a plain ffmpeg restream process, then restore the zoom the camera had + when its stream started. Priority is to stop pipelines first, then processes. Returns the camera_id that was stopped or None if nothing was running. """ + with STREAM_LOCK: + stopped = _stop_first_running_stream(app) + if stopped: + _restore_zoom(stopped) + return stopped + + +def _stop_first_running_stream(app: Optional["FastAPI"]) -> Optional[str]: global _APP if app is None: diff --git a/pyro_camera_api/tests/test_stream_zoom.py b/pyro_camera_api/tests/test_stream_zoom.py index 7eea42aa..83e16051 100644 --- a/pyro_camera_api/tests/test_stream_zoom.py +++ b/pyro_camera_api/tests/test_stream_zoom.py @@ -4,43 +4,61 @@ # See LICENSE or go to for full license details. +import threading from unittest.mock import MagicMock, patch -from pyro_camera_api.api import routes_stream +from pyro_camera_api.services import stream -STOP = "pyro_camera_api.api.routes_stream.stop_any_running_stream" -REGISTRY = "pyro_camera_api.api.routes_stream.CAMERA_REGISTRY" +STOP = "pyro_camera_api.services.stream._stop_first_running_stream" +REGISTRY = "pyro_camera_api.services.stream.CAMERA_REGISTRY" -def test_stop_stream_restores_the_pre_stream_zoom(): - """A static varifocal camera at a hand-set zoom must get it back, not 0.""" +def _camera(zoom): cam = MagicMock() - cam.get_focus_level.return_value = {"focus": 336, "zoom": 7} + if isinstance(zoom, Exception): + cam.get_focus_level.side_effect = zoom + else: + cam.get_focus_level.return_value = {"focus": 336, "zoom": zoom} + return cam + + +def test_every_stop_path_restores_the_pre_stream_zoom(): + """/stop_stream, replacement in start_stream and the idle stopper all go + through stop_any_running_stream, so a hand-set zoom survives each of them.""" + cam = _camera(7) with patch.dict(REGISTRY, {"cam": cam}, clear=True), patch(STOP, return_value="cam"): - routes_stream._pre_stream_zoom["cam"] = routes_stream._read_zoom("cam") - routes_stream.stop_stream(MagicMock()) + stream.save_pre_stream_zoom("cam") + assert stream.stop_any_running_stream(app=None) == "cam" cam.start_zoom_focus.assert_called_once_with(position=7) + assert "cam" not in stream._pre_stream_zoom -def test_stop_stream_falls_back_to_zero(): - """Unknown pre-stream zoom keeps the historical reset to 0.""" - cam = MagicMock() - cam.get_focus_level.side_effect = OSError("unreachable") +def test_unknown_pre_stream_zoom_falls_back_to_zero(): + cam = _camera(OSError("unreachable")) with patch.dict(REGISTRY, {"cam": cam}, clear=True), patch(STOP, return_value="cam"): - routes_stream._pre_stream_zoom["cam"] = routes_stream._read_zoom("cam") - routes_stream.stop_stream(MagicMock()) + stream.save_pre_stream_zoom("cam") + stream.stop_any_running_stream(app=None) cam.start_zoom_focus.assert_called_once_with(position=0) -def test_stop_stream_holds_the_startup_lock(): +def test_nothing_stopped_means_no_zoom_command(): + cam = _camera(7) + with patch.dict(REGISTRY, {"cam": cam}, clear=True), patch(STOP, return_value=None): + assert stream.stop_any_running_stream(app=None) is None + cam.start_zoom_focus.assert_not_called() + + +def test_stop_and_restore_hold_the_stream_lock(): """Otherwise a concurrent start_stream could save a zoom this stop pops.""" held = [] - def stop(_app): - held.append(routes_stream._START_STREAM_LOCK.locked()) - return "cam" + def probe_lock(**_): + t = threading.Thread(target=lambda: held.append(not stream.STREAM_LOCK.acquire(blocking=False))) + t.start() + t.join() - with patch.dict(REGISTRY, {"cam": MagicMock()}, clear=True), patch(STOP, side_effect=stop): - routes_stream.stop_stream(MagicMock()) + cam = MagicMock() + cam.start_zoom_focus.side_effect = probe_lock + with patch.dict(REGISTRY, {"cam": cam}, clear=True), patch(STOP, return_value="cam"): + stream.stop_any_running_stream(app=None) assert held == [True] - assert not routes_stream._START_STREAM_LOCK.locked()