From f470e9e86e1c8d1c4e18c144ad4c01d74e42e7f5 Mon Sep 17 00:00:00 2001 From: Comma Device Date: Fri, 4 Sep 2026 21:08:17 +0800 Subject: [PATCH] ui streamer: low-latency /ctrl (canvas + WS binary, latest-frame, half-res) - /ctrl: replace MJPEG with WebSocket binary frames + canvas (createImageBitmap), cutting browser buffering latency - capture: only encode the newest frame (drop backlog) - encode at STREAM_SCALE (default 0.5) + configurable fps/quality - add /config endpoint for live tuning without restart - launch_env defaults: STREAM_FPS=20, STREAM_QUALITY=45, STREAM_SCALE=0.5 --- launch_env.sh | 5 +- openpilot/system/ui/lib/ui_stream.py | 129 ++++++++++++++++++++++----- 2 files changed, 112 insertions(+), 22 deletions(-) diff --git a/launch_env.sh b/launch_env.sh index 9f9733810..3faa6ba35 100755 --- a/launch_env.sh +++ b/launch_env.sh @@ -24,5 +24,6 @@ export STAGING_ROOT="/data/safe_staging" # Comma4-UI-Streamer (MJPEG UI stream) - set STREAM=0 to disable export STREAM=1 export STREAM_PORT=8082 -export STREAM_QUALITY=50 -export STREAM_FPS=10 +export STREAM_QUALITY=45 +export STREAM_FPS=20 +export STREAM_SCALE=0.5 diff --git a/openpilot/system/ui/lib/ui_stream.py b/openpilot/system/ui/lib/ui_stream.py index 39ffa84e3..8bc8cb24c 100644 --- a/openpilot/system/ui/lib/ui_stream.py +++ b/openpilot/system/ui/lib/ui_stream.py @@ -9,6 +9,9 @@ import time import base64 import hashlib import struct +import base64 +import hashlib +import struct from http.server import HTTPServer, BaseHTTPRequestHandler from socketserver import ThreadingMixIn @@ -39,8 +42,7 @@ class StreamState: self.event.clear() -_CTRL_HTML = '\n\nopenpilot remote\n\n\n
CTRL: OFF
\n
ENABLE TOUCH
\n' - +_CTRL_HTML = '\n\nopenpilot remote\n\n\n
CTRL: OFF
\n
q45 f20 s0.5
\n
ENABLE TOUCH
\n' _OVERLAY_HTML = """ openpilot live @@ -231,6 +233,10 @@ class StreamHandler(BaseHTTPRequestHandler): self.send_header("Content-Length", str(len(_CTRL_HTML))) self.end_headers() self.wfile.write(_CTRL_HTML.encode()) + elif self.path == "/wsstream": + _handle_ws_stream(self) + elif self.path.startswith("/config"): + self._handle_config() elif self.path == "/stream": self.send_response(200) self.send_header("Content-Type", "multipart/x-mixed-replace; boundary=--frame") @@ -285,6 +291,26 @@ class StreamHandler(BaseHTTPRequestHandler): self.send_response(404) self.end_headers() + def _handle_config(self): + try: + from urllib.parse import urlparse, parse_qs + qs = parse_qs(urlparse(self.path).query) + if qs.get("fps"): + _cfg["fps"] = max(1, min(60, int(qs["fps"][0]))) + if qs.get("quality"): + _cfg["quality"] = max(1, min(95, int(qs["quality"][0]))) + if qs.get("scale"): + _cfg["scale"] = max(0.1, min(1.0, float(qs["scale"][0]))) + body = json.dumps(_cfg).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + except Exception: + self.send_response(500) + self.end_headers() + def log_message(self, *a): pass @@ -295,10 +321,19 @@ _counter = 0 _state = None _counter = 0 _last_tel = 0.0 -_enc_queue = None _app = None _slot_pressed = [False, False] _inject_logged = False +_pending = None +_pending_lock = threading.Lock() +_pending_evt = threading.Event() +_ws_clients = [] +_ws_clients_lock = threading.Lock() +_cfg = { + "fps": int(os.getenv("STREAM_FPS", "20")), + "quality": int(os.getenv("STREAM_QUALITY", "45")), + "scale": float(os.getenv("STREAM_SCALE", "0.5")), +} @@ -415,27 +450,53 @@ def _handle_ws(handler): except Exception: pass +def _broadcast_frame(jpeg): + dead = [] + with _ws_clients_lock: + clients = list(_ws_clients) + for c in clients: + try: + _ws_send_frame(c, 0x2, jpeg) + except Exception: + dead.append(c) + if dead: + with _ws_clients_lock: + for c in dead: + if c in _ws_clients: + _ws_clients.remove(c) + + def _encode_worker(): while True: - item = _enc_queue.get() + _pending_evt.wait() + _pending_evt.clear() + with _pending_lock: + item = _pending if item is None: - break - raw, w, h, quality = item + continue + raw, w, h = item try: + scale = float(_cfg.get("scale", 0.5)) + q = int(_cfg.get("quality", 45)) + tw = max(1, int(round(w * scale))) + th = max(1, int(round(h * scale))) img = Image.frombytes("RGBA", (w, h), raw).transpose(Image.FLIP_TOP_BOTTOM).convert("RGB") + if (tw, th) != (w, h): + img = img.resize((tw, th), Image.BILINEAR) buf = io.BytesIO() - img.save(buf, "JPEG", quality=quality) + img.save(buf, "JPEG", quality=q) + jpeg = buf.getvalue() if _state is not None: - _state.update(buf.getvalue()) + _state.update(jpeg) + _broadcast_frame(jpeg) except Exception: pass def start(port=8082): - """Start the MJPEG HTTP server + JPEG encode worker in background threads.""" - global _state, _enc_queue + """Start the HTTP server + JPEG encode worker in background threads.""" + global _state _state = StreamState() - _enc_queue = queue.Queue(maxsize=3) threading.Thread(target=_encode_worker, daemon=True).start() srv = ThreadingHTTPServer(("0.0.0.0", port), StreamHandler) srv._state = _state @@ -443,6 +504,36 @@ def start(port=8082): return _state +def _handle_ws_stream(handler): + try: + key = handler.headers.get("Sec-WebSocket-Key", "") + if not key: + return + handler.send_response(101, "Switching Protocols") + handler.send_header("Upgrade", "websocket") + handler.send_header("Connection", "Upgrade") + handler.send_header("Sec-WebSocket-Accept", _ws_accept(key)) + handler.end_headers() + handler.wfile.flush() + with _ws_clients_lock: + _ws_clients.append(handler.wfile) + try: + while True: + opcode, payload = _ws_read_frame(handler.rfile) + if opcode is None: + break + if opcode == 0x8: + break + if opcode == 0x9: + _ws_send_frame(handler.wfile, 0xA, payload) + finally: + with _ws_clients_lock: + if handler.wfile in _ws_clients: + _ws_clients.remove(handler.wfile) + except Exception: + pass + + def _write_telemetry(): global _last_tel now = time.time() @@ -498,15 +589,15 @@ def _write_telemetry(): def capture_frame(app, quality=50, target_fps=10): - global _app + """Render loop hook: snapshot newest frame for the encode worker.""" + global _app, _counter, _pending if _app is None: _app = app - """Call from the render loop: snapshot texture, queue raw for async JPEG encode.""" - global _counter if _state is None or app._render_texture is None: return _counter += 1 - skip = max(1, int(getattr(app, "_target_fps", 60)) // max(1, int(target_fps))) + fps = int(_cfg.get("fps", 20)) + skip = max(1, int(getattr(app, "_target_fps", 60)) // max(1, fps)) if _counter % skip != 0: return _write_telemetry() @@ -515,10 +606,8 @@ def capture_frame(app, quality=50, target_fps=10): w, h = si.width, si.height raw = bytes(rl.ffi.buffer(si.data, w * h * 4)) rl.unload_image(si) - if _enc_queue is not None: - try: - _enc_queue.put_nowait((raw, w, h, int(quality))) - except Exception: - pass + with _pending_lock: + _pending = (raw, w, h) + _pending_evt.set() except Exception: pass