diff --git a/openpilot/starpilot/galaxy/camera_snapshot.py b/openpilot/starpilot/galaxy/camera_snapshot.py index 0be1273463..67e31fc9f8 100644 --- a/openpilot/starpilot/galaxy/camera_snapshot.py +++ b/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: diff --git a/openpilot/starpilot/galaxy/camera_snapshot_worker.py b/openpilot/starpilot/galaxy/camera_snapshot_worker.py index 83be981eb7..fedf28c8d1 100644 --- a/openpilot/starpilot/galaxy/camera_snapshot_worker.py +++ b/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 diff --git a/openpilot/starpilot/galaxy/tests/test_camera_snapshot.py b/openpilot/starpilot/galaxy/tests/test_camera_snapshot.py index 71f6a94a05..d56bb88388 100644 --- a/openpilot/starpilot/galaxy/tests/test_camera_snapshot.py +++ b/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}) diff --git a/openpilot/starpilot/galaxy/tests/test_cameras.mjs b/openpilot/starpilot/galaxy/tests/test_cameras.mjs index b4fea073ba..b62d8f8409 100644 --- a/openpilot/starpilot/galaxy/tests/test_cameras.mjs +++ b/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. diff --git a/openpilot/starpilot/galaxy/web/js/cameras.js b/openpilot/starpilot/galaxy/web/js/cameras.js index e20642dc0a..71c111e221 100644 --- a/openpilot/starpilot/galaxy/web/js/cameras.js +++ b/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]) diff --git a/prebuilt.json b/prebuilt.json index 6ba1e83a32..a7a8ea3a0b 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "b70105c5c6d81af8d335fd194275e59eb00a818f85508e1e733ece13326547b9", "schema_version": 1} +{"protected_inventory_sha256": "7d2c0493d8fb460d182d3c4f3db52e09047e435d65c90cad58050821d723575b", "schema_version": 1}