Camera UI

Serve fresh camera snapshots with explicit image responses and expose live crop controls without placing file or camera startup work on the UI thread.

Co-authored-by: Prabhaav Pillai <prabhaav@msu.edu>
Co-authored-by: dirwin31 <83434411+dirwin31@users.noreply.github.com>

openpilot/starpilot/galaxy/camera_request.py

@@ -0,0 +1,87 @@
+"""Short, namespaced camera leases. Manager alone owns camera processes."""
+import fcntl
+import hashlib
+import os
+from pathlib import Path
+import stat
+import tempfile
+import time
+
+TTL_NS = 15_000_000_000
+FRAME_TTL_NS = 5_000_000_000
+
+
+def boot_ns():
+  return time.clock_gettime_ns(getattr(time, 'CLOCK_BOOTTIME', time.CLOCK_MONOTONIC))
+
+
+def _path(name):
+  prefix = os.environ.get('OPENPILOT_PREFIX', 'd')
+  namespace = hashlib.sha256(prefix.encode()).hexdigest()[:16]
+  return Path('/dev/shm') / f'starpilot-camera-{os.getuid()}-{namespace}' / name
+
+
+def _read(path):
+  fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+  try:
+    info = os.fstat(fd)
+    if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_size > 32:
+      raise ValueError('Invalid camera lease')
+    return int(os.read(fd, 32))
+  finally:
+    os.close(fd)
+
+
+def _write(path, value):
+  path = Path(path)
+  path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+  info = path.parent.lstat()
+  if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077:
+    raise OSError('Camera lease directory is not private')
+  lock = os.open(path.with_name(path.name + '.lock'), os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600)
+  temporary = None
+  try:
+    info = os.fstat(lock)
+    if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid():
+      raise OSError('Invalid camera lease lock')
+    fcntl.flock(lock, fcntl.LOCK_EX)
+    try:
+      previous = _read(path)
+      if abs(previous - value) <= TTL_NS:
+        value = max(value, previous)
+    except (OSError, ValueError):
+      pass
+    fd, temporary = tempfile.mkstemp(prefix=path.name + '.', dir=path.parent)
+    try:
+      os.write(fd, str(value).encode())
+    finally:
+      os.close(fd)
+    os.replace(temporary, path)
+    temporary = None
+  finally:
+    if temporary is not None:
+      os.unlink(temporary)
+    os.close(lock)
+
+
+def request(clock=boot_ns, path=None):
+  _write(path or _path('lease'), clock() + TTL_NS)
+
+
+def requested(clock=boot_ns, path=None):
+  try:
+    remaining = _read(path or _path('lease')) - clock()
+    return 0 < remaining <= TTL_NS
+  except (OSError, ValueError):
+    return False
+
+
+def mark_frame(clock=boot_ns, path=None):
+  _write(path or _path('cabin-frame'), clock())
+
+
+def frame_available(clock=boot_ns, path=None):
+  try:
+    return 0 <= clock() - _read(path or _path('cabin-frame')) <= FRAME_TTL_NS
+  except (OSError, ValueError):
+    return False

openpilot/starpilot/galaxy/camera_snapshot.py

@@ -0,0 +1,71 @@
+"""One bounded parked snapshot from an already-running camera owner."""
+
+import subprocess
+import sys
+import threading
+import time
+
+
+CAMERAS = frozenset(("cabin", "wide", "narrow"))
+MAX_JPEG_BYTES = 1_000_000
+_CAPTURE_LOCK = threading.Lock()
+
+
+class SnapshotUnavailable(Exception):
+  pass
+
+
+class SnapshotDenied(SnapshotUnavailable):
+  pass
+
+
+class CameraSnapshot:
+  def __init__(self, *, spawn=subprocess.Popen, clock=time.monotonic):
+    self.spawn, self.clock = spawn, clock
+    self.last_request = {}
+
+  def capture(self, camera: str, *, permitted) -> bytes:
+    if camera not in CAMERAS:
+      raise ValueError("Unknown camera")
+    if not permitted():
+      raise SnapshotDenied
+    if not _CAPTURE_LOCK.acquire(blocking=False):
+      raise SnapshotUnavailable("Another snapshot is in progress")
+    process = None
+    try:
+      if self.clock() - self.last_request.get(camera, float('-inf')) < 1:
+        raise SnapshotUnavailable("Wait before another snapshot")
+      self.last_request[camera] = self.clock()
+      from openpilot.starpilot.galaxy.camera_request import request, boot_ns
+      request()
+      frame_after = boot_ns()
+      deadline = self.clock() + 12
+      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:
+        if not permitted():
+          raise SnapshotDenied
+        try:
+          body, _ = process.communicate(timeout=min(.1, max(.001, deadline - self.clock())))
+        except subprocess.TimeoutExpired:
+          continue
+        if not permitted():
+          raise SnapshotDenied
+        if (process.returncode != 0 or not 4 <= len(body) <= MAX_JPEG_BYTES or
+            not body.startswith(b'\xff\xd8') or not body.endswith(b'\xff\xd9')):
+          raise SnapshotUnavailable("No fresh camera frame")
+        if camera == "cabin":
+          from openpilot.starpilot.galaxy.camera_request import mark_frame
+          mark_frame(clock=lambda: frame_after)
+        return body
+      raise SnapshotUnavailable("Camera timed out")
+    except OSError as error:
+      raise SnapshotUnavailable("Camera unavailable") from error
+    finally:
+      if process is not None:
+        if process.poll() is None:
+          process.kill()
+        process.wait()
+        if process.stdout is not None:
+          process.stdout.close()
+      _CAPTURE_LOCK.release()

