"""Browser front end: boot firmware, drive the buttons, watch the screen and serial log.""" import http.server import json import shutil import subprocess import urllib.parse from pathlib import Path from .. import emu from . import sdcard PAGE = Path(__file__).resolve().parent / "index.html" MAX_UPLOAD = 512 * 1024 * 1024 LONG_POLL_SECONDS = 25 UPLOADS = {"firmware": "firmware.bin", "sdcard": "sdcard.img"} def settings_path(state): return state / "web.json" def settings(state): try: return json.loads(settings_path(state).read_text()) except (FileNotFoundError, ValueError): return {} def save_settings(state, values): settings_path(state).write_text(json.dumps(settings(state) | values)) def blank_sdcard(path, size_mb): """Create a FAT32 card so the UI works without the user supplying an image.""" mkfs = shutil.which("mkfs.fat") if not mkfs: raise RuntimeError("mkfs.fat not found; upload an SD card image instead") with path.open("wb") as image: image.truncate(size_mb * 1024 * 1024) subprocess.run([mkfs, "-F", "32", str(path)], check=True, capture_output=True) def serve(state, host, port, log_bytes): state.mkdir(parents=True, exist_ok=True, mode=0o700) capture = state / "web-screen.png" class Handler(http.server.BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def reply(self, body, content_type="text/plain; charset=utf-8", status=200, headers=()): if isinstance(body, str): body = body.encode() self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") for name, value in headers: self.send_header(name, value) self.end_headers() self.wfile.write(body) def reply_json(self, value, status=200): self.reply(json.dumps(value), "application/json", status) def query(self): return urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query) def body(self): length = int(self.headers.get("Content-Length", 0)) if length > MAX_UPLOAD: raise RuntimeError(f"upload larger than {MAX_UPLOAD // 1024 // 1024} MiB") return self.rfile.read(length) def status_payload(self): stored = settings(state) return { "running": emu.is_running(state), "variant": stored.get("variant", "x3"), "firmware": stored.get("firmware"), "sdcard": stored.get("sdcard"), "buttons": list(emu.BUTTONS), "variants": list(emu.VARIANTS), } def do_GET(self): route = urllib.parse.urlparse(self.path).path query = self.query() try: if route == "/": return self.reply(PAGE.read_bytes(), "text/html; charset=utf-8") if route == "/status": return self.reply_json(self.status_payload()) if route == "/screen.png": emu.require_running(state) # Long Poll - The panel only redraws on a real refresh, so block until the # guest bumps the frame counter instead of re-capturing on a timer. since = query.get("since", [None])[0] generation = emu.frame_generation(state) if since is not None and str(generation) == since: generation = emu.await_frame(state, generation, LONG_POLL_SECONDS) if generation is None: return self.reply(b"", status=304) emu.screendump(state, capture) return self.reply( capture.read_bytes(), "image/png", headers=[("X-Frame-Generation", str(generation))], ) if route == "/files": path = query.get("path", ["/"])[0] return self.reply_json(sdcard.listing(state / UPLOADS["sdcard"], path)) if route == "/files/read": path = query.get("path", [""])[0] data = sdcard.read(state / UPLOADS["sdcard"], path) try: text = data.decode() except UnicodeDecodeError: return self.reply_json({"binary": True, "size": len(data)}) return self.reply_json({"binary": False, "text": text}) if route == "/log": # Tail only; a long-lived emulator log outgrows what the page can hold. return self.reply((state / "serial.log").read_bytes()[-log_bytes:]) except (RuntimeError, OSError) as error: return self.reply(str(error), status=500) self.send_error(404) def do_POST(self): route = urllib.parse.urlparse(self.path).path query = self.query() try: if route == "/upload": name = query.get("name", [""])[0] if name not in UPLOADS: return self.send_error(404) data = self.body() if not data: raise RuntimeError(f"empty {name} upload") (state / UPLOADS[name]).write_bytes(data) filename = query.get("filename", [name])[0] save_settings(state, {name: filename}) return self.reply_json(self.status_payload()) if route == "/sdcard/blank": size_mb = max(8, min(int(query.get("size-mb", ["64"])[0]), 4096)) blank_sdcard(state / UPLOADS["sdcard"], size_mb) save_settings(state, {"sdcard": f"blank {size_mb} MiB"}) return self.reply_json(self.status_payload()) if route == "/boot": variant = query.get("variant", ["x3"])[0] hold = int(query.get("power-hold-ms", ["2000"])[0]) firmware = state / UPLOADS["firmware"] card = state / UPLOADS["sdcard"] if not firmware.is_file(): raise RuntimeError("upload a firmware image first") if not card.is_file(): blank_sdcard(card, 64) save_settings(state, {"sdcard": "blank 64 MiB"}) if emu.is_running(state): emu.shutdown(state) emu.launch(state, firmware, card, variant=variant, power_hold_ms=hold) save_settings(state, {"variant": variant}) return self.reply_json(self.status_payload()) if route == "/stop": if emu.is_running(state): emu.shutdown(state) return self.reply_json(self.status_payload()) if route.startswith("/files/"): # Editing detaches the card from a live guest, mirroring "apply and reboot". if emu.is_running(state): emu.shutdown(state) image = state / UPLOADS["sdcard"] path = query.get("path", [""])[0] if route == "/files/write": sdcard.write(image, path, self.body()) elif route == "/files/mkdir": sdcard.mkdir(image, path) elif route == "/files/delete": sdcard.delete(image, path, query.get("directory", [""])[0] == "true") elif route == "/files/clear": size_mb = max(8, image.stat().st_size // 1024 // 1024) if image.is_file() else 64 blank_sdcard(image, size_mb) save_settings(state, {"sdcard": f"blank {size_mb} MiB"}) else: return self.send_error(404) return self.reply_json(self.status_payload()) if route == "/button": name = query.get("name", [""])[0] if name not in emu.BUTTONS: return self.send_error(404) emu.require_running(state) edge = query.get("edge", [""])[0] if edge in ("down", "up"): emu.set_button(state, name, down=edge == "down") else: emu.press(state, name, hold_ms=int(query.get("hold-ms", ["500"])[0])) return self.reply("ok") except (RuntimeError, OSError, ValueError, subprocess.CalledProcessError) as error: return self.reply(str(error), status=500) self.send_error(404) def log_message(self, *_): pass server = http.server.ThreadingHTTPServer((host, port), Handler) print(f"http://{host}:{server.server_address[1]}") print(f"state: {state}") try: server.serve_forever() except KeyboardInterrupt: pass