svc-vision: camera live view, moved from app-flutter's test tool

Camera-agnostic (--camera auto finds any V4L2 capture device and finds it
again after a replug; non-MJPEG cameras are encoded), says when the camera
is unavailable, paced per decision 28. Python, standard library only.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz

.editorconfig

@@ -0,0 +1,20 @@
+root = true
+
+[*]
+charset = utf-8
+end_of_line = lf
+insert_final_newline = true
+trim_trailing_whitespace = true
+indent_style = space
+
+[*.{rs,toml}]
+indent_size = 4
+
+[*.{dart,json,yaml,yml}]
+indent_size = 2
+
+[*.md]
+trim_trailing_whitespace = false
+
+[*.py]
+indent_size = 4

.gitignore

@@ -0,0 +1,2 @@
+__pycache__/
+*.pyc

README.md

@@ -0,0 +1,28 @@
+---
+type: reference
+---
+# Vision
+
+Part of **Project Yukart**, making an old 2008 Mitsubishi Z27AG a bit smarter with modern technologies.  
+This file serves as the minimalist portal to other docs
+
+
+## What this repo is about
+
+Runs on the Orin. Owns the camera: capture, the live view to the tablet, and
+later recording and inference. Camera-agnostic: any V4L2 capture device, in
+practice a USB (UVC) webcam.
+
+| Program | Installed as | Does |
+|---|---|---|
+| `src/liveview.py` | `yukart-vision.service` | Camera frames to display clients on `:7710`, paced to the link |
+
+```bash
+python3 src/liveview.py --camera auto          # anywhere with a camera, GStreamer and v4l-utils
+scripts/deploy.sh                              # install yukart-vision on the car Orin
+```
+
+The client is `app-flutter`'s live view (`lib/camview.dart`).
+
+- For implementation practice guidelines, refer to the [guidelines](docs/GUIDELINES.md);  
+- For implementation plans and history, refer to [them](docs/impl/);

deploy/yukart-vision.service

@@ -0,0 +1,15 @@
+[Unit]
+Description=Yukart vision: camera live view (svc-vision)
+Documentation=gitea:yukart/svc-vision
+After=network.target
+StartLimitIntervalSec=0
+
+[Service]
+User=mikkeli
+# --camera auto finds a camera, and finds it again after a replug; name a /dev/v4l/by-id path to pin one.
+ExecStart=/usr/bin/python3 /home/mikkeli/infra/svc-vision/liveview.py --camera auto --listen 0.0.0.0:7710
+Restart=always
+RestartSec=2
+
+[Install]
+WantedBy=multi-user.target

docs/GUIDELINES.md

