feat(reolink): detect a motorised lens instead of assuming static cameras have none

pyro_camera_api/pyro_camera_api/api/routes_control.py

@@ -1013,6 +1013,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/camera/adapters/reolink.py

@@ -71,6 +71,51 @@ def __init__(
                 len(self.cam_poses),
                 len(self.cam_azimuths),
             )
+        self._has_motorised_lens: Optional[bool] = None
+
+    def _probe_zoom_support(self) -> Optional[bool]:
+        """Ask the camera whether it reports a zoom position.
+
+        Returns None only when the camera could not be asked, so the caller can
+        tell a transport failure apart from a device that answered. A camera
+        replying with a non-zero Reolink code has answered: it does not serve
+        GetZoomFocus, which settles the question.
+        """
+        try:
+            response = requests.post(
+                self._build_url("GetZoomFocus"),
+                json=[{"cmd": "GetZoomFocus", "action": 0, "param": {"channel": 0}}],
+                verify=False,  # nosec: B501
+            )
+        except Exception as exc:
+            logger.warning("[%s] lens probe could not reach the camera: %s", self.ip_address, exc)
+            return None
+        if response.status_code != 200:
+            logger.warning("[%s] lens probe got HTTP %s", self.ip_address, response.status_code)
+            return None
+        payload = response.json()
+        if payload[0].get("code") != 0:
+            logger.info("[%s] camera does not serve GetZoomFocus: fixed lens", self.ip_address)
+            return False
+        return payload[0]["value"]["ZoomFocus"].get("zoom", {}).get("pos") is not None
+
+    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.
+
+        The answer is cached once the camera gives one, since a lens cannot grow
+        a motor at runtime and zoom commands are frequent. Only an unanswered
+        probe is retried: caching that would strand a camera over one bad
+        request, and re-probing a camera that already said no would cost a
+        request on every command for the rest of the process.
+        """
+        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."""
@@ -232,7 +277,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 = [
                 {
@@ -249,7 +294,7 @@ 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 = [
@@ -310,7 +355,7 @@ def capture_and_score(pos: int) -> float:
                 image.save(f"{folder}/focus_{pos}.jpg")
             return score_local
 
-        if self.cam_type == "static":
+        if not self.has_motorised_lens():
             return 720
 
         if self.focus_position is None:

pyro_camera_api/tests/test_reolink_lens.py

@@ -0,0 +1,112 @@
+# 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
+
+from pyro_camera_api.camera.adapters.reolink import ReolinkCamera
+
+POST = "pyro_camera_api.camera.adapters.reolink.requests.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",  # noqa: S106
+        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_camera_that_rejects_the_command_is_a_settled_answer():
+    """A non-zero Reolink code is the camera answering that it does not serve
+    GetZoomFocus. Re-probing it would cost a request on every command forever."""
+    cam = _camera(cam_type="static")
+    with patch(POST, return_value=_reply(code=-9)) as post:
+        assert cam.has_motorised_lens() is False
+        assert cam.has_motorised_lens() is False
+        assert post.call_count == 1
+
+
+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 PTZ camera as fixed-lens for the rest of the process."""
+    cam = _camera(cam_type="ptz")
+    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="ptz")
+    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