Camera live view (test): MJPEG over TCP from tool/camview_server.py, exposure and gain controls

Its own paced connection beside the boundary (project-yukart decision 27,
proposed); latest-wins per client, SO_MAX_PACING_RATE 4 MB/s.

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

lib/camview.dart

@@ -0,0 +1,259 @@
+// Camera live view, for testing: shows the Orin camera's MJPEG frames as they
+// arrive, with exposure and gain controls. Pairs with tool/camview_server.py.
+// Not the boundary: its own connection (project-yukart decision 27, proposed).
+
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+import 'dart:typed_data';
+
+import 'package:flutter/material.dart';
+
+const camviewPort = 7710;
+
+class CamView extends ChangeNotifier {
+  CamView(this.host);
+  final String host;
+
+  Socket? _s;
+  bool _running = false;
+  String status = 'not started';
+  Uint8List? frame;
+  int frames = 0;
+  double fps = 0, kbPerFrame = 0, mbPerS = 0;
+
+  /// Delay beyond the fastest of the last ~300 frames: queueing, independent of the
+  /// tens of ms by which the Orin's and the tablet's clocks disagree.
+  double lagMs = 0;
+  final _delays = <double>[];
+  String lastReply = '';
+
+  final _window = <(int, int)>[]; // (arrival µs, bytes) over the last second
+  final _clock = Stopwatch()..start();
+
+  void start() {
+    if (_running) return;
+    _running = true;
+    _loop();
+  }
+
+  Future<void> stop() async {
+    _running = false;
+    _s?.destroy();
+  }
+
+  void send(String line) => _s?.write('$line\n');
+
+  Future<void> _loop() async {
+    while (_running) {
+      status = 'connecting to $host:$camviewPort';
+      notifyListeners();
+      try {
+        final s = await Socket.connect(
+          host,
+          camviewPort,
+          timeout: const Duration(seconds: 3),
+        );
+        _s = s;
+        status = 'connected';
+        notifyListeners();
+        await _read(s);
+        status = 'closed';
+      } catch (e) {
+        status = '$e'.split('\n').first;
+      }
+      _s = null;
+      notifyListeners();
+      if (_running) await Future<void>.delayed(const Duration(seconds: 1));
+    }
+  }
+
+  Future<void> _read(Socket s) async {
+    var data = Uint8List(0); // unconsumed bytes
+    await for (final chunk in s) {
+      if (data.isEmpty) {
+        data = chunk;
+      } else {
+        data =
+            (BytesBuilder(copy: false)
+                  ..add(data)
+                  ..add(chunk))
+                .takeBytes();
+      }
+      var off = 0;
+      while (data.length - off >= 16) {
+        final h = ByteData.sublistView(data, off, off + 16);
+        final len = h.getUint32(0);
+        final sentAt = h.getFloat64(4);
+        final body = len == 0
+            ? h.getUint32(12)
+            : len; // length 0: a text reply, its length in the third field
+        if (data.length - off - 16 < body) break;
+        final bytes = Uint8List.sublistView(data, off + 16, off + 16 + body);
+        off += 16 + body;
+        if (len == 0) {
+          lastReply = utf8.decode(bytes);
+        } else {
+          _onFrame(Uint8List.fromList(bytes), sentAt);
+        }
+      }
+      data = Uint8List.sublistView(data, off);
+    }
+  }
+
+  void _onFrame(Uint8List jpeg, double sentAt) {
+    final now = _clock.elapsedMicroseconds;
+    _window.add((now, jpeg.length));
+    _window.removeWhere((e) => now - e.$1 > 1000000);
+    frames++;
+    fps = _window.length.toDouble();
+    final bytes = _window.fold<int>(0, (a, e) => a + e.$2);
+    mbPerS = bytes / 1e6;
+    kbPerFrame = jpeg.length / 1024;
+    final delay = DateTime.now().microsecondsSinceEpoch / 1000 - sentAt * 1000;
+    _delays.add(delay);
+    if (_delays.length > 300) _delays.removeAt(0);
+    lagMs = delay - _delays.reduce((a, b) => a < b ? a : b);
+    frame = jpeg;
+    notifyListeners();
+  }
+
+  @override
+  void dispose() {
+    _running = false;
+    _s?.destroy();
+    super.dispose();
+  }
+}
+
+class CamViewPage extends StatefulWidget {
+  const CamViewPage({super.key, required this.host});
+  final String host;
+
+  @override
+  State<CamViewPage> createState() => _CamViewPageState();
+}
+
+class _CamViewPageState extends State<CamViewPage> {
+  late final cam = CamView(widget.host)..start();
+  bool auto = true;
+  double exposure = 100; // 100 µs units: 10 ms
+  double gain = 0;
+
+  @override
+  void dispose() {
+    cam.dispose();
+    super.dispose();
+  }
+
+  @override
+  Widget build(BuildContext context) {
+    return Scaffold(
+      backgroundColor: Colors.black,
+      appBar: AppBar(title: const Text('Camera (test)')),
+      body: SafeArea(
+        child: ListenableBuilder(
+          listenable: cam,
+          builder: (context, _) {
+            final f = cam.frame;
+            return Column(
+              crossAxisAlignment: CrossAxisAlignment.stretch,
+              children: [
+                Expanded(
+                  child: f == null
+                      ? Center(
+                          child: Text(
+                            cam.status,
+                            style: const TextStyle(color: Colors.white70),
+                          ),
+                        )
+                      : Image.memory(
+                          f,
+                          gaplessPlayback: true,
+                          fit: BoxFit.contain,
+                        ),
+                ),
+                Container(
+                  color: Theme.of(context).colorScheme.surfaceContainerHigh,
+                  padding: const EdgeInsets.symmetric(
+                    horizontal: 12,
+                    vertical: 6,
+                  ),
+                  child: Wrap(
+                    spacing: 16,
+                    runSpacing: 4,
+                    crossAxisAlignment: WrapCrossAlignment.center,
+                    children: [
+                      Text(cam.status),
+                      Text('${cam.fps.toStringAsFixed(0)} fps'),
+                      Text('${cam.kbPerFrame.toStringAsFixed(0)} KB/frame'),
+                      Text('${cam.mbPerS.toStringAsFixed(2)} MB/s'),
+                      Text('lag +${cam.lagMs.toStringAsFixed(0)} ms'),
+                      FilterChip(
+                        label: const Text('Auto exposure'),
+                        selected: auto,
+                        onSelected: (v) {
+                          setState(() => auto = v);
+                          cam.send(
+                            v
+                                ? 'exposure auto'
+                                : 'exposure ${exposure.round()}',
+                          );
+                        },
+                      ),
+                      SizedBox(
+                        width: 260,
+                        child: Row(
+                          children: [
+                            Text(
+                              'exp ${(exposure / 10).toStringAsFixed(1)} ms',
+                            ),
+                            Expanded(
+                              child: Slider(
+                                min: 1,
+                                max: 600,
+                                value: exposure,
+                                onChanged: auto
+                                    ? null
+                                    : (v) => setState(() => exposure = v),
+                                onChangeEnd: (v) =>
+                                    cam.send('exposure ${v.round()}'),
+                              ),
+                            ),
+                          ],
+                        ),
+                      ),
+                      SizedBox(
+                        width: 220,
+                        child: Row(
+                          children: [
+                            Text('gain ${gain.round()}'),
+                            Expanded(
+                              child: Slider(
+                                min: 0,
+                                max: 240,
+                                value: gain,
+                                onChanged: (v) => setState(() => gain = v),
+                                onChangeEnd: (v) =>
+                                    cam.send('gain ${v.round()}'),
+                              ),
+                            ),
+                          ],
+                        ),
+                      ),
+                      if (cam.lastReply.isNotEmpty)
+                        Text(
+                          cam.lastReply,
+                          style: const TextStyle(fontSize: 11),
+                        ),
+                    ],
+                  ),
+                ),
+              ],
+            );
+          },
+        ),
+      ),
+    );
+  }
+}