@@ -0,0 +1,35 @@
+---
+type: guidelines
+---
+# Vision guidelines
+
+## 1. Camera-agnostic
+- Any V4L2 capture device works. Never hard-code a node such as `/dev/video0`:
+  numbering follows enumeration order. Take a `/dev/v4l/by-id/` path, or find
+  the camera at runtime (`--camera auto`).  
+- Use what the camera offers. MJPEG passes through untouched; anything else is
+  encoded here. A requested size or rate the camera lacks falls back to its own
+  default rather than failing.  
+- UVC control names differ between cameras and kernels. Look them up on the
+  device; reply that a control is missing rather than guessing.
+
+## 2. One process owns the camera
+- V4L2 capture is single-reader. This service opens the device; recording and
+  inference take frames from it, never from the device.  
+- UVC controls reset when the camera re-enumerates. Set them on open, not once.
+
+## 3. The tablet link is the constraint
+- Every socket to a display is paced (`SO_MAX_PACING_RATE`, project-yukart
+  decision 28). Unpaced bursts stall the USB link for seconds.  
+- Latest frame wins. A slow client skips frames; it never gets a queue of old ones.  
+- Say when there is no camera. A client must be able to tell "unavailable" from
+  "slow".
+
+## 4. Python, on purpose
+- Python 3.10 (the Orin's), standard library only; GStreamer and v4l-utils come
+  from the host. The exception to Rust for Orin services is decision 29: camera
+  pipelines and inference live in the Python ecosystem.  
+- No pip installs on the car Orin until there is a reason; then a venv under
+  `~/infra/svc-vision/`, never the system Python.
+
+## 5. To be added

docs/impl/plans/INITIAL.md

@@ -0,0 +1,31 @@
+---
+type: plan
+status: active
+updated: 2026-10-12
+---
+# Initial plan
+
+## Scope
+
+The camera end to end on the Orin: capture, the live view, recording, and later
+inference. Hardware facts for the current camera are in orin-nano-env-setup
+`docs/impl/reports/USB-CAMERA.md`; tablet link limits in app-flutter
+`docs/impl/reports/LINK-TEST-2026-10-11.md`.
+
+## Order
+
+0. **Live view** — in place: `src/liveview.py` as `yukart-vision.service` on
+   `:7710`. Started as a test tool in app-flutter and moved here once the live
+   view became a product feature (decision 27). Protocol in the module docstring.
+1. **Capture fan-out.** One capture process, frames to several consumers (live
+   view, recorder, inference). Needed before a second consumer exists.
+2. **Recording.** H.264 in software (`x264enc`, ~1.3–2 cores at 1080p) in
+   segments on the recorder's sessions, under the recordings bound.
+3. **Exposure for sign reading.** Manual exposure with own metering to hold the
+   frame rate; LED signs flicker under short exposures.
+4. **Inference** — deferred (decision 7): road signs, speed limits first.
+
+## Open
+
+- The protocol lives in the docstring. It moves to contracts when a second
+  client exists or it changes shape.

scripts/deploy.sh

@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+# Install svc-vision on an Orin as the yukart-vision service.
+#
+#   usage: scripts/deploy.sh [host]        (default: mikkeli-orin-nano)
+#
+# Python 3.10 and the standard library only: nothing to build. The host needs
+# gst-launch-1.0 (GStreamer, part of JetPack) and v4l2-ctl (v4l-utils).
+set -euo pipefail
+
+HOST="${1:-mikkeli-orin-nano}"
+cd "$(dirname "$0")/.."
+
+ssh "$HOST" 'set -e
+  command -v gst-launch-1.0 >/dev/null || { echo "missing gst-launch-1.0" >&2; exit 1; }
+  command -v v4l2-ctl >/dev/null || { echo "missing v4l2-ctl (v4l-utils)" >&2; exit 1; }
+  python3 -c "import sys; assert sys.version_info >= (3, 10)"
+  mkdir -p ~/infra/svc-vision'
+scp -q src/liveview.py "$HOST:infra/svc-vision/liveview.py.new"
+scp -q deploy/yukart-vision.service "$HOST:infra/svc-vision/yukart-vision.service"
+ssh "$HOST" 'set -e
+  mv ~/infra/svc-vision/liveview.py.new ~/infra/svc-vision/liveview.py
+  sudo install -m 0644 ~/infra/svc-vision/yukart-vision.service /etc/systemd/system/yukart-vision.service
+  sudo systemctl daemon-reload
+  sudo systemctl enable yukart-vision.service >/dev/null
+  sudo systemctl restart yukart-vision.service
+  sleep 2
+  systemctl --no-pager --lines=5 status yukart-vision.service'

src/liveview.py

@@ -0,0 +1,274 @@
+#!/usr/bin/env python3
+"""Camera live view: MJPEG frames to display clients on 7710/tcp (project-yukart decision 27).
+
+Camera-agnostic: any V4L2 capture device. --camera names one; "auto" takes the
+first USB camera by id, then any capture node, and looks again on every capture
+restart, so a camera plugged in later is found. MJPEG cameras pass through with
+no CPU; others are encoded here (jpegenc).
+
+  server → client, per frame:  [u32 length][f64 capture time, unix s][u32 seq][JPEG]
+               per text:       [u32 0][f64 time][u32 text length][UTF-8 text]
+  client → server, lines:      "exposure auto" | "exposure <100 µs units>" | "gain <n>"
+                               → one text reply, "ok <state>" or "err <reason>"
+
+Texts also carry availability: "camera unavailable" when no frame has come for
+UNAVAILABLE_S (no camera, or capture stopped), "camera ok" when frames resume.
+
+Latest-wins: each client gets the newest frame when it is ready for one, never a
+queue of old ones. Sockets are paced (SO_MAX_PACING_RATE, decision 28) because
+unpaced bursts stall the USB link to the tablet for seconds.
+
+Python 3.10, standard library only; needs gst-launch-1.0 and v4l2-ctl on the host.
+"""
+
+import argparse
+import glob
+import ipaddress
+import os
+import socket
+import struct
+import subprocess
+import threading
+import time
+
+SO_MAX_PACING_RATE = 47
+UNAVAILABLE_S = 3.0
+
+# UVC control names differ between kernels; the first one the camera has wins.
+EXPOSURE_MODE = ("auto_exposure", "exposure_auto")
+EXPOSURE_TIME = ("exposure_time_absolute", "exposure_absolute")
+GAIN = ("gain",)
+MODE_MANUAL, MODE_AUTO = 1, 3  # V4L2_EXPOSURE_MANUAL, V4L2_EXPOSURE_APERTURE_PRIORITY
+
+
+def log(msg):
+    print(msg, flush=True)
+
+
+def run(*args):
+    r = subprocess.run(args, capture_output=True, text=True)
+    return r.returncode == 0, (r.stdout + r.stderr).strip()
+
+
+def is_capture(dev):
+    ok, out = run("v4l2-ctl", "-d", dev, "--info")
+    # "Device Caps" lists what this node does; metadata nodes of the same camera lack Video Capture.
+    caps = out.split("Device Caps", 1)[-1]
+    return ok and "Video Capture" in caps
+
+
+def find_camera(choice):
+    if choice != "auto":
+        return choice if os.path.exists(choice) else None
+    candidates = sorted(glob.glob("/dev/v4l/by-id/*-video-index0")) + sorted(glob.glob("/dev/video*"))
+    for dev in candidates:
+        if is_capture(dev):
+            return dev
+    return None
+
+
+def has_mjpeg(dev):
+    _, out = run("v4l2-ctl", "-d", dev, "--list-formats")
+    return "'MJPG'" in out
+
+
+class Latest:
+    """The newest frame, and a condition to wait for the next one."""
+
+    def __init__(self):
+        self.cond = threading.Condition()
+        self.frame = None  # (seq, capture_time, jpeg)
+        self.dev = None  # the device being captured, for controls
+
+    def put(self, seq, t, jpeg):
+        with self.cond:
+            self.frame = (seq, t, jpeg)
+            self.cond.notify_all()
+
+    def after(self, seq, timeout):
+        with self.cond:
+            self.cond.wait_for(lambda: self.frame and self.frame[0] > seq, timeout)
+            return self.frame
+
+
+def pipeline(dev, w, h, fps, exact):
+    """gst-launch arguments; exact=False drops size and rate for cameras that do not offer them."""
+    size = [f"width={w}", f"height={h}", f"framerate={fps}/1"] if exact else []
+    if has_mjpeg(dev):
+        src = [f"image/jpeg,{','.join(size)}" if size else "image/jpeg"]
+    else:
+        src = [f"video/x-raw,{','.join(size)}" if size else "video/x-raw", "!", "videoconvert", "!", "jpegenc", "quality=80"]
+    # multipartmux frames each JPEG with a Content-Length header, so no JPEG parsing is needed.
+    return ["gst-launch-1.0", "-q", "v4l2src", f"device={dev}", "!", *src, "!",
+            "multipartmux", "boundary=yukartframe", "!", "fdsink", "fd=1"]
+
+
+def capture(choice, w, h, fps, latest):
+    seq = 0
+    exact = True
+    said = None
+    while True:
+        dev = find_camera(choice)
+        if dev is None:
+            if said != "none":
+                log(f"no camera ({choice}); looking again every 2 s")
+                said = "none"
+            latest.dev = None
+            time.sleep(2)
+            continue
+        latest.dev = dev
+        cmd = pipeline(dev, w, h, fps, exact)
+        if said != (dev, exact):
+            log(f"capturing {dev}: {' '.join(cmd[3:-6])}")
+            said = (dev, exact)
+        p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
+        out = p.stdout
+        got = 0
+        try:
+            while True:
+                length = None
+                while True:  # headers up to the blank line
+                    line = out.readline()
+                    if not line:
+                        raise EOFError
+                    line = line.strip()
+                    if line.lower().startswith(b"content-length:"):
+                        length = int(line.split(b":")[1])
+                    elif not line and length is not None:
+                        break
+                jpeg = out.read(length)
+                if len(jpeg) < length:
+                    raise EOFError
+                seq += 1
+                got += 1
+                latest.put(seq, time.time(), jpeg)
+        except EOFError:
+            pass
+        p.kill()
+        p.wait()
+        if got == 0:
+            exact = not exact  # the camera may not offer the requested mode; alternate with its own default
+        time.sleep(1)
+
+
+def control_name(dev, names):
+    _, out = run("v4l2-ctl", "-d", dev, "--list-ctrls")
+    have = {line.split()[0] for line in out.splitlines() if line.strip()}
+    return next((n for n in names if n in have), None)
+
+
+def control(latest, line):
+    dev = latest.dev
+    parts = line.split()
+    if dev is None:
+        return "err no camera"
+    if len(parts) != 2 or parts[0] not in ("exposure", "gain"):
+        return "err expected: exposure auto|<n> or gain <n>"
+    name, value = parts
+    mode, etime, gain = (control_name(dev, n) for n in (EXPOSURE_MODE, EXPOSURE_TIME, GAIN))
+    if name == "exposure" and value == "auto" and mode:
+        ok, msg = run("v4l2-ctl", "-d", dev, "-c", f"{mode}={MODE_AUTO}")
+    elif name == "exposure" and value.isdigit() and mode and etime:
+        ok, msg = run("v4l2-ctl", "-d", dev, "-c", f"{mode}={MODE_MANUAL}", "-c", f"{etime}={value}")
+    elif name == "gain" and value.isdigit() and gain:
+        ok, msg = run("v4l2-ctl", "-d", dev, "-c", f"{gain}={value}")
+    elif not value.isdigit() and value != "auto":
+        return "err bad value"
+    else:
+        return f"err this camera has no {name} control"
+    present = [n for n in (mode, etime, gain) if n]
+    _, state = run("v4l2-ctl", "-d", dev, "-C", ",".join(present))
+    return ("ok " if ok else "err ") + (msg or " ".join(state.split()))
+
+
+def serve_client(conn, peer, latest, pace):
+    conn.setsockopt(socket.SOL_SOCKET, SO_MAX_PACING_RATE, pace)
+    conn.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 256 * 1024)  # bound what can queue behind a frame
+    conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+    alive = threading.Event()
+    alive.set()
+    send = threading.Lock()  # frames and texts share the socket; never interleave them
+
+    def send_text(text):
+        # Texts go in-band as a frame with length 0 and the text after the header.
+        msg = text.encode()
+        with send:
+            conn.sendall(struct.pack("!IdI", 0, time.time(), len(msg)) + msg)
+
+    def reader():
+        buf = b""
+        try:
+            while alive.is_set():
+                b = conn.recv(1024)
+                if not b:
+                    break
+                buf += b
+                while b"\n" in buf:
+                    line, buf = buf.split(b"\n", 1)
+                    text = line.decode(errors="replace").strip()
+                    reply = control(latest, text)
+                    log(f"{peer}: {text} -> {reply}")
+                    send_text(reply)
+        except OSError:
+            pass
+        alive.clear()
+
+    threading.Thread(target=reader, daemon=True).start()
+    seq = 0
+    sent = 0
+    available = None  # unknown until the first frame or the first timeout
+    try:
+        while alive.is_set():
+            f = latest.after(seq, UNAVAILABLE_S)
+            if not f or f[0] <= seq:
+                if available is not False:
+                    available = False
+                    send_text("camera unavailable")
+                continue
+            if available is not True:
+                available = True
+                send_text("camera ok")
+            seq, t, jpeg = f
+            with send:
+                conn.sendall(struct.pack("!IdI", len(jpeg), t, seq) + jpeg)
+            sent += 1
+    except OSError:
+        pass
+    alive.clear()
+    log(f"{peer}: gone after {sent} frames")
+    conn.close()
+
+
+def allowed(ip):
+    a = ipaddress.ip_address(ip)
+    return a.is_loopback or a.is_private or a in ipaddress.ip_network("100.64.0.0/10")
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
+    ap.add_argument("--camera", default="auto", help='a V4L2 device path, or "auto"')
+    ap.add_argument("--size", default="1280x720")
+    ap.add_argument("--fps", type=int, default=30)
+    ap.add_argument("--listen", default="0.0.0.0:7710")
+    ap.add_argument("--pace", type=int, default=4_000_000, help="bytes/s per client")
+    a = ap.parse_args()
+    w, h = map(int, a.size.split("x"))
+    host, port = a.listen.rsplit(":", 1)
+    latest = Latest()
+    threading.Thread(target=capture, args=(a.camera, w, h, a.fps, latest), daemon=True).start()
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    s.bind((host, int(port)))
+    s.listen(4)
+    log(f"liveview: camera {a.camera}, {w}x{h}@{a.fps}, on tcp {a.listen}, paced {a.pace / 1e6} MB/s per client")
+    while True:
+        conn, addr = s.accept()
+        if not allowed(addr[0]):
+            conn.close()
+            continue
+        log(f"{addr[0]}: connected")
+        threading.Thread(target=serve_client, args=(conn, addr[0], latest, a.pace), daemon=True).start()
+
+
+if __name__ == "__main__":
+    main()