openpilot/starpilot/galaxy/camera_snapshot_worker.py

@@ -0,0 +1,61 @@
+"""CPU-only frame reader. Does not start or stop camerad or change Params."""
+
+from io import BytesIO
+import sys
+import time
+
+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.system.camerad.snapshot import extract_image
+
+
+STREAMS = {"cabin": VisionStreamType.VISION_STREAM_CABIN, "wide": VisionStreamType.VISION_STREAM_WIDE_ROAD,
+           "narrow": VisionStreamType.VISION_STREAM_NARROW_ROAD}
+
+
+def boot_ns():
+  return time.clock_gettime_ns(getattr(time, "CLOCK_BOOTTIME", time.CLOCK_MONOTONIC))
+
+
+def capture(camera, *, client_factory=VisionIpcClient, clock=boot_ns):
+  started = clock()
+  client = client_factory("camerad", STREAMS[camera], True)
+  while not client.connect(False):
+    if clock() - started > 9_000_000_000:
+      return None
+    time.sleep(.1)
+  while clock() - started < 10_000_000_000:
+    frame = client.recv(timeout_ms=100)
+    now = clock()
+    # Native camerad does not populate VIPC extra.valid; use capture freshness.
+    if frame is None or not started < client.timestamp_sof <= client.timestamp_eof <= now:
+      continue
+    if now - client.timestamp_sof > 500_000_000:
+      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
+        not width <= stride <= 8192 or stride % 2 or not stride * height <= uv_offset <= stride * 4096 or
+        uv_offset % stride or len(frame.data) < uv_offset + stride * uv_height):
+      return None
+    image = Image.fromarray(extract_image(frame))
+    image.thumbnail((1280, 720))
+    output = BytesIO()
+    image.save(output, format="JPEG", quality=85)
+    body = output.getvalue()
+    return body if len(body) <= MAX_JPEG_BYTES else None
+  return None
+
+
+if __name__ == '__main__':
+  try:
+    result = capture(sys.argv[1]) if len(sys.argv) == 2 and sys.argv[1] in STREAMS else None
+    if result is not None:
+      sys.stdout.buffer.write(result)
+    else:
+      sys.exit(1)
+  except Exception:
+    sys.exit(1)

openpilot/starpilot/galaxy/tests/test_camera_request.py

@@ -0,0 +1,32 @@
+from concurrent.futures import ThreadPoolExecutor
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from openpilot.starpilot.galaxy import camera_request
+
+
+class TestCameraRequest(unittest.TestCase):
+  def test_concurrent_latest_expiry_and_finite_lease(self):
+    with tempfile.TemporaryDirectory() as directory:
+      path = Path(directory) / 'lease'
+      with ThreadPoolExecutor(max_workers=8) as pool:
+        list(pool.map(lambda n: camera_request.request(clock=lambda: 100 + n, path=path), range(128)))
+      self.assertEqual(int(path.read_text()), 227 + camera_request.TTL_NS)
+      self.assertTrue(camera_request.requested(clock=lambda: 228, path=path))
+      self.assertFalse(camera_request.requested(clock=lambda: 227 + camera_request.TTL_NS, path=path))
+      self.assertFalse(camera_request.requested(clock=lambda: 1, path=path))
+
+  def test_prefix_isolation_and_frame_receipt_aging(self):
+    with patch.dict(os.environ, OPENPILOT_PREFIX='camera-a'):
+      first = camera_request._path('lease')
+    with patch.dict(os.environ, OPENPILOT_PREFIX='camera-b'):
+      self.assertNotEqual(first, camera_request._path('lease'))
+    with tempfile.TemporaryDirectory() as directory:
+      path = Path(directory) / 'frame'
+      camera_request.mark_frame(clock=lambda: 100, path=path)
+      self.assertTrue(camera_request.frame_available(clock=lambda: 101, path=path))
+      self.assertFalse(camera_request.frame_available(clock=lambda: 101 + camera_request.FRAME_TTL_NS, path=path))
+      self.assertFalse(camera_request.frame_available(clock=lambda: 99, path=path))

