#!/usr/bin/env python3 """Web remote control for the UV-K5 emulator. Serves the LCD as a live stream and maps on-screen buttons to the keypad model, so the emulated radio can be driven from a browser. Start the emulator first (tools/run.sh), then: python3 tools/webui.py --frame-addr 0x200013DC --status-addr 0x2000175C Then open http://127.0.0.1:8080/ Two things worth knowing: * The QMP socket takes a single client, so tools/key.py cannot talk to the same emulator while this server is running. * There is no authentication. It binds loopback and anyone who reaches the port has full control of the emulated radio. Do not expose it. """ import argparse import json import os import time from flask import Flask, Response, jsonify, request from uvk5_keys import KEYS, is_valid, normalise from uvk5_logs import LogBuffer from uvk5_stream import FramePump KEYPAD_PATH = "/machine/keypad" # Firmware thresholds, from App/misc.c: # key_debounce_10ms = 2 -> 20 ms to register a press # key_repeat_delay_10ms = 40 -> 400 ms counts as HELD, a different event # # This is a property of the firmware, not a tunable. # # Past this point the firmware also auto-repeats, every key_repeat_10ms = 80 ms. # So a 500 ms press moving a menu cursor several steps is correct, not a bug: it is # what a real radio does when you hold the button that long. Measured: 500 ms moved # the cursor 3 steps, 800 ms moved 7, 1500 ms moved 15 -- all consistent with # (duration - 400) / 80. Do not try to suppress it in the UI; the fix for an # accidental repeat is a shorter press, not a filter that hides held events. FIRMWARE_HELD_MS = 400 # The browser sends the duration it wants and the server holds the key for exactly # that long. It must not be reproduced by sending `down` and `up` as two requests: # over a slow link the round trip between them *becomes* the press duration. # Measured against this server at 400 ms RTT, an intended tap arrived as a 407 ms # hold, so every short press was dispatched as a held key and handlers like # MAIN_Key_MENU did nothing. Jitter either side of the threshold is what made it # look intermittent rather than simply broken. # # Default when a request omits hold_ms, i.e. for scripts and curl. The browser # always sends a measured duration, so this does not apply to normal use. Kept # short because the request does not return until the hold finishes, making the # value latency the caller pays directly. See MIN_HOLD_MS for why 60 ms. TAP_MS = 60 # A hold longer than this is a stuck key or a typo, not intent. MAX_HOLD_MS = 5000 # Floor for a measured press. A very fast click can measure under the debounce # window, where the firmware would not register it at all. # # Measured, and the sample size mattered: at 12 trials per value, 20 ms registered # only 5/12 while 30 ms was 12/12. The nominal 20 ms debounce is not enough on its # own because KEYBOARD_Poll samples each column 8 times wanting 2 matching reads. # 60 ms is double the proven floor. An earlier 4-trial sweep called 30 ms reliable # and would have shipped a flaky value. MIN_HOLD_MS = 60 # Lines kept in the browser's log pane. The pane is a fixed-height scroll box, so # older lines move up out of view; this caps the DOM behind it, which would # otherwise grow all session even though only a screenful is visible. MAX_LOG_LINES = 500 BOUNDARY = "uvk5frame" TARGET_FPS = 15 # Resend the current frame at least this often even when nothing changed. # # Sending only on change saves bandwidth but makes a static screen # indistinguishable from a dead connection, and a client that joined mid-idle # would sit blank until something moved. Slow rather than stopped. IDLE_FRAME_INTERVAL_S = 2.0 # Physical layout of the UV-K5 keypad, for the on-screen grid. KEY_GRID = [ ["MENU", "UP", "DOWN", "EXIT"], ["1", "2", "3", "STAR"], ["4", "5", "6", "0"], ["7", "8", "9", "F"], ] SIDE_KEYS = ["SIDE1", "SIDE2"] # Keyboard shortcuts -> radio keys. KEY_BINDINGS = { "ArrowUp": "UP", "ArrowDown": "DOWN", "Enter": "MENU", "Escape": "EXIT", "KeyF": "F", "KeyM": "MENU", "BracketLeft": "SIDE1", "BracketRight": "SIDE2", "Digit0": "0", "Digit1": "1", "Digit2": "2", "Digit3": "3", "Digit4": "4", "Digit5": "5", "Digit6": "6", "Digit7": "7", "Digit8": "8", "Digit9": "9", } POWER_ACTIONS = ("on", "off", "reset", "pause", "resume") def create_app(client, frame_addr: int, status_addr: int, scale: int = 4, supervisor=None, log=None): app = Flask(__name__) if log is None: log = LogBuffer() app.config["LOG"] = log # One background grabber for every client. client may be None: the emulator # can be powered off, and the page still has to load. pump = FramePump(client, frame_addr, status_addr, fps=TARGET_FPS, scale=scale) pump.start() app.config["PUMP"] = pump app.config["SUPERVISOR"] = supervisor def client_ip(): """The address of whoever made this request. Behind the nginx reverse proxy REMOTE_ADDR is always 127.0.0.1, so the first hop of X-Forwarded-For is what identifies the real client. Only the first entry is trusted: the rest of the chain can be set by the caller. """ forwarded = request.headers.get("X-Forwarded-For", "") if forwarded: first = forwarded.split(",")[0].strip() if first: return first return request.remote_addr def active_client(): """The live QMP client, or None when the emulator is off. The supervisor is authoritative once present: it replaces the client on every power cycle, so the one captured at create_app time goes stale. """ if supervisor is not None: return supervisor.client() return client def set_press(value: str): target = active_client() if target is None: raise LookupError("emulator is off") target.command("qom-set", path=KEYPAD_PATH, property="press", value=value) @app.get("/") def index(): return Response(render_index(scale), mimetype="text/html") @app.get("/api/status") def api_status(): target = active_client() if target is None: return jsonify(powered=False, status="off") try: info = target.command("query-status") except Exception as exc: # The emulator can die under us; that is a state to report, not a 500. return jsonify(powered=False, status="unreachable", error=str(exc)) return jsonify(powered=True, **info) @app.get("/api/logs") def api_logs(): since = request.args.get("since", type=int, default=0) return jsonify(entries=log.entries(since=since), cursor=log.cursor()) @app.post("/api/power/") def api_power(action): action = (action or "").strip().lower() if action not in POWER_ACTIONS: return jsonify(error=f"unknown action {action!r}", valid=list(POWER_ACTIONS)), 400 if supervisor is None: return jsonify( error="power control needs a supervisor; this server was " "started without one"), 409 # Refuse to kill a process we did not start. Reset is fine either way, # since system_reset does not end anything. if action == "off" and not supervisor.owns_process(): return jsonify( error="this server attached to an emulator it did not start, " "so it will not stop it. Restart without --attach to " "manage the process here."), 409 # Attribute the action here: the supervisor has no request context, and on # a shared log "who powered it off" is the useful part. log.add("power", f"{action} requested", ip=client_ip()) {"on": supervisor.power_on, "off": supervisor.power_off, "reset": supervisor.reset, "pause": supervisor.pause, "resume": supervisor.resume}[action]() # Point the pump at whatever client is live now. rebind(None) blanks the # screen, so power off actually goes dark instead of freezing on the last # frame. pump.rebind(supervisor.client()) return jsonify(ok=True, action=action, powered=supervisor.is_running()) @app.post("/api/key") def api_key(): body = request.get_json(silent=True) or {} key = normalise(body.get("key", "")) action = (body.get("action") or "tap").strip().lower() if not is_valid(key): # Log refusals too: a silently dropped key is indistinguishable from # a dead button in the browser. log.add("key", f"{body.get('key')!r} rejected: not a key on this model", ip=client_ip()) return jsonify(error=f"unknown key {body.get('key')!r}", valid=list(KEYS)), 400 if action not in ("down", "up", "tap"): return jsonify(error=f"unknown action {action!r}", valid=["down", "up", "tap"]), 400 hold_raw = body.get("hold_ms") if hold_raw is None: hold_ms = TAP_MS else: try: hold_ms = int(hold_raw) except (TypeError, ValueError): return jsonify( error=f"hold_ms must be a number, got {hold_raw!r}"), 400 if hold_ms < 0: return jsonify(error="hold_ms must not be negative"), 400 hold_ms = min(hold_ms, MAX_HOLD_MS) try: if action == "down": log.add("key", f"{key} down", ip=client_ip()) set_press(key) elif action == "up": log.add("key", f"{key} up", ip=client_ip()) set_press("") else: # Label by what the FIRMWARE will conclude, so the log says what # the radio saw. That boundary is 400 ms (key_repeat_delay_10ms), # not the UI's hold threshold -- conflating the two is what caused # the tap+held double send in the first place. kind = "held" if hold_ms >= FIRMWARE_HELD_MS else "tap" log.add("key", f"{key} {kind} {hold_ms}ms", ip=client_ip()) # Hold here, locally. See the note on TAP_MS: doing this as two # requests puts the network round trip inside the press duration. set_press(key) time.sleep(hold_ms / 1000) set_press("") except LookupError: log.add("key", f"{key} ignored: emulator is off", ip=client_ip()) return jsonify(error="emulator is off; press On first"), 409 return jsonify(ok=True, key=key, action=action, hold_ms=hold_ms) @app.post("/api/release-all") def api_release_all(): """Safety valve: an empty press clears every key in the model.""" try: set_press("") except LookupError: return jsonify(error="emulator is off"), 409 return jsonify(ok=True) def wait_for_frame(timeout: float = 2.0): deadline = time.monotonic() + timeout while time.monotonic() < deadline: png = pump.latest() if png is not None: return png time.sleep(0.02) return None @app.get("/frame.png") def frame_png(): png = wait_for_frame() if png is None: return jsonify(error="no frame available; is the emulator on?"), 503 return Response(png, mimetype="image/png", headers={"Cache-Control": "no-store"}) @app.get("/stream") def stream(): # limit exists for tests; unset means stream until the client leaves. limit = request.args.get("limit", type=int) interval = 1.0 / TARGET_FPS def frames(): sent, seen, last_sent_at = 0, -1, 0.0 while limit is None or sent < limit: png = pump.latest() generation = pump.generation() now = time.monotonic() # Send on change, and otherwise at the idle keepalive rate. Change # detection alone leaves a static screen looking like a dead # connection, and a client joining mid-idle would stay blank. stale = now - last_sent_at >= IDLE_FRAME_INTERVAL_S if png is not None and (generation != seen or stale or limit is not None): seen = generation last_sent_at = now yield (b"--" + BOUNDARY.encode() + b"\r\n" b"Content-Type: image/png\r\n" b"Content-Length: " + str(len(png)).encode() + b"\r\n\r\n" + png + b"\r\n") sent += 1 else: time.sleep(interval) return Response(frames(), mimetype=f"multipart/x-mixed-replace; boundary={BOUNDARY}", headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}) return app def render_index(scale: int) -> str: grid = "\n".join( "
" + "".join( f'' for k in row) + "
" for row in KEY_GRID ) sides = "".join( f'' for k in SIDE_KEYS ) return f""" UV-K5 remote
-
radio LCD
{sides}
{grid}
connecting...
Logs (firmware serial, qemu, power)

  

