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 f6feb295..92322681 100644 --- a/pyro_camera_api/pyro_camera_api/api/routes_control.py +++ b/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: 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 c94ffc2f..1494ec02 100644 --- a/pyro_camera_api/pyro_camera_api/api/routes_stream.py +++ b/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,8 +172,9 @@ 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, @@ -164,17 +182,21 @@ def stop_stream(request: Request): """ 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") 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 edb2ff59..2a35fe4d 100644 --- a/pyro_camera_api/pyro_camera_api/camera/adapters/reolink.py +++ b/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,7 +329,7 @@ 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 @@ -274,7 +337,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 = [ @@ -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"] 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..eb673eab --- /dev/null +++ b/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 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 diff --git a/pyro_camera_api/tests/test_stream_zoom.py b/pyro_camera_api/tests/test_stream_zoom.py new file mode 100644 index 00000000..7eea42aa --- /dev/null +++ b/pyro_camera_api/tests/test_stream_zoom.py @@ -0,0 +1,46 @@ +# 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.api import routes_stream + +STOP = "pyro_camera_api.api.routes_stream.stop_any_running_stream" +REGISTRY = "pyro_camera_api.api.routes_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.""" + cam = MagicMock() + cam.get_focus_level.return_value = {"focus": 336, "zoom": 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()) + cam.start_zoom_focus.assert_called_once_with(position=7) + + +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") + 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()) + cam.start_zoom_focus.assert_called_once_with(position=0) + + +def test_stop_stream_holds_the_startup_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" + + with patch.dict(REGISTRY, {"cam": MagicMock()}, clear=True), patch(STOP, side_effect=stop): + routes_stream.stop_stream(MagicMock()) + assert held == [True] + assert not routes_stream._START_STREAM_LOCK.locked()