openpilot/starpilot/galaxy/tests/test_camera_snapshot.py

@@ -0,0 +1,125 @@
+from io import BytesIO
+import subprocess
+import sys
+from types import SimpleNamespace
+import unittest
+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_snapshot_worker import capture
+
+
+class CameraSnapshotTest(unittest.TestCase):
+  def setUp(self):
+    for name in ("request", "mark_frame"):
+      replacement = patch("openpilot.starpilot.galaxy.camera_request." + name)
+      setattr(self, name + "_mock", replacement.start())
+      self.addCleanup(replacement.stop)
+
+  def test_denied_does_not_launch_reader(self):
+    owner = CameraSnapshot(spawn=lambda *args, **kwargs: self.fail("launched"))
+    with self.assertRaises(SnapshotDenied):
+      owner.capture('cabin', permitted=lambda: False)
+    with self.assertRaises(ValueError):
+      owner.capture('../camera', permitted=lambda: True)
+
+  def test_revocation_kills_and_reaps_blocked_reader(self):
+    children = []
+    def spawn(args, **kwargs):
+      self.assertEqual(args[-1], 'cabin')
+      with self.assertRaises(SnapshotUnavailable):
+        CameraSnapshot(spawn=lambda *a, **k: self.fail("concurrent reader")).capture(
+          'wide', permitted=lambda: True)
+      children.append(subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'], **kwargs))
+      return children[-1]
+    calls = 0
+    def permitted():
+      nonlocal calls
+      calls += 1
+      return calls < 3
+    with self.assertRaises(SnapshotDenied):
+      CameraSnapshot(spawn=spawn).capture('cabin', permitted=permitted)
+    self.assertIsNotNone(children[0].poll())
+    self.assertTrue(children[0].stdout.closed)
+
+  def test_deadline_kills_blocked_native_reader(self):
+    children = []
+    ticks = iter(range(100))
+    def spawn(args, **kwargs):
+      children.append(subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(30)'], **kwargs))
+      return children[-1]
+    with self.assertRaises(SnapshotUnavailable):
+      CameraSnapshot(spawn=spawn, clock=lambda: next(ticks)).capture('cabin', permitted=lambda: True)
+    self.assertIsNotNone(children[0].poll())
+    self.assertTrue(children[0].stdout.closed)
+
+  def test_reader_result_and_final_authority_check(self):
+    def spawn(args, **kwargs):
+      return subprocess.Popen([sys.executable, '-c', "import sys; sys.stdout.buffer.write(bytes([255,216,255,217]))"], **kwargs)
+    owner = CameraSnapshot(spawn=spawn)
+    self.assertEqual(owner.capture('wide', permitted=lambda: True), b'\xff\xd8\xff\xd9')
+    self.assertEqual(owner.capture('cabin', permitted=lambda: True), b'\xff\xd8\xff\xd9')
+    self.assertEqual(self.request_mock.call_count, 2)
+    self.assertEqual(self.mark_frame_mock.call_count, 1) # Cabin availability receipt only.
+    with self.assertRaises(SnapshotUnavailable):
+      owner.capture('wide', permitted=lambda: True)
+    calls = 0
+    def permitted():
+      nonlocal calls
+      calls += 1
+      return calls < 3
+    with self.assertRaises(SnapshotDenied):
+      CameraSnapshot(spawn=spawn).capture('wide', permitted=permitted)
+
+  def test_real_nv12_conversion_accepts_only_post_request_fresh_native_frame(self):
+    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:
+      valid = False  # Native camerad leaves the extra metadata bit unset.
+      timestamp_sof = now - 1
+      timestamp_eof = now - 1
+      frames = 0
+      def connect(self, blocking):
+        self.blocking = blocking
+        return True
+      def recv(self, timeout_ms):
+        nonlocal now
+        self.frames += 1
+        now += 100_000_000
+        if self.frames > 1:
+          self.timestamp_sof = now - 20_000_000
+          self.timestamp_eof = now - 10_000_000
+        return frame
+    client = Client()
+    result = capture('cabin', client_factory=lambda *args: client, clock=lambda: now)
+    self.assertFalse(client.blocking)
+    self.assertEqual(client.frames, 2)
+    image = Image.open(BytesIO(result))
+    self.assertEqual(image.size, (4, 4))
+    self.assertEqual(image.mode, 'RGB')
+    frame.width = 99999
+    self.assertIsNone(capture('cabin', client_factory=lambda *args: client, clock=lambda: now))
+
+
+  def test_native_frames_with_stale_future_or_reversed_timestamps_never_convert(self):
+    for fault in ('stale', 'future', 'reversed'):
+      with self.subTest(fault=fault):
+        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:
+          valid = False
+          timestamp_sof = timestamp_eof = 0
+          def connect(self, blocking): return True
+          def recv(self, timeout_ms, *, fault=fault, frame=frame, now=now):
+            now[0] += 100_000_000
+            self.timestamp_sof = now[0] - (600_000_000 if fault == 'stale' else 20_000_000)
+            self.timestamp_eof = now[0] + 1 if fault == 'future' else self.timestamp_sof - 1 if fault == 'reversed' else now[0] - 10_000_000
+            return frame
+        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()

openpilot/starpilot/galaxy/tests/test_cameras.mjs

@@ -0,0 +1,66 @@
+import assert from "node:assert/strict"
+import { readFileSync } from "node:fs"
+import { CamerasPage, CameraSnapshotFeed } from "../web/js/cameras.js"
+import { SettingsFeed, SettingsPage } from "../web/js/settings.js"
+
+const requests = []
+const states = []
+const feed = new SettingsFeed({ publish: (state) => states.push(state),
+  fetcher: (url, options) => new Promise((resolve) => requests.push({ url, options, resolve })),
+  later: () => 1, cancelTimer: () => {} })
+feed.start("pip")
+assert.equal(requests[0].url, "./api/settings/pages/pip")
+assert.equal(feed.page, "pip")
+feed.stop()
+requests[0].resolve({ ok: true, status: 200, json: async () => ({ page: "pip", rows: [] }) })
+await Promise.resolve()
+await Promise.resolve()
+assert.equal(states.at(-1).status, "idle") // Hidden or signed-out page cannot restore saved values.
+
+const mounted = { mode: "local", initialPage: "pip", state: { developerOpen: false }, feed: { start(page) { assert.equal(page, "pip") } } }
+SettingsPage.mounted.call(mounted)
+let returned = 0
+const page = { initialPage: "pip", returnTo: () => returned++, state: { query: "anything", data: { page: "pip" } },
+  feed: { load() { throw new Error("Root PiP Back must return to Cameras") } } }
+SettingsPage.methods.back.call(page)
+assert.equal(returned, 1)
+assert.equal(page.state.query, "")
+
+assert.match(CamerasPage.template, /Adjust the camera crop with a live cabin preview/)
+assert.doesNotMatch(CamerasPage.template, /live camera preview is unavailable/)
+assert.match(CamerasPage.template, /Sentry[\s\S]*View motion events/)
+assert.match(CamerasPage.template, /Saved motion settings/)
+assert.match(CamerasPage.template, /V-ASM[\s\S]*Open saved settings/)
+assert.match(CamerasPage.template, /live camera image and current warning status are unavailable/)
+assert.match(CamerasPage.template, /snapshots.capture\(camera\)/)
+assert.match(SettingsPage.template, /\{\{ title \}\}/)
+const app = readFileSync(new URL("../web/js/app.js", import.meta.url), "utf8")
+assert.match(app, /<PipPage v-else-if="route\.path === '\/cameras\/pip'"/)
+assert.match(app, /route\.path === '\/cameras\/sentry-settings'[\s\S]*initial-page="sentry"/)
+assert.match(app, /CamerasPage v-else-if="route\.path === '\/cameras'"/)
+
+const captures = [], snapshots = [], revoked = []
+let timer
+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: () => {} })
+const pending = camera.capture("cabin")
+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 }) })
+await pending
+assert.equal(snapshots.at(-1).image, "blob:camera")
+camera.stop()
+assert.deepEqual(revoked, ["blob:camera"])
+const late = camera.capture("wide")
+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 }) })
+await late
+assert.equal(snapshots.at(-1).image, "")
+const timed = camera.capture("cabin")
+timer()
+captures[2].resolve({ status: 401 })
+await timed
+assert.match(snapshots.at(-1).error, /timed out/)