feat(reolink): detect a motorised lens on static cameras

pyro_camera_api/pyro_camera_api/api/routes_control.py

@@ -1084,6 +1084,14 @@ def zoom_camera(camera_ip: str, level: int):
     if not hasattr(cam, "start_zoom_focus"):
         raise HTTPException(status_code=400, detail="Camera does not support zoom control")
 
+    # hasattr only proves the adapter has the method, not that this camera can
+    # act on it: the Reolink adapter drops the command for a fixed lens and
+    # returns None, so without this the route would answer 200 for a zoom that
+    # never happened.
+    has_motorised_lens = getattr(cam, "has_motorised_lens", None)
+    if has_motorised_lens is not None and not has_motorised_lens():
+        raise HTTPException(status_code=400, detail="Camera does not have a motorised lens")
+
     lock = _acquire_or_409(camera_ip)
 
     try:

pyro_camera_api/pyro_camera_api/api/routes_stream.py

@@ -32,6 +32,22 @@
 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
@@ -85,6 +101,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)
             cfg_stream = STREAMS[camera_ip]
             input_url: str = cfg_stream["input_url"]
             output_url: str = cfg_stream["output_url"]
@@ -155,26 +172,31 @@ def stop_stream(request: Request):
     """
     Stop the currently running live stream, regardless of which camera is streaming.
 
-    After stopping streaming, a best-effort zoom reset is applied to position 0
-    for cameras that support zoom control. Failures during zoom reset are ignored.
+    After stopping streaming, a best-effort zoom reset restores the zoom the
+    camera had when the stream started (0 if unknown), for cameras that support
+    zoom control. Failures during zoom reset are ignored.
 
     Returns:
         JSON confirmation and camera IP whose stream was stopped,
         or a message indicating that no stream was active.
     """
     update_command_time()
     app = request.app
-    stopped_cam = stop_any_running_stream(app)
-    if stopped_cam:
+    # 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=0)
-                logger.info("[%s] Zoom reset to position 0 after stream stop", stopped_cam)
+                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}
-    return {"message": "No active stream was running"}
+    return {"message": f"Stream for {stopped_cam} stopped. Zoom reset if supported.", "camera_ip": stopped_cam}
 
 
 @router.get("/status")

pyro_camera_api/pyro_camera_api/camera/adapters/reolink.py

@@ -52,6 +52,10 @@ def init_poolmanager(self, *args: Any, **kwargs: Any) -> None:
 _session.headers["Connection"] = "close"
 _session.mount("https://", _LegacyTLSAdapter())
 
