Wait for settled wide-camera snapshot frames

openpilot/starpilot/galaxy/camera_snapshot.py

@@ -8,6 +8,8 @@ import time
 
 CAMERAS = frozenset(("cabin", "wide", "narrow"))
 MAX_JPEG_BYTES = 1_000_000
+# Worker budgets leave two seconds for process startup, encoding and cleanup.
+CAPTURE_TIMEOUT_SECONDS = {"cabin": 24, "wide": 19, "narrow": 12}
 _CAPTURE_LOCK = threading.Lock()
 
 
@@ -39,7 +41,7 @@ class CameraSnapshot:
       from openpilot.starpilot.galaxy.camera_request import request, boot_ns
       request()
       frame_after = boot_ns()
-      deadline = self.clock() + (24 if camera == "cabin" else 12)
+      deadline = self.clock() + CAPTURE_TIMEOUT_SECONDS[camera]
       process = self.spawn([sys.executable, '-m', 'openpilot.starpilot.galaxy.camera_snapshot_worker', camera],
                            stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, close_fds=True)
       while self.clock() < deadline:

openpilot/starpilot/galaxy/camera_snapshot_worker.py

@@ -8,13 +8,14 @@ from PIL import Image
 
 from msgq.visionipc import VisionIpcClient
 from openpilot.cereal.visionipc import VisionStreamType
-from openpilot.starpilot.galaxy.camera_snapshot import MAX_JPEG_BYTES
+from openpilot.starpilot.galaxy.camera_snapshot import CAPTURE_TIMEOUT_SECONDS, MAX_JPEG_BYTES
 from openpilot.system.camerad.snapshot import extract_image
 
 
 STREAMS = {"cabin": VisionStreamType.VISION_STREAM_CABIN, "wide": VisionStreamType.VISION_STREAM_WIDE_ROAD,
            "narrow": VisionStreamType.VISION_STREAM_NARROW_ROAD}
 CABIN_SETTLE_NS = 10_000_000_000
+WIDE_SETTLE_NS = 5_000_000_000
 
 
 def boot_ns():
@@ -29,7 +30,9 @@ def capture(camera, *, client_factory=VisionIpcClient, clock=boot_ns):
       return None
     time.sleep(.1)
   connected = clock()
-  while clock() - started < (22_000_000_000 if camera == "cabin" else 10_000_000_000):
+  deadline = started + (CAPTURE_TIMEOUT_SECONDS[camera] - 2) * 1_000_000_000
+  wide_ready_after = None
+  while clock() < deadline:
     frame = client.recv(timeout_ms=100)
     now = clock()
     if camera == "cabin" and now - connected < CABIN_SETTLE_NS:
@@ -39,6 +42,13 @@ def capture(camera, *, client_factory=VisionIpcClient, clock=boot_ns):
       continue
     if now - client.timestamp_sof > 500_000_000:
       continue
