Log capture health so a video outage points at a layer

README.md

@@ -19,6 +19,7 @@ that handles robot data.
 - Publish robot laser.
 - Execute callbacks on Custom Action execution.
 - Execute scripts (or any program) in response to Custom Action execution.
+- Stream camera frames from RTSP (or anything OpenCV opens).
 
 ## Quick Start
 
@@ -97,6 +98,64 @@ the code.
    This will clean up various Python and build generated files so that you can
    ensure that you are working in a clean environment.
 
+## Camera streaming
+
+Install the optional **video** extra (see `requirements-video.txt`), which pulls
+in OpenCV:
+
+`pip install inorbit-edge[video]`
+
+Register a camera on a session. Frames are streamed only while the platform asks
+for video -- for example when a user opens a camera view -- and the camera id is
+the topic id the InOrbit camera must be configured with (`"0"` for the first
+one):
+
+```python
+from inorbit_edge.video import OpenCVCamera
+
+session.register_camera(
+    "0", OpenCVCamera("rtsp://user:pass@192.0.2.10:554/stream1", rate=5)
+)
+```
+
+For RTSP, set OpenCV's FFmpeg options **before** the first capture is opened
+(they are read by OpenCV when it opens the stream, so export them or set them in
+`os.environ` at import time):
+
+```bash
+export OPENCV_FFMPEG_CAPTURE_OPTIONS="rtsp_transport;tcp|timeout;3000000"
+```
+
+`rtsp_transport;tcp` because some cameras reject UDP, and `timeout`
+(microseconds) bounds socket reads: without it, a camera that stops answering
+mid-stream is only noticed after OpenCV's 30s watchdog, which delays both the
+reopen and shutdown.
+
+`OpenCVCamera` settings, all optional:
+
+| Setting | Default | Meaning |
+|---------|---------|---------|
+| `rate` | `10` | Frames per second published |
+| `scaling` | `0.3` | Downscale factor applied before JPEG encoding |
+| `quality` | `35` | JPEG quality, 1-100 |
+| `stale_frame_seconds` | `3.0` | Stop serving the buffered frame once it is older than this, so a stream that died shows no video instead of a frozen picture. `None` keeps serving the last frame |
+| `api_preference` | auto | OpenCV backend to open with; URL sources default to `cv2.CAP_FFMPEG` |
+| `REOPEN_BACKOFF_SECONDS` | `0.5s` to `10s` | Class attribute: delay before each attempt to rebuild a capture whose grabs are failing |
+| `HEALTH_LOG_SECONDS` | `60.0` | Class attribute: how often the capture health line below is logged |
+
+Each camera logs one health line per window, which is usually enough to tell
+where video stopped:
+
+```
+Capture health: grabbed=1800 served=60 stale=0 reopens=0 in the last 60s
+```
+
+No line at all means the platform never requested video; `grabbed=0` means the
+stream is unreachable; frames grabbed but not served means nothing is consuming
+them; frames served with nothing visible in the platform points at the MQTT
+side. The same signals are exported as the `video_frames_grabbed`,
+`video_frames_stale` and `video_capture_reopens` counters (see Metrics below).
+
 ## Metrics
 
 The SDK is capable of collecting internal metrics such as number of calls to

inorbit_edge/metrics.py

@@ -214,6 +214,15 @@ def setup_prometheus_meter_provider(
 publish_camera_frame_counter = meter.create_counter(
     "calls_publish_camera_frame", "1", "number of calls to publish camera frames"
 )
+video_frames_grabbed_counter = meter.create_counter(
+    "video_frames_grabbed", "1", "number of video frames read off a camera"
+)
+video_frames_stale_counter = meter.create_counter(
+    "video_frames_stale", "1", "number of video frames withheld for being too old"
+)
+video_capture_reopens_counter = meter.create_counter(
+    "video_capture_reopens", "1", "number of times a video capture was rebuilt"
+)
 publish_pose_counter = meter.create_counter(
     "calls_publish_pose", "1", "number of calls to publish poses"
 )

inorbit_edge/tests/test_video.py

@@ -143,6 +143,35 @@ def test_the_staleness_window_follows_the_publish_rate():
     )
 
 
+def test_health_log_separates_capture_from_publishing(mocker):
+    """The counters say which layer stopped: capture, publishing or neither."""
+    frame = numpy.zeros((16, 16, 3), dtype=numpy.uint8)
+    capture = FakeCapture(grab_ok=True, frame=frame, delay=0.002)
+    camera = OpenCVCamera("rtsp://camera.invalid/stream", rate=1, scaling=0.5)
+    camera.HEALTH_LOG_SECONDS = 60.0  # no window rollover during the test
+    mocker.patch.object(camera, "_open_capture", return_value=capture)
+
+    camera.open()
+    try:
+        assert _wait_until(lambda: camera._grabbed > 0)
+        camera.get_frame_jpg()
+    finally:
+        # Stop the capture thread before ageing the buffered frame, so it cannot
+        # refresh the timestamp underneath the staleness check.
+        camera.close()
+
+    frame_data, _ts = camera._frame
+    camera._frame = (frame_data, time.monotonic() - camera.stale_frame_seconds - 1)
+    camera.get_frame_jpg()
+
+    assert camera._served == 1
+    assert camera._stale == 1
+
+    grabbed = camera._grabbed
+    camera._log_health()
+    assert grabbed > 0 and camera._grabbed == 0, "window was not reset"
+
+
 def test_get_frame_jpg_returns_no_frame_while_the_capture_is_reopening():
     camera = OpenCVCamera("rtsp://camera.invalid/stream", rate=1)
 

