"""Lightweight Pi-side UI for testing the production-mode + LiveKit flow. Run on the Pi: /opt/robopark-pi-client/.venv/bin/python /opt/robopark-pi-client/pi_ui.py \ --scheduler-url http://192.168.1.241:8080 \ --port 8081 Then open http://192.168.1.159:8081/ in any browser on the same LAN. What it shows / does: - Live status of this device as seen by the scheduler (online/offline, last heartbeat) - Current scheduler settings (production_mode, LiveKit configured) - Big "Join Conversation" button that asks the scheduler for a LiveKit token and opens https://meet.livekit.io/custom?livekitUrl=...&token=... in a new tab - "Send motor command" widget that fires a JSON command into the MotorBridge (same wire format as the LiveKit data channel) - "Tail logs" pane that shows the last lines of the robopark-pi-client service """ import argparse import json import logging import os import subprocess from pathlib import Path from typing import Optional import httpx from flask import Flask, jsonify, render_template_string, request logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s") log = logging.getLogger("robopark-pi.ui") app = Flask(__name__) # Lazy-loaded MotorBridge import so the UI can start even if the SDK is missing. _BRIDGE = None def _bridge(): global _BRIDGE if _BRIDGE is None: try: from motor_bridge import MotorBridge _BRIDGE = MotorBridge( motor_server_url=os.environ.get("MOTOR_SERVER_URL", "http://127.0.0.1:8001"), dry_run=os.environ.get("DRY_RUN_MOTORS", "0") == "1", ) except Exception as e: log.warning(f"MotorBridge unavailable: {e}") _BRIDGE = None return _BRIDGE STATE_FILE = Path(os.environ.get("STATE_FILE", "/etc/robopark/device.json")) def _load_device() -> dict: if not STATE_FILE.exists(): return {} try: return json.loads(STATE_FILE.read_text()) except Exception as e: log.warning(f"could not read {STATE_FILE}: {e}") return {} def _read_service_log(lines: int = 60) -> str: try: out = subprocess.run( ["journalctl", "-u", "robopark-pi-client", "-n", str(lines), "--no-pager", "-o", "cat"], capture_output=True, text=True, timeout=5, ) return out.stdout except Exception as e: return f"(could not read journal: {e})" HTML = r""" RoboPark Pi Console

🤖 RoboPark Pi Console

ceopi1

This device

Device ID—

Status…

Last heartbeat—

Motor server—

Scheduler

URL—

Production mode…

LiveKit…

LiveKit — join conversation

Asks the scheduler for a short-lived token, then opens meet.livekit.io in a new tab. You can publish mic + cam and hear the agent; the Pi's audio hardware is forwarded by the main client.

Room Identity

Motor test

Fires the same JSON payload the LiveKit data channel would send into the local motor bridge. No wiring needed.

Motor Seconds

Service log (last 60 lines)

(loading…)

""" @app.route("/") def index(): return render_template_string(HTML) @app.route("/api/status") def api_status(): dev = _load_device() out = { "scheduler_url": app.config["SCHEDULER_URL"], "device_id": dev.get("device_id"), "status": None, "last_heartbeat": None, "motor_server_url": dev.get("motor_server_url"), "production_mode": None, "livekit_configured": False, "livekit_url": None, } try: with httpx.Client(timeout=5.0) as c: r = c.get(f"{app.config['SCHEDULER_URL']}/api/settings") if r.status_code == 200: s = r.json() out["production_mode"] = bool(s.get("production_mode")) r2 = c.get(f"{app.config['SCHEDULER_URL']}/api/livekit/config") if r2.status_code == 200: lk = r2.json() out["livekit_configured"] = bool(lk.get("url") and lk.get("has_secret")) out["livekit_url"] = lk.get("url") if dev.get("device_id"): r3 = c.get(f"{app.config['SCHEDULER_URL']}/api/devices/{dev['device_id']}") if r3.status_code == 200: d = r3.json() out["status"] = d.get("status") out["last_heartbeat"] = d.get("last_heartbeat") except Exception as e: log.warning(f"status fetch failed: {e}") return jsonify(out) @app.route("/api/production_mode", methods=["PUT"]) def api_prod_mode(): payload = request.get_json(force=True) or {} new = bool(payload.get("production_mode", False)) with httpx.Client(timeout=5.0) as c: r = c.put(f"{app.config['SCHEDULER_URL']}/api/settings", json={"production_mode": new}) r.raise_for_status() return jsonify({"status": "ok", "production_mode": new}) @app.route("/api/join", methods=["POST"]) def api_join(): payload = request.get_json(force=True) or {} room = payload.get("room") or "robopark-pi-test" identity = payload.get("identity") or ("pi-ui-" + os.urandom(3).hex()) with httpx.Client(timeout=5.0) as c: r = c.post(f"{app.config['SCHEDULER_URL']}/api/livekit/token", json={"room": room, "identity": identity, "name": "Pi UI", "ttl_seconds": 3600}) if r.status_code != 200: return jsonify({"error": r.text}), r.status_code return jsonify(r.json()) @app.route("/api/motor", methods=["POST"]) def api_motor(): payload = request.get_json(force=True) or {} bridge = _bridge() if bridge is None: return jsonify({"error": "motor bridge not loaded (import failed)"}), 503 import asyncio result = asyncio.run(bridge.handle_command(json.dumps(payload))) return jsonify({"result": result}) @app.route("/api/log") def api_log(): return jsonify({"lines": _read_service_log(60)}) def parse_args(): p = argparse.ArgumentParser() p.add_argument("--scheduler-url", required=True) p.add_argument("--host", default="0.0.0.0") p.add_argument("--port", type=int, default=8081) p.add_argument("--debug", action="store_true") return p.parse_args() def main(): args = parse_args() app.config["SCHEDULER_URL"] = args.scheduler_url.rstrip("/") log.info(f"Pi UI on http://{args.host}:{args.port}/ -> scheduler {app.config['SCHEDULER_URL']}") app.run(host=args.host, port=args.port, debug=args.debug, threaded=True) if __name__ == "__main__": main()