lib/main.dart

@@ -6,6 +6,7 @@ import 'dart:math';
 import 'package:flutter/material.dart';
 
 import 'boundary.dart';
+import 'camview.dart';
 import 'linktest.dart';
 import 'net.dart';
 
@@ -170,6 +171,13 @@ class LinkBar extends StatelessWidget {
               icon: const Icon(Icons.speed),
               label: const Text('Link test'),
             ),
+            TextButton.icon(
+              onPressed: () => Navigator.of(context).push(
+                MaterialPageRoute<void>(builder: (_) => CamViewPage(host: Uri.parse(b.url).host)),
+              ),
+              icon: const Icon(Icons.videocam),
+              label: const Text('Camera'),
+            ),
           ]),
         ),
       ),

tool/camview_server.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""Camera live view for testing (lib/camview.dart). Development tool, not the vision service.
+
+Streams the USB camera's own MJPEG frames, unmodified, to any client on 7710/tcp:
+
+  server → client, per frame:  [u32 length][f64 capture time, unix s][u32 seq][JPEG]
+               per reply:      [u32 0][f64 time][u32 text length][UTF-8 text]
+  client → server, lines:      "exposure auto" | "exposure <100 µs units>" | "gain <0-240>"
+                               → one reply, "ok <state>" or "err <reason>"
+
+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, project-yukart decision
+28) because unpaced bursts stall the USB link to the tablet for seconds.
+
+Run on the Orin: python3 camview_server.py [--size 1280x720] [--fps 30]
+"""
+
+import argparse
+import ipaddress
+import socket
+import struct
+import subprocess
+import threading
+import time
+
+SO_MAX_PACING_RATE = 47
+CONTROLS = {"exposure", "gain"}
+
+
+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)
+
+    def put(self, seq, t, jpeg):
+        with self.cond:
+            self.frame = (seq, t, jpeg)
+            self.cond.notify_all()
+
+    def after(self, seq, timeout=5.0):
+        with self.cond:
+            self.cond.wait_for(lambda: self.frame and self.frame[0] > seq, timeout)
+            return self.frame
+
+
+def capture(dev, w, h, fps, latest):
+    # multipartmux frames each JPEG with a Content-Length header, so no JPEG parsing is needed.
+    cmd = ["gst-launch-1.0", "-q", "v4l2src", f"device={dev}", "!",
+           f"image/jpeg,width={w},height={h},framerate={fps}/1", "!",
+           "multipartmux", "boundary=yukartframe", "!", "fdsink", "fd=1"]
+    seq = 0
+    while True:
+        p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
+        out = p.stdout
+        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
+                latest.put(seq, time.time(), jpeg)
+        except EOFError:
+            print("capture ended; restarting in 1 s", flush=True)
+        p.kill()
+        p.wait()
+        time.sleep(1)
+
+
+def v4l2(dev, *args):
+    r = subprocess.run(["v4l2-ctl", "-d", dev, *args], capture_output=True, text=True)
+    return r.returncode == 0, (r.stdout + r.stderr).strip()
+
+
+def control(dev, line):
+    parts = line.split()
+    if len(parts) != 2 or parts[0] not in CONTROLS:
+        return "err expected: exposure auto|<n> or gain <n>"
+    name, value = parts
+    if name == "exposure" and value == "auto":
+        ok, msg = v4l2(dev, "-c", "auto_exposure=3")
+    elif name == "exposure" and value.isdigit():
+        ok, msg = v4l2(dev, "-c", "auto_exposure=1", "-c", f"exposure_time_absolute={value}")
+    elif name == "gain" and value.isdigit():
+        ok, msg = v4l2(dev, "-c", f"gain={value}")
+    else:
+        return "err bad value"
+    _, state = v4l2(dev, "-C", "auto_exposure,exposure_time_absolute,gain")
+    return ("ok " if ok else "err ") + (msg or " ".join(state.split()))
+
+
+def serve_client(conn, peer, dev, 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 replies share the socket; never interleave them
+
+    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)
+                    reply = control(dev, line.decode(errors="replace").strip())
+                    print(f"{peer}: {line.decode(errors='replace').strip()} -> {reply}", flush=True)
+                    # Replies go in-band as a frame with length 0 and the text after the header.
+                    msg = reply.encode()
+                    with send:
+                        conn.sendall(struct.pack("!IdI", 0, time.time(), len(msg)) + msg)
+        except OSError:
+            pass
+        alive.clear()
+
+    threading.Thread(target=reader, daemon=True).start()
+    seq = 0
+    sent = 0
+    try:
+        while alive.is_set():
+            f = latest.after(seq)
+            if not f or f[0] <= seq:
+                continue
+            seq, t, jpeg = f
+            with send:
+                conn.sendall(struct.pack("!IdI", len(jpeg), t, seq) + jpeg)
+            sent += 1
+    except OSError:
+        pass
+    alive.clear()
+    print(f"{peer}: gone after {sent} frames", flush=True)
+    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()
+    ap.add_argument("--dev", default="/dev/video0")
+    ap.add_argument("--size", default="1280x720")
+    ap.add_argument("--fps", type=int, default=30)
+    ap.add_argument("--port", type=int, default=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"))
+    latest = Latest()
+    threading.Thread(target=capture, args=(a.dev, 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(("0.0.0.0", a.port))
+    s.listen(4)
+    print(f"camview: {a.dev} {w}x{h}@{a.fps} MJPEG on tcp {a.port}, paced {a.pace / 1e6} MB/s per client", flush=True)
+    while True:
+        conn, addr = s.accept()
+        if not allowed(addr[0]):
+            conn.close()
+            continue
+        print(f"{addr[0]}: connected", flush=True)
+        threading.Thread(target=serve_client, args=(conn, addr[0], a.dev, latest, a.pace), daemon=True).start()
+
+
+if __name__ == "__main__":
+    main()