inorbit_edge/video.py

@@ -7,7 +7,16 @@
 # * CameraStreamer: Consumes frames from a camera and send them to the platform.
 #
 # Future improvements / TODOs:
-#   * Honor module states camera settings, like rate, size and quality.
+#   * Honor module states camera settings, like rate, size and quality. The
+#     platform already sends them (`modules/set_state` carries a per-camera
+#     rate/quality/is_on in `cameras_config`); with those honored, a frame
+#     should be decoded when one is published rather than on a fixed interval.
+#   * Support non-live sources (video files, VOD). The capture loop drains the
+#     source continuously, which a live stream paces by itself because grab()
+#     blocks on the socket; a file does not block, so the loop races through it
+#     as fast as the decoder manages (3.7k frames/s over ~9 cores for a 1080p30
+#     clip). Those should be grabbed only when a frame is wanted, once callers
+#     can say which kind a URL is.
 #   * Decouple CameraStreamer from image processing and move it to robot.py
 #   * Complete type annotations
 
@@ -16,6 +25,12 @@
 import time
 from abc import ABC, abstractmethod
 
+from inorbit_edge.metrics import (
+    video_capture_reopens_counter,
+    video_frames_grabbed_counter,
+    video_frames_stale_counter,
+)
+
 try:
     import cv2
 except Exception:
@@ -83,6 +98,8 @@ class OpenCVCamera(Camera):
     #: Floor for the derived window: without it a fast publish rate would
     #: withhold frames on ordinary jitter (rate=30 alone gives 0.1s).
     MIN_STALE_FRAME_SECONDS = 1.0
+    #: How often the capture thread logs a one-line capture health summary.
+    HEALTH_LOG_SECONDS = 60.0
 
     def __init__(
         self,
@@ -124,6 +141,12 @@ def __init__(
         # publish is ever used, so convert at twice the publish rate and let the
         # rest of the stream drain through grab() alone.
         self._retrieve_interval = 0.5 / max(rate, 1)
+        # Counters for the health log, reset each window. Written by the capture
+        # thread and by get_frame_jpg(); a lost increment doesn't matter.
+        self._grabbed = 0
+        self._served = 0
+        self._stale = 0
+        self._reopens = 0
 
     def _open_capture(self):
         """Return a new ``cv2.VideoCapture`` for this camera's source."""
@@ -169,9 +192,12 @@ def get_frame_jpg(self):
             # The stream stopped delivering. Publishing this again would show a
             # frozen image as if it were live -- report no frame instead.
             self.logger.debug(f"Withholding a video frame {age:.1f}s old")
+            self._stale += 1
+            video_frames_stale_counter.add(1)
             return None, 0, 0, ts
         height, width = frame[0].shape[:2]
         jpg, w, h = convert_frame(frame[0], width, height, self.scaling, self.quality)
+        self._served += 1
         return jpg, w, h, ts
 
     def _run(self):
@@ -184,6 +210,7 @@ def _run(self):
         """
         failures = 0
         next_retrieve = 0.0
+        next_health_log = time.monotonic() + self.HEALTH_LOG_SECONDS
         while self.running:
             try:
                 # Try to grab always the latest frame
@@ -197,7 +224,12 @@ def _run(self):
                         f"Video stream recovered after {failures} failed grabs"
                     )
                     failures = 0
+                self._grabbed += 1
+                video_frames_grabbed_counter.add(1)
                 now = time.monotonic()
+                if now >= next_health_log:
+                    next_health_log = now + self.HEALTH_LOG_SECONDS
+                    self._log_health()
                 if now < next_retrieve:
                     continue
                 try:
@@ -216,6 +248,22 @@ def _run(self):
             failures += 1
             self._reopen(failures)
 
+    def _log_health(self):
+        """Log one line summarizing the last window, and start a new one
+
+        Reading it tells which layer stopped without attaching a debugger: no
+        line at all means the platform never asked for video (no camera module
+        loaded); ``grabbed=0`` means the stream is unreachable; frames grabbed
+        but not served means the streamer is not consuming them; frames served
+        with nothing visible in the platform points at the MQTT side.
+        """
+        self.logger.info(
+            f"Capture health: grabbed={self._grabbed} served={self._served} "
+            f"stale={self._stale} reopens={self._reopens} in the last "
+            f"{self.HEALTH_LOG_SECONDS:.0f}s"
+        )
+        self._grabbed = self._served = self._stale = self._reopens = 0
+
     def _reopen(self, failures):
         """Release and rebuild the capture, backing off first"""
         if failures == 1:
@@ -235,6 +283,8 @@ def _reopen(self, failures):
         except Exception as e:
             self.logger.error(f"Failed to reopen video stream {e}")
             return
+        self._reopens += 1
+        video_capture_reopens_counter.add(1)
         with self.capture_mutex:
             self.capture = capture