diff --git a/pyro_camera_api/pyro_camera_api/api/routes_control.py b/pyro_camera_api/pyro_camera_api/api/routes_control.py index 5f2e8165..4d188225 100644 --- a/pyro_camera_api/pyro_camera_api/api/routes_control.py +++ b/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: diff --git a/pyro_camera_api/pyro_camera_api/camera/adapters/reolink.py b/pyro_camera_api/pyro_camera_api/camera/adapters/reolink.py index b5fdb303..94040bf2 100644 --- a/pyro_camera_api/pyro_camera_api/camera/adapters/reolink.py +++ b/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: diff --git a/pyro_camera_api/tests/test_reolink_lens.py b/pyro_camera_api/tests/test_reolink_lens.py new file mode 100644 index 00000000..26da6ab2 --- /dev/null +++ b/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 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