+    if camera == "wide":
+      # A live IPC connection can precede camera frames. Count warm-up from the
+      # first fresh frame, then require an exposure captured after that interval.
+      if wide_ready_after is None:
+        wide_ready_after = client.timestamp_sof + WIDE_SETTLE_NS
+      if client.timestamp_sof < wide_ready_after:
+        continue
     width, height, stride, uv_offset = frame.width, frame.height, frame.stride, frame.uv_offset
     uv_height = ((height // 2 + 15) // 16) * 16
     if (not 0 < width <= 4096 or not 0 < height <= 4096 or width % 2 or height % 2 or

openpilot/starpilot/galaxy/tests/test_camera_snapshot.py

@@ -8,7 +8,8 @@ from unittest.mock import patch
 from PIL import Image
 import numpy as np
 
-from openpilot.starpilot.galaxy.camera_snapshot import CameraSnapshot, SnapshotDenied, SnapshotUnavailable
+from openpilot.starpilot.galaxy.camera_request import TTL_NS
+from openpilot.starpilot.galaxy.camera_snapshot import CAPTURE_TIMEOUT_SECONDS, CameraSnapshot, SnapshotDenied, SnapshotUnavailable
 from openpilot.starpilot.galaxy.camera_snapshot_worker import capture
 
 
@@ -123,3 +124,137 @@ class CameraSnapshotTest(unittest.TestCase):
         with patch('openpilot.starpilot.galaxy.camera_snapshot_worker.extract_image') as convert:
           self.assertIsNone(capture('cabin', client_factory=lambda *args: Client(), clock=lambda now=now: now[0]))
           convert.assert_not_called()
+
+  def test_only_wide_adds_five_seconds_of_fresh_frames(self):
+    for camera, expected_frames in (('narrow', 1), ('wide', 51), ('cabin', 100)):
+      with self.subTest(camera=camera):
+        now = [1_000_000_000]
+        data = np.full(80, 128, dtype=np.uint8)
+        frame = SimpleNamespace(width=4, height=4, stride=4, uv_offset=16, data=data)
+        class Client:
+          frames = 0
+          def connect(self, blocking): return True
+          def recv(self, timeout_ms, *, now=now, data=data, frame=frame):
+            self.frames += 1
+            now[0] += 100_000_000
+            self.timestamp_sof, self.timestamp_eof = now[0] - 20_000_000, now[0] - 10_000_000
+            data[:16] = 0 if self.frames <= 50 else 128
+            return frame
+        client = Client()
+        body = capture(camera, client_factory=lambda *args, client=client: client, clock=lambda now=now: now[0])
+        self.assertEqual(client.frames, expected_frames)
+        self.assertEqual(float(np.asarray(Image.open(BytesIO(body))).mean()), 0 if camera == 'narrow' else 128)
+
+  def test_wide_settles_after_delayed_connection_and_first_frame(self):
+    started = 1_000_000_000
+    now = [started]
+    frame = SimpleNamespace(width=4, height=4, stride=4, uv_offset=16, data=np.full(80, 128, dtype=np.uint8))
+    class Client:
+      def connect(self, blocking): return now[0] >= started + 8_900_000_000
+      def recv(self, timeout_ms):
+        now[0] += 100_000_000
+        if now[0] < started + 10_500_000_000:
+          return None
+        self.timestamp_sof, self.timestamp_eof = now[0] - 20_000_000, now[0] - 10_000_000
+        return frame
+    def sleep(seconds):
+      now[0] += int(seconds * 1e9)
+    with patch('openpilot.starpilot.galaxy.camera_snapshot_worker.time.sleep', side_effect=sleep):
+      body = capture('wide', client_factory=lambda *args: Client(), clock=lambda: now[0])
+    self.assertIsNotNone(body)
+    self.assertEqual(now[0] - started, 15_500_000_000)
+
+  def test_wide_rejects_pre_settle_exposure_received_after_boundary(self):
+    now = [1_000_000_000]
+    frame = SimpleNamespace(width=4, height=4, stride=4, uv_offset=16, data=np.full(80, 128, dtype=np.uint8))
+    class Client:
+      frames = 0
+      def connect(self, blocking): return True
+      def recv(self, timeout_ms):
+        self.frames += 1
+        if self.frames == 1:
+          now[0] += 100_000_000
+          self.timestamp_sof = now[0] - 20_000_000
+          self.ready_after = self.timestamp_sof + 5_000_000_000
+        else:
+          now[0] = self.ready_after + 200_000_000
+          self.timestamp_sof = self.ready_after - (1 if self.frames == 2 else 0)
+        self.timestamp_eof = now[0] - 10_000_000
+        return frame
+    client = Client()
+    self.assertIsNotNone(capture('wide', client_factory=lambda *args: client, clock=lambda: now[0]))
+    self.assertEqual(client.frames, 3)
+    self.assertEqual(client.timestamp_sof, client.ready_after)
+
+  def test_invalid_wide_frames_do_not_start_settling(self):
+    for fault in ('missing', 'pre_request', 'stale', 'future', 'reversed'):
+      with self.subTest(fault=fault):
+        started = 1_000_000_000
+        now = [started]
+        frame = SimpleNamespace(width=4, height=4, stride=4, uv_offset=16, data=np.full(80, 128, dtype=np.uint8))
+        class Client:
+          frames = 0
+          def connect(self, blocking): return True
+          def recv(self, timeout_ms, *, fault=fault, now=now, frame=frame, started=started):
+            self.frames += 1
+            now[0] += 3_000_000_000 if self.frames == 2 else 100_000_000
+            self.timestamp_sof, self.timestamp_eof = now[0] - 20_000_000, now[0] - 10_000_000
+            if self.frames == 1:
+              if fault == 'missing':
+                return None
+              if fault == 'pre_request':
+                self.timestamp_sof = started
+              if fault == 'stale':
+                self.timestamp_sof = now[0] - 600_000_000
+              if fault == 'future':
+                self.timestamp_eof = now[0] + 1
+              if fault == 'reversed':
+                self.timestamp_eof = self.timestamp_sof - 1
+            return frame
+        client = Client()
+        self.assertIsNotNone(capture('wide', client_factory=lambda *args, client=client: client, clock=lambda now=now: now[0]))
+        self.assertEqual(client.frames, 52)
+        self.assertEqual(now[0] - started, 8_100_000_000)
+
+  def test_wide_missing_or_late_frames_timeout_inside_owner_and_lease(self):
+    for late_frame in (False, True):
+      with self.subTest(late_frame=late_frame):
+        now = [1_000_000_000]
+        frame = SimpleNamespace(width=4, height=4, stride=4, uv_offset=16, data=np.full(80, 128, dtype=np.uint8))
+        class Client:
+          def connect(self, blocking): return True
+          def recv(self, timeout_ms, *, now=now, frame=frame, late_frame=late_frame):
+            now[0] += 100_000_000
+            self.timestamp_sof, self.timestamp_eof = now[0] - 20_000_000, now[0] - 10_000_000
+            return frame if late_frame and now[0] >= 15_000_000_000 else None
+        with patch('openpilot.starpilot.galaxy.camera_snapshot_worker.extract_image') as convert:
+          self.assertIsNone(capture('wide', client_factory=lambda *args: Client(), clock=lambda now=now: now[0]))
+          convert.assert_not_called()
+        self.assertEqual(now[0] - 1_000_000_000, 17_000_000_000)
+        self.assertLess(17, CAPTURE_TIMEOUT_SECONDS['wide'])
+        self.assertLess(CAPTURE_TIMEOUT_SECONDS['wide'] * 1_000_000_000, TTL_NS)
+
+  def test_wide_owner_reaps_reader_on_deadline_or_revocation_during_settling(self):
+    for revoke_at in (None, 3):
+      with self.subTest(revoke_at=revoke_at):
+        now = [0.0]
+        class Process:
+          stdout = BytesIO()
+          killed = waited = False
+          def communicate(self, timeout, *, now=now):
+            now[0] += timeout
+            raise subprocess.TimeoutExpired('snapshot worker', timeout)
+          def poll(self): return -9 if self.killed else None
+          def kill(self): self.killed = True
+          def wait(self): self.waited = True
+        process = Process()
+        def spawn(args, *, process=process, **kwargs):
+          self.assertEqual(args[-1], 'wide')
+          return process
+        owner = CameraSnapshot(spawn=spawn, clock=lambda now=now: now[0])
+        error = SnapshotUnavailable if revoke_at is None else SnapshotDenied
+        with self.assertRaises(error):
+          owner.capture('wide', permitted=lambda now=now, revoke_at=revoke_at: revoke_at is None or now[0] < revoke_at)
+        self.assertAlmostEqual(now[0], 19 if revoke_at is None else revoke_at)
+        self.assertTrue(process.killed and process.waited and process.stdout.closed)
+        self.assertEqual(CAPTURE_TIMEOUT_SECONDS, {'cabin': 24, 'wide': 19, 'narrow': 12})

openpilot/starpilot/galaxy/tests/test_cameras.mjs

@@ -42,12 +42,13 @@ assert.match(app, /SentryEventsPage[^>]*\/cameras\/events[^>]*\/cameras\/sentry-
 assert.match(app, /CamerasPage v-else-if="route\.path === '\/cameras'"/)
 
 const captures = [], snapshots = [], revoked = []
-let timer
+let timer, delay
 const camera = new CameraSnapshotFeed({ publish: (value) => snapshots.push(value), unauthorized: () => snapshots.push("unauthorized"),
   fetcher: (url, options) => new Promise((resolve) => captures.push({ url, options, resolve })),
   createURL: () => "blob:camera", revokeURL: (url) => revoked.push(url),
-  later: (fn) => { timer = fn; return 1 }, cancelTimer: () => {} })
+  later: (fn, ms) => { timer = fn; delay = ms; return 1 }, cancelTimer: () => {} })
 const pending = camera.capture("cabin")
+assert.equal(delay, 26000)
 assert.equal(captures[0].url, "./api/cameras/snapshot")
 assert.deepEqual(JSON.parse(captures[0].options.body), { camera: "cabin" })
 captures[0].resolve({ ok: true, status: 200, headers: new Map([["Content-Type", "image/jpeg; charset=binary"]]), blob: async () => ({ size: 100 }) })
@@ -56,6 +57,7 @@ assert.equal(snapshots.at(-1).image, "blob:camera")
 camera.stop()
 assert.deepEqual(revoked, ["blob:camera"])
 const late = camera.capture("wide")
+assert.equal(delay, 20500) // Leave room for startup, five seconds of fresh frames, and the 19s owner deadline.
 camera.stop()
 assert.equal(captures[1].options.signal.aborted, true)
 captures[1].resolve({ ok: true, status: 200, headers: new Map([["Content-Type", "image/jpeg"]]), blob: async () => ({ size: 100 }) })
@@ -66,6 +68,10 @@ timer()
 captures[2].resolve({ status: 401 })
 await timed
 assert.match(snapshots.at(-1).error, /timed out/)
+const narrow = camera.capture("narrow")
+assert.equal(delay, 13500)
+camera.stop()
+await narrow
 
 // Browsers throw "Illegal invocation" when setTimeout/clearTimeout run with a `this`
 // other than the global object; Node does not, so mimic the browser check here.

openpilot/starpilot/galaxy/web/js/cameras.js

@@ -35,7 +35,7 @@ export class CameraSnapshotFeed {
     this.timer = this.later(() => {
       if (this.request !== request) return
       expire(new Error("Snapshot timed out. Try taking a new snapshot."))
-    }, camera === "cabin" ? 26000 : 13500)
+    }, camera === "cabin" ? 26000 : camera === "wide" ? 20500 : 13500)
     try {
       const response = await Promise.race([this.fetcher("./api/cameras/snapshot", { method: "POST", credentials: "same-origin",
         cache: "no-store", signal: request.signal, headers: { "Content-Type": "application/json" }, body: JSON.stringify({ camera }) }), deadline])

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "b70105c5c6d81af8d335fd194275e59eb00a818f85508e1e733ece13326547b9", "schema_version": 1}
+{"protected_inventory_sha256": "7d2c0493d8fb460d182d3c4f3db52e09047e435d65c90cad58050821d723575b", "schema_version": 1}