diff --git a/lib/camview.dart b/lib/camview.dart new file mode 100644 index 0000000..8a2987f --- /dev/null +++ b/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 = []; + 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 stop() async { + _running = false; + _s?.destroy(); + } + + void send(String line) => _s?.write('$line\n'); + + Future _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.delayed(const Duration(seconds: 1)); + } + } + + Future _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(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 createState() => _CamViewPageState(); +} + +class _CamViewPageState extends State { + 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), + ), + ], + ), + ), + ], + ); + }, + ), + ), + ); + } +} diff --git a/lib/main.dart b/lib/main.dart index 4629c69..ce5fc4f 100644 --- a/lib/main.dart +++ b/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(builder: (_) => CamViewPage(host: Uri.parse(b.url).host)), + ), + icon: const Icon(Icons.videocam), + label: const Text('Camera'), + ), ]), ), ), diff --git a/tool/camview_server.py b/tool/camview_server.py new file mode 100644 index 0000000..dcd98bc --- /dev/null +++ b/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 " or "err " + +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| or gain " + 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()