How long you hold a key is measured here and sent as one number, so the firmware sees exactly the press you made. Hold past 400 ms for a long press, which the firmware treats as a separate event and which repeats. Arrows move, Enter is MENU, Esc is EXIT, digits map straight through. No PTT button -- the keypad model has no PTT line.

""" def main() -> int: ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--qmp", default="/tmp/uvk5-qmp.sock") ap.add_argument("--frame-addr", type=lambda v: int(v, 0), required=True, help="address of gFrameBuffer (moves between builds)") ap.add_argument("--status-addr", type=lambda v: int(v, 0), required=True, help="address of gStatusLine") ap.add_argument("--host", default="127.0.0.1") ap.add_argument("--port", type=int, default=8080) ap.add_argument("--scale", type=int, default=4) ap.add_argument("--attach", action="store_true", help="attach to an emulator started elsewhere (run.sh) " "instead of managing one. Off is then refused, since " "this server did not start that process.") ap.add_argument("--qemu", default=os.path.expanduser( "~/qemu-build/qemu-7.2+dfsg/build/qemu-system-arm")) ap.add_argument("--elf", default=os.path.expanduser( "~/uvk5-port/uvk5-sat/build/CW/nr7y.cw.elf")) ap.add_argument("--flash", default=os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "assets", "flash.img")) ap.add_argument("--gdb-port", type=int, default=1234) args = ap.parse_args() from uvk5_qmp import QmpClient from uvk5_supervisor import Supervisor, default_launcher, wait_for_socket def connect(): if not wait_for_socket(args.qmp, timeout=15): raise RuntimeError(f"QMP socket never appeared at {args.qmp}") return QmpClient(args.qmp) # One buffer shared by the supervisor and the HTTP layer, so power events, # QEMU stderr and firmware serial all land in the same place. log = LogBuffer() supervisor = Supervisor( launch=default_launcher(args.qemu, args.flash, args.elf, args.qmp, gdb_port=args.gdb_port), connect=connect, log=log) if args.attach: # Someone else owns the process; adopt it so the screen works, but Off # will refuse. supervisor.adopt(connect()) # Otherwise the emulator stays OFF on purpose. The user presses On, so the # page behaves like walking up to a machine rather than finding it booted. app = create_app(supervisor.client(), args.frame_addr, args.status_addr, args.scale, supervisor=supervisor, log=log) print(f"serving on http://{args.host}:{args.port}/") print("attached to a running emulator" if args.attach else "emulator is OFF; press On in the browser to boot it") print("no authentication: anyone who can reach this port controls the radio") app.run(host=args.host, port=args.port, threaded=True) return 0 if __name__ == "__main__": raise SystemExit(main())