+# Zoom/focus reads and the zoom restore run under the stream lock in
+# routes_stream, so a camera that stops answering must not hang them forever.
+_ZOOM_FOCUS_TIMEOUT = 5
+
 
 class ReolinkCamera(BaseCamera, PTZMixin, FocusMixin):
     """
@@ -96,6 +100,65 @@ def __init__(
                 len(self.cam_poses),
                 len(self.cam_azimuths),
             )
+        self._has_motorised_lens: Optional[bool] = None
+        self._lens_probe_warned = False
+
+    def _probe_zoom_support(self) -> Optional[bool]:
+        """Ask the camera whether it reports a zoom position.
+
+        Returns None whenever the probe is inconclusive, so the caller can
+        distinguish transport, HTTP, Reolink API and parsing failures from a
+        successful response that establishes whether a zoom position is
+        available. Never raises.
+        """
+        try:
+            response = _session.post(
+                self._build_url("GetZoomFocus"),
+                json=[{"cmd": "GetZoomFocus", "action": 0, "param": {"channel": 0}}],
+                verify=False,  # nosec: B501
+                timeout=_ZOOM_FOCUS_TIMEOUT,
+            )
+            if response.status_code != 200:
+                self._warn_probe_failure("got HTTP %s", response.status_code)
+                return None
+            reply = response.json()[0]
+            if reply.get("code") != 0:
+                self._warn_probe_failure("was rejected by the camera: %s", reply.get("error"))
+                return None
+            return reply["value"]["ZoomFocus"].get("zoom", {}).get("pos") is not None
+        except Exception as exc:
+            self._warn_probe_failure("failed: %s", exc)
+            return None
+
+    def _warn_probe_failure(self, msg: str, *args: Any) -> None:
+        # Failed probes are retried on every zoom/focus command, so only the
+        # first failure per camera is worth a warning.
+        log = logger.debug if self._lens_probe_warned else logger.warning
+        self._lens_probe_warned = True
+        log("[%s] lens probe " + msg, self.ip_address, *args)
+
+    def has_motorised_lens(self) -> bool:
+        """Whether this camera's lens can be driven.
+
+        ``cam_type`` describes how the camera is mounted, not what optics it
+        carries. A "static" camera is one that does not pan or tilt, which says
+        nothing about zoom: Reolink bullets such as the RLC-811A or the P430 sit
+        fixed on their mast and still ship a motorised varifocal lens.
+
+        A definitive answer from a successful response is cached, since a lens
+        cannot grow a motor at runtime and zoom commands are frequent. Failed
+        probes are retried so one transient error cannot strand a capable camera
+        until the service restarts.
+
+        Only static cameras are probed: PTZ cameras always drive their lens, and
+        probing them would let a transient error drop a zoom command that used
+        to be sent unconditionally.
+        """
+        if self.cam_type != "static":
+            return True
+        if self._has_motorised_lens is None:
+            self._has_motorised_lens = self._probe_zoom_support()
+        return self._has_motorised_lens is True
 
     def _build_url(self, command: str) -> str:
         """Constructs a URL for API commands to the camera."""
@@ -257,7 +320,7 @@ def set_auto_focus(self, disable: bool):
         return self._handle_response(response, "Set AutoFocus settings successfully.")
 
     def start_zoom_focus(self, position: int):
-        if self.cam_type != "static":
+        if self.has_motorised_lens():
             url = self._build_url("StartZoomFocus")
             data: Any = [
                 {
@@ -266,15 +329,15 @@ def start_zoom_focus(self, position: int):
                     "param": {"ZoomFocus": {"channel": 0, "pos": position, "op": "ZoomPos"}},
                 }
             ]
-            response = _session.post(url, json=data, verify=False)  # nosec: B501
+            response = _session.post(url, json=data, verify=False, timeout=_ZOOM_FOCUS_TIMEOUT)  # nosec: B501
             return self._handle_response(response, "Started ZoomFocus successfully.")
         return None
 
     def set_manual_focus(self, position: int):
         """
         Set manual focus to a specific position.
         """
-        if self.cam_type != "static":
+        if self.has_motorised_lens():
             self.focus_position = position
             url = self._build_url("StartZoomFocus")
             data: Any = [
@@ -292,7 +355,7 @@ def get_focus_level(self):
         """Retrieve the current manual focus and zoom positions."""
         url = self._build_url("GetZoomFocus")
         data: Any = [{"cmd": "GetZoomFocus", "action": 0, "param": {"channel": 0}}]
-        response = _session.post(url, json=data, verify=False)  # nosec: B501
+        response = _session.post(url, json=data, verify=False, timeout=_ZOOM_FOCUS_TIMEOUT)  # nosec: B501
         result = self._handle_response(response, "Got zoom/focus values")
         if result and result[0]["code"] == 0:
             zoom_focus = result[0]["value"]["ZoomFocus"]

pyro_camera_api/tests/test_reolink_lens.py

@@ -0,0 +1,158 @@
+# Copyright (C) 2022-2026, Pyronear.
+
+# This program is licensed under the Apache License 2.0.
+# See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
+
+
+from unittest.mock import MagicMock, patch
+
+import requests
+
+from pyro_camera_api.camera.adapters.reolink import ReolinkCamera
+
+POST = "pyro_camera_api.camera.adapters.reolink._session.post"
+
+
+def _reply(status=200, code=0, zoom_pos=0):
+    """A GetZoomFocus response as the camera would send it."""
+    resp = MagicMock()
+    resp.status_code = status
+    zoom = {} if zoom_pos is None else {"pos": zoom_pos}
+    resp.json.return_value = [{"code": code, "value": {"ZoomFocus": {"focus": {"pos": 336}, "zoom": zoom}}}]
+    return resp
+
+
+def _camera(cam_type="static"):
+    return ReolinkCamera(
+        camera_id="cam",
+        ip_address="192.168.1.10",
+        username="user",
+        password="pwd",
+        cam_type=cam_type,
+    )
+
+
+def test_static_camera_with_a_varifocal_lens_can_zoom():
+    """A bullet camera does not pan, which says nothing about its optics."""
+    cam = _camera(cam_type="static")
+    with patch(POST, return_value=_reply(zoom_pos=0)):
+        assert cam.has_motorised_lens() is True
+
+
+def test_fixed_lens_camera_reports_no_zoom_position():
+    cam = _camera(cam_type="static")
+    with patch(POST, return_value=_reply(zoom_pos=None)):
+        assert cam.has_motorised_lens() is False
+
+
+def test_a_rejected_probe_is_not_cached():
+    """A non-zero Reolink code is not evidence about the camera's optics."""
+    cam = _camera(cam_type="static")
+    with patch(POST, side_effect=[_reply(code=-9), _reply(zoom_pos=0)]) as post:
+        assert cam.has_motorised_lens() is False
+        assert cam.has_motorised_lens() is True
+        assert post.call_count == 2
+
+
+def test_unreachable_camera_is_treated_as_fixed_lens():
+    """Probing must not raise: a camera that cannot be asked keeps its commands
+    from being sent rather than taking the whole call down."""
+    cam = _camera()
+    with patch(POST, side_effect=OSError("unreachable")):
+        assert cam.has_motorised_lens() is False
+
+
+def test_an_unanswered_probe_is_not_cached():
+    """A transport failure says nothing about the optics. Caching it would
+    strand a varifocal camera as fixed-lens for the rest of the process."""
+    cam = _camera(cam_type="static")
+    with patch(POST, side_effect=OSError("unreachable")):
+        assert cam.has_motorised_lens() is False
+    with patch(POST, return_value=_reply(zoom_pos=0)):
+        assert cam.has_motorised_lens() is True
+
+
+def test_an_http_error_is_not_cached_either():
+    cam = _camera(cam_type="static")
+    with patch(POST, return_value=_reply(status=500)):
+        assert cam.has_motorised_lens() is False
+    with patch(POST, return_value=_reply(zoom_pos=4)):
+        assert cam.has_motorised_lens() is True
+
+
+def test_capability_is_probed_once():
+    """Zoom and focus commands are frequent; the lens cannot grow a motor."""
+    cam = _camera()
+    with patch(POST, return_value=_reply(zoom_pos=4)) as post:
+        cam.has_motorised_lens()
+        cam.has_motorised_lens()
+        assert post.call_count == 1
+
+
+def test_zoom_command_is_sent_to_a_static_varifocal_camera():
+    """The regression this change is about: the command used to be dropped for
+    every static camera, silently returning None with no request made."""
+    cam = _camera(cam_type="static")
+    with (
+        patch(POST, return_value=_reply(zoom_pos=0)) as post,
+        patch.object(ReolinkCamera, "_handle_response", return_value="ok"),
+    ):
+        assert cam.start_zoom_focus(32) == "ok"
+        # one for the probe, one for the command
+        assert post.call_count == 2
+        payload = post.call_args.kwargs["json"][0]
+        assert payload["param"]["ZoomFocus"]["pos"] == 32
+        assert payload["param"]["ZoomFocus"]["op"] == "ZoomPos"
+
+
+def test_zoom_command_is_not_sent_to_a_fixed_lens_camera():
+    cam = _camera(cam_type="static")
+    with patch(POST, return_value=_reply(zoom_pos=None)) as post:
+        assert cam.start_zoom_focus(32) is None
+        # only the probe
+        assert post.call_count == 1
+
+
+def test_ptz_camera_is_never_probed():
+    """PTZ cameras keep sending zoom commands unconditionally, as before."""
+    cam = _camera(cam_type="ptz")
+    with patch(POST, side_effect=OSError("unreachable")) as post:
+        assert cam.has_motorised_lens() is True
+        assert post.call_count == 0
+
+
+def test_a_malformed_reply_does_not_raise():
+    """A 200 with a body we did not expect must stay inconclusive, not raise
+    out of start_zoom_focus and turn the route's 400 into a 500."""
+    not_json = MagicMock(status_code=200)
+    not_json.json.side_effect = ValueError("Expecting value")
+    no_zoom_focus = MagicMock(status_code=200)
+    no_zoom_focus.json.return_value = [{"code": 0, "value": {}}]
+    cam = _camera()
+    for reply in (not_json, no_zoom_focus):
+        with patch(POST, return_value=reply):
+            assert cam.has_motorised_lens() is False
+    with patch(POST, return_value=_reply(zoom_pos=0)):
+        assert cam.has_motorised_lens() is True
+
+
+def test_a_failing_probe_warns_once():
+    """Failed probes are retried on every command, the log must not fill up."""
+    cam = _camera()
+    with (
+        patch(POST, return_value=_reply(code=1)),
+        patch("pyro_camera_api.camera.adapters.reolink.logger") as log,
+    ):
+        for _ in range(3):
+            cam.has_motorised_lens()
+    assert log.warning.call_count == 1
+    assert log.debug.call_count == 2
+
+
+def test_a_hung_camera_times_out_instead_of_blocking():
+    """The probe runs under the stream lock: it must be bounded, and a
+    timeout must stay inconclusive rather than raise."""
+    cam = _camera()
+    with patch(POST, side_effect=requests.Timeout("read timed out")) as post:
+        assert cam.has_motorised_lens() is False
+    assert post.call_args.kwargs["timeout"] > 0