fix(stream): restore the pre-stream zoom on every stream stop

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,35 +26,14 @@
     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
 
 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")

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:

pyro_camera_api/tests/test_stream_zoom.py

@@ -4,43 +4,61 @@
 # See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> 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()