evan/qemu-xteink
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(xteink): add a web front end for the emulator

Browse Source 
Split the emulator script into an xteink package so the CLI and the new
web UI share one QEMU driver. The web UI uploads firmware and an SD image,
boots them, drives the buttons, and long-polls the panel frame generation
so the screen updates on each e-ink refresh.
This commit is contained in:
Evan Reichard
2026-07-26 11:18:28 -04:00
parent 4cfbdfbd15
commit 0feece9036
8 changed files with 700 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.xteink.md
+3
View File
@@ -42,8 +42,11 @@ nix run .#xteink-emu -- wait --timeout 30 --match 'ready'
nix run .#xteink-emu -- button bottom-2 --hold-ms 100
nix run .#xteink-emu -- screenshot --output screen.png
nix run .#xteink-emu -- stop
nix run .#xteink-emu -- web --host 127.0.0.1 --port 8080
```
The code lives in `scripts/xteink/` (`emu.py` drives QEMU, `cli.py` is the command line, `web/` is the browser UI); `scripts/xteink-emu.py` is the entry point. `web` serves a page that uploads firmware and an SD image, boots them, drives the buttons, and streams the panel and serial log. The screen redraws off the panel's `frame-generation` counter, so it updates on each real e-ink refresh rather than on a timer.
State defaults to `/tmp/xteink-emu-$UID`; pass `--state-dir` to any subcommand for parallel instances. `interactive` opens the display and maps Left/Right, 1–4, and P to the matching device buttons. Hold P for a long power press and close the window to stop. `wait` advances a persistent log cursor after each match. The emulator exposes the open Wi-Fi network `qemu`; guest traffic uses QEMU's user-mode NAT.
### WebAssembly
flake.nix
+1 -1
View File
@@ -97,7 +97,7 @@
export XTEINK_BIOS=${qemu-native-xteink}/share/qemu
export XTEINK_BOOTLOADER=${./scripts/assets/esp32c3-bootloader.bin}
export XTEINK_PARTITIONS=${./scripts/assets/xteink-partitions.bin}
exec uv run --script ${./scripts/xteink-emu.py} "$@"
exec uv run --script ${./scripts}/xteink-emu.py "$@"
'';
};
in
scripts/xteink-emu.py
Executable
+15
View File
@@ -0,0 +1,15 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from xteink.cli import run_cli
if __name__ == "__main__":
run_cli()
scripts/xteink/__init__.py
View File
scripts/xteink/cli.py
+166
View File
@@ -0,0 +1,166 @@
"""Command line front end for the xteink emulator."""
import argparse
import sys
from pathlib import Path
from . import emu
from .web import serve
def state_dir(args):
return Path(args.state_dir).expanduser().resolve()
def run(args, interactive):
state = state_dir(args)
process = emu.launch(
state,
args.firmware,
args.sdcard,
variant=args.variant,
power_hold_ms=args.power_hold_ms,
interactive=interactive,
)
print(f"started PID {process.pid}")
print(f"state: {state}")
if not interactive:
print(f"serial log: {state / 'serial.log'}")
return
print("keys: Left/Right, 1-4 bottom buttons, P power; close the window to stop")
try:
returncode = process.wait()
except KeyboardInterrupt:
if process.poll() is None:
try:
emu.qmp_command(state, "quit")
except OSError:
process.terminate()
returncode = process.wait()
finally:
(state / "pid").unlink(missing_ok=True)
(state / "qmp.sock").unlink(missing_ok=True)
if returncode:
raise RuntimeError(f"QEMU exited with status {returncode}")
def start(args):
run(args, interactive=False)
def interactive(args):
run(args, interactive=True)
def stop(args):
print(f"stopped PID {emu.shutdown(state_dir(args))}")
def wait(args):
line = emu.wait_for_log(state_dir(args), args.match, args.timeout)
print(line, end="" if line.endswith("\n") else "\n")
def button(args):
state = state_dir(args)
emu.require_running(state)
if (args.down or args.up) and len(args.name) != 1:
raise RuntimeError("--down/--up accept exactly one button")
for name in args.name:
if args.down or args.up:
emu.set_button(state, name, down=args.down)
return
emu.press(state, name, args.samples, args.timeout, args.hold_ms)
def screenshot(args):
state = state_dir(args)
emu.require_running(state)
print(emu.screendump(state, args.output, args.settle_ms))
def web(args):
serve(state_dir(args), args.host, args.port, args.log_bytes)
def parser():
root = argparse.ArgumentParser(prog="xteink-emu")
commands = root.add_subparsers(dest="command", required=True)
def add_state(command):
command.add_argument("--state-dir", default=emu.default_state_dir())
def add_run_arguments(command):
add_state(command)
command.add_argument("--firmware", required=True)
command.add_argument("--sdcard", required=True)
command.add_argument("--variant", choices=emu.VARIANTS, default="x3")
command.add_argument("--power-hold-ms", type=int, default=2000)
command = commands.add_parser("start")
add_run_arguments(command)
command.set_defaults(handler=start)
command = commands.add_parser("interactive")
add_run_arguments(command)
command.set_defaults(handler=interactive)
command = commands.add_parser("stop")
add_state(command)
command.set_defaults(handler=stop)
command = commands.add_parser("wait")
add_state(command)
command.add_argument("--timeout", type=float, required=True, help="seconds")
command.add_argument("--match", required=True, help="regular expression")
command.set_defaults(handler=wait)
command = commands.add_parser("button")
add_state(command)
command.add_argument("name", nargs="+", choices=emu.BUTTONS, help="one or more buttons pressed in order")
command.add_argument("--hold-ms", type=int, default=500, help="hold time for the power button")
command.add_argument("--samples", type=int, default=4, help="guest ADC reads required per press and release")
command.add_argument("--timeout", type=float, default=90, help="seconds to wait for guest ADC sampling")
button_mode = command.add_mutually_exclusive_group()
button_mode.add_argument("--down", action="store_true", help="press and hold")
button_mode.add_argument("--up", action="store_true", help="release")
command.set_defaults(handler=button)
command = commands.add_parser("screenshot")
add_state(command)
command.add_argument("--output", required=True)
command.add_argument("--settle-ms", type=int, default=1500, help="wait for the panel to refresh before capturing")
command.set_defaults(handler=screenshot)
command = commands.add_parser("web")
add_state(command)
command.add_argument("--host", default="127.0.0.1")
command.add_argument("--port", type=int, default=8080)
command.add_argument("--log-bytes", type=int, default=200_000, help="tail size of the serial log")
command.set_defaults(handler=web)
return root
def main():
args = parser().parse_args()
if getattr(args, "timeout", 0) < 0:
raise RuntimeError("--timeout must not be negative")
if getattr(args, "hold_ms", 0) < 0:
raise RuntimeError("--hold-ms must not be negative")
if getattr(args, "samples", 1) < 1:
raise RuntimeError("--samples must be positive")
if getattr(args, "settle_ms", 0) < 0:
raise RuntimeError("--settle-ms must not be negative")
if getattr(args, "power_hold_ms", 0) < 0:
raise RuntimeError("--power-hold-ms must not be negative")
args.handler(args)
def run_cli():
try:
main()
except (RuntimeError, OSError) as error:
print(f"xteink-emu: {error}", file=sys.stderr)
sys.exit(1)
scripts/xteink/emu.py
+104 -162
View File
@@ -1,17 +1,11 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Core emulator control shared by the CLI and the web UI."""
import argparse
import json
import os
import re
import signal
import socket
import subprocess
import sys
import tempfile
import time
from pathlib import Path
@@ -20,7 +14,7 @@ FLASH_SIZE = 16 * 1024 * 1024
APP_OFFSET = 0x10000
PARTITIONS_OFFSET = 0x8000
RELEASED_ADC = 4095
PROJECT_ROOT = Path(__file__).resolve().parent.parent
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
DEFAULT_PATHS = {
"XTEINK_QEMU": PROJECT_ROOT / "dist/qemu-native/bin/qemu-system-riscv32",
"XTEINK_BIOS": PROJECT_ROOT / "dist/qemu-native/share/qemu",
@@ -36,16 +30,13 @@ BUTTONS = {
"bottom-4": ("/machine/adc", "adci[1]", 5, RELEASED_ADC),
"power": ("/machine/gpio", "input-level[3]", False, True),
}
VARIANTS = ("x3", "x4")
def default_state_dir():
return Path(tempfile.gettempdir()) / f"xteink-emu-{os.getuid()}"
def state_dir(args):
return Path(args.state_dir).expanduser().resolve()
def require_env(name):
path = Path(os.environ.get(name, DEFAULT_PATHS[name]))
if not path.exists():
@@ -72,6 +63,17 @@ def process_alive(pid):
return True
def is_running(state):
return process_alive(pid_from(state))
def require_running(state):
pid = pid_from(state)
if not process_alive(pid):
raise RuntimeError(f"xteink-emu is not running in {state}")
return pid
def qmp_command(state, execute, arguments=None, timeout=5):
with socket.socket(socket.AF_UNIX) as sock:
sock.settimeout(timeout)
@@ -97,13 +99,6 @@ def qmp_command(state, execute, arguments=None, timeout=5):
return send(command)
def require_running(state):
pid = pid_from(state)
if not process_alive(pid):
raise RuntimeError(f"xteink-emu is not running in {state}")
return pid
def build_flash(firmware, output):
app = firmware.read_bytes()
if len(app) == FLASH_SIZE:
@@ -123,14 +118,25 @@ def build_flash(firmware, output):
output.write_bytes(flash)
def run(args, interactive):
state = state_dir(args)
def hold_power(state, hold_ms):
path, prop, pressed, released = BUTTONS["power"]
qmp_command(state, "qom-set", {"path": path, "property": prop, "value": pressed})
try:
time.sleep(hold_ms / 1000)
finally:
qmp_command(state, "qom-set", {"path": path, "property": prop, "value": released})
def launch(state, firmware, sdcard, variant="x3", power_hold_ms=2000, interactive=False):
"""Boot QEMU in `state`, returning the process once its QMP socket answers."""
old_pid = pid_from(state)
if process_alive(old_pid):
raise RuntimeError(f"xteink-emu is already running as PID {old_pid} in {state}")
if variant not in VARIANTS:
raise RuntimeError(f"unknown variant: {variant}")
firmware = Path(args.firmware).expanduser().resolve()
sdcard = Path(args.sdcard).expanduser().resolve()
firmware = Path(firmware).expanduser().resolve()
sdcard = Path(sdcard).expanduser().resolve()
if not firmware.is_file():
raise RuntimeError(f"firmware not found: {firmware}")
if not sdcard.is_file():
@@ -150,7 +156,7 @@ def run(args, interactive):
command = [
str(require_env("XTEINK_QEMU")),
"-machine", "xteink,variant=x3",
"-machine", f"xteink,variant={variant}",
"-accel", "tcg",
"-L", str(require_env("XTEINK_BIOS")),
"-display", "gtk" if interactive else "none",
@@ -183,53 +189,16 @@ def run(args, interactive):
raise RuntimeError(f"QEMU exited during startup\n{message}")
try:
qmp_command(state, "query-status")
if args.power_hold_ms:
path, prop, pressed, released = BUTTONS["power"]
qmp_command(state, "qom-set", {"path": path, "property": prop, "value": pressed})
try:
time.sleep(args.power_hold_ms / 1000)
finally:
qmp_command(state, "qom-set", {"path": path, "property": prop, "value": released})
break
if power_hold_ms:
hold_power(state, power_hold_ms)
return process
except (FileNotFoundError, ConnectionRefusedError, socket.timeout):
time.sleep(0.05)
else:
process.terminate()
raise RuntimeError("QEMU did not open its QMP socket")
print(f"started PID {process.pid}")
print(f"state: {state}")
if not interactive:
print(f"serial log: {state / 'serial.log'}")
return
print("keys: Left/Right, 1-4 bottom buttons, P power; close the window to stop")
try:
returncode = process.wait()
except KeyboardInterrupt:
if process.poll() is None:
try:
qmp_command(state, "quit")
except OSError:
process.terminate()
returncode = process.wait()
finally:
(state / "pid").unlink(missing_ok=True)
(state / "qmp.sock").unlink(missing_ok=True)
if returncode:
raise RuntimeError(f"QEMU exited with status {returncode}")
process.terminate()
raise RuntimeError("QEMU did not open its QMP socket")
def start(args):
run(args, interactive=False)
def interactive(args):
run(args, interactive=True)
def stop(args):
state = state_dir(args)
def shutdown(state):
pid = require_running(state)
try:
qmp_command(state, "quit")
@@ -243,16 +212,15 @@ def stop(args):
os.kill(pid, signal.SIGKILL)
(state / "pid").unlink(missing_ok=True)
(state / "qmp.sock").unlink(missing_ok=True)
print(f"stopped PID {pid}")
return pid
def wait_for_log(args):
state = state_dir(args)
def wait_for_log(state, match, timeout):
"""Scan the serial log from the persistent cursor and return the first matching line."""
require_running(state)
log_path = state / "serial.log"
offset_path = state / "wait.offset"
try:
pattern = re.compile(args.match)
pattern = re.compile(match)
except re.error as error:
raise RuntimeError(f"invalid regex: {error}") from error
try:
@@ -260,111 +228,85 @@ def wait_for_log(args):
except (FileNotFoundError, ValueError):
offset = 0
deadline = time.monotonic() + args.timeout
with log_path.open("r", errors="replace") as log:
deadline = time.monotonic() + timeout
with (state / "serial.log").open("r", errors="replace") as log:
log.seek(offset)
while True:
line = log.readline()
if line:
if pattern.search(line):
offset_path.write_text(f"{log.tell()}\n")
print(line, end="" if line.endswith("\n") else "\n")
return
return line
continue
if time.monotonic() >= deadline:
raise RuntimeError(f"timed out after {args.timeout:g}s waiting for /{args.match}/")
if not process_alive(pid_from(state)):
raise RuntimeError(f"timed out after {timeout:g}s waiting for /{match}/")
if not is_running(state):
raise RuntimeError("QEMU exited while waiting for serial output")
time.sleep(0.05)
def button(args):
state = state_dir(args)
require_running(state)
path, prop, pressed, released = BUTTONS[args.name]
if args.down:
qmp_command(state, "qom-set", {"path": path, "property": prop, "value": pressed})
return
if args.up:
qmp_command(state, "qom-set", {"path": path, "property": prop, "value": released})
return
qmp_command(state, "qom-set", {"path": path, "property": prop, "value": pressed})
def frame_generation(state):
"""Panel refresh counter; changes only when the e-ink display actually redraws."""
return qmp_command(state, "qom-get", {"path": "/machine", "property": "frame-generation"})
def await_frame(state, since, timeout):
"""Block until the panel redraws, returning the new generation or None on timeout."""
deadline = time.monotonic() + timeout
while True:
current = frame_generation(state)
if current != since:
return current
if time.monotonic() >= deadline:
return None
time.sleep(0.05)
def adc_samples(state, prop):
"""Number of times the guest has read this ADC channel."""
channel = prop[prop.index("[") + 1 : prop.index("]")]
return qmp_command(state, "qom-get", {"path": "/machine/adc", "property": f"adcsamples[{channel}]"})
def await_samples(state, prop, count, timeout):
"""Block until the guest reads the channel again.
Holding a button for a fixed wall-clock time is unreliable: the firmware only polls the
resistor ladder between e-ink refreshes, which can block for seconds.
"""
start = adc_samples(state, prop)
deadline = time.monotonic() + timeout
while adc_samples(state, prop) - start < count:
if time.monotonic() >= deadline:
raise RuntimeError(f"timed out after {timeout:g}s waiting for the guest to sample {prop}")
time.sleep(0.01)
def set_button(state, name, down):
path, prop, pressed, released = BUTTONS[name]
qmp_command(state, "qom-set", {"path": path, "property": prop, "value": pressed if down else released})
def press(state, name, samples=4, timeout=90, hold_ms=500):
path, prop, _, _ = BUTTONS[name]
adc = not path.endswith("gpio")
set_button(state, name, down=True)
try:
time.sleep(args.hold_ms / 1000)
if adc:
await_samples(state, prop, samples, timeout)
else:
time.sleep(hold_ms / 1000)
finally:
qmp_command(state, "qom-set", {"path": path, "property": prop, "value": released})
set_button(state, name, down=False)
if adc:
await_samples(state, prop, samples, timeout)
def screenshot(args):
state = state_dir(args)
require_running(state)
output = Path(args.output).expanduser().resolve()
def screendump(state, output, settle_ms=0):
# The e-ink panel needs a full refresh cycle before a capture shows the latest frame.
if settle_ms:
time.sleep(settle_ms / 1000)
output = Path(output).expanduser().resolve()
output.parent.mkdir(parents=True, exist_ok=True)
qmp_command(state, "screendump", {"filename": str(output), "format": "png"}, timeout=30)
print(output)
def parser():
root = argparse.ArgumentParser(prog="xteink-emu")
commands = root.add_subparsers(dest="command", required=True)
def add_state(command):
command.add_argument("--state-dir", default=default_state_dir())
def add_run_arguments(command):
add_state(command)
command.add_argument("--firmware", required=True)
command.add_argument("--sdcard", required=True)
command.add_argument("--power-hold-ms", type=int, default=2000)
command = commands.add_parser("start")
add_run_arguments(command)
command.set_defaults(handler=start)
command = commands.add_parser("interactive")
add_run_arguments(command)
command.set_defaults(handler=interactive)
command = commands.add_parser("stop")
add_state(command)
command.set_defaults(handler=stop)
command = commands.add_parser("wait")
add_state(command)
command.add_argument("--timeout", type=float, required=True, help="seconds")
command.add_argument("--match", required=True, help="regular expression")
command.set_defaults(handler=wait_for_log)
command = commands.add_parser("button")
add_state(command)
command.add_argument("name", choices=BUTTONS)
command.add_argument("--hold-ms", type=int, default=100)
button_mode = command.add_mutually_exclusive_group()
button_mode.add_argument("--down", action="store_true", help="press and hold")
button_mode.add_argument("--up", action="store_true", help="release")
command.set_defaults(handler=button)
command = commands.add_parser("screenshot")
add_state(command)
command.add_argument("--output", required=True)
command.set_defaults(handler=screenshot)
return root
def main():
args = parser().parse_args()
if getattr(args, "timeout", 0) < 0:
raise RuntimeError("--timeout must not be negative")
if getattr(args, "hold_ms", 0) < 0:
raise RuntimeError("--hold-ms must not be negative")
if getattr(args, "power_hold_ms", 0) < 0:
raise RuntimeError("--power-hold-ms must not be negative")
args.handler(args)
if __name__ == "__main__":
try:
main()
except (RuntimeError, OSError) as error:
print(f"xteink-emu: {error}", file=sys.stderr)
sys.exit(1)
return output
scripts/xteink/web/__init__.py
+183
View File
@@ -0,0 +1,183 @@
"""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
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 == "/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"]
sdcard = state / UPLOADS["sdcard"]
if not firmware.is_file():
raise RuntimeError("upload a firmware image first")
if not sdcard.is_file():
blank_sdcard(sdcard, 64)
save_settings(state, {"sdcard": "blank 64 MiB"})
if emu.is_running(state):
emu.shutdown(state)
emu.launch(state, firmware, sdcard, 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 == "/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
scripts/xteink/web/index.html
+228
View File
@@ -0,0 +1,228 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>xteink emulator</title>
<style>
:root { color-scheme: light; font-family: Inter, ui-sans-serif, system-ui, sans-serif; color: #20211f; background: #e9e5dc; }
* { box-sizing: border-box; }
body { margin: 0; }
main { width: min(100% - 2rem, 70rem); margin: 0 auto; padding: 2.5rem 0 4rem; }
header { margin-bottom: 1.5rem; }
header p { max-width: 42rem; }
h1 { margin: 0; font-size: clamp(2rem, 6vw, 4rem); line-height: 1; }
.eyebrow { margin-bottom: .5rem; font-size: .75rem; font-weight: 800; letter-spacing: .16em; text-transform: uppercase; }
.boot-panel { display: flex; flex-wrap: wrap; gap: 1rem; align-items: end; padding: 1rem; border: 1px solid #c8c2b6; border-radius: .75rem; background: #f7f4ed; }
label { display: grid; gap: .4rem; font-size: .85rem; font-weight: 700; }
input, select, button { font: inherit; }
input, select { min-height: 2.6rem; padding: .5rem; border: 1px solid #aaa398; border-radius: .4rem; background: white; }
button { min-height: 2.75rem; padding: .55rem 1rem; border: 1px solid #55534e; border-radius: .5rem; color: #fff; background: #343532; font-weight: 750; cursor: pointer; touch-action: none; }
button:hover:not(:disabled) { background: #11120f; }
button:focus-visible { outline: 3px solid #d2752b; outline-offset: 2px; }
button:disabled { cursor: not-allowed; opacity: .4; }
button.active { transform: translateY(2px); background: #d2752b; }
.danger { border-color: #b94a3b; color: #a52d20; background: transparent; }
.danger:hover:not(:disabled) { color: #fff; background: #a52d20; }
.status { min-height: 1.5rem; margin: 1rem 0; }
.status.error { color: #a52d20; font-weight: 700; }
.workspace { display: grid; grid-template-columns: 1fr 1fr; gap: clamp(1.5rem, 4vw, 3rem); align-items: start; }
.device { display: flex; justify-content: center; }
.device-frame { display: grid; grid-template-columns: auto minmax(14rem, 30rem) auto; grid-template-rows: auto auto auto; gap: .6rem; }
.screen-shell { grid-column: 2; grid-row: 2; padding: 1.1rem; border-radius: 1.2rem; background: #30312e; box-shadow: 0 1rem 2.5rem #5d584c40; }
img { display: block; width: 100%; height: auto; aspect-ratio: 528 / 792; background: #fff; image-rendering: pixelated; }
.device-frame button { min-height: 0; border-radius: .6rem; font-size: .75rem; }
.btn-power { grid-column: 2; grid-row: 1; height: 1.8rem; }
.btn-side { grid-row: 2; width: 2rem; writing-mode: vertical-rl; }
.btn-left { grid-column: 1; }
.btn-right { grid-column: 3; }
.btn-bottom { grid-column: 2; grid-row: 3; display: grid; grid-template-columns: repeat(4, 1fr); gap: .6rem; }
.btn-bottom button { height: 1.8rem; }
.inspector { min-width: 0; overflow: hidden; padding: 1rem; border: 1px solid #c8c2b6; border-radius: .75rem; background: #f7f4ed; }
.log-header { display: flex; justify-content: space-between; align-items: center; margin-bottom: .5rem; font-size: .85rem; color: #625e55; }
.log-header label { display: flex; gap: .3rem; align-items: center; font-weight: 400; }
.log { margin: 0; height: 38rem; overflow: auto; padding: .75rem 1rem; border-radius: .6rem; background: #1c1d1a; color: #d6d0c2; font: .78rem/1.4 ui-monospace, monospace; white-space: pre-wrap; word-break: break-word; }
@media (max-width: 48rem) { main { padding-top: 1.5rem; } .workspace { grid-template-columns: 1fr; } }
</style>
</head>
<body>
<main>
<header>
<p class="eyebrow">ESP32-C3 emulator</p>
<h1>xteink X3 / X4</h1>
<p>Upload a firmware <code>.bin</code> (release app image or a full 16 MB flash) and boot it in the emulator running on this host. The SD card image persists between boots.</p>
</header>
<form id="boot-form" class="boot-panel">
<label>Firmware (.bin)
<input id="firmware" type="file" accept=".bin,application/octet-stream">
</label>
<label>SD card image (optional)
<input id="sdcard" type="file" accept=".img,.bin,application/octet-stream">
</label>
<label>Device
<select id="variant"><option value="x3">X3</option><option value="x4">X4 (display stub)</option></select>
</label>
<button id="boot" type="submit">Boot firmware</button>
<button id="stop" type="button" class="danger">Stop</button>
</form>
<p id="status" class="status" role="status">Loading.</p>
<div class="workspace">
<section class="device" aria-label="Emulated xteink device">
<div class="device-frame">
<button type="button" class="btn-power" data-button="power" data-hold-ms="2000">power</button>
<button type="button" class="btn-side btn-left" data-button="left">left</button>
<div class="screen-shell"><img id="screen" alt="E-ink display"></div>
<button type="button" class="btn-side btn-right" data-button="right">right</button>
<div class="btn-bottom">
<button type="button" data-button="bottom-1">1</button>
<button type="button" data-button="bottom-2">2</button>
<button type="button" data-button="bottom-3">3</button>
<button type="button" data-button="bottom-4">4</button>
</div>
</div>
</section>
<section class="inspector" aria-label="Serial log">
<div class="log-header">
<span>Serial log</span>
<label><input id="follow" type="checkbox" checked> follow</label>
</div>
<pre id="log" class="log" role="log" aria-live="polite"></pre>
</section>
</div>
</main>
<script>
const $ = (id) => document.getElementById(id);
const status = $("status"), screen = $("screen"), log = $("log"), follow = $("follow");
const deviceButtons = [...document.querySelectorAll("[data-button]")];
let running = false;
function report(message, error) {
status.textContent = message;
status.classList.toggle("error", Boolean(error));
}
async function call(path, options) {
const response = await fetch(path, options);
const text = await response.text();
if (!response.ok) throw new Error(text || response.statusText);
return text;
}
function applyStatus(payload) {
// A fresh QEMU process restarts its frame counter, so drop the cursor across boots.
if (payload.running !== running) generation = null;
running = payload.running;
for (const button of deviceButtons) button.disabled = !running;
$("stop").disabled = !running;
$("variant").value = payload.variant;
const parts = [running ? "Running." : "Stopped."];
if (payload.firmware) parts.push(`firmware: ${payload.firmware}`);
if (payload.sdcard) parts.push(`sd: ${payload.sdcard}`);
report(parts.join(" "), false);
if (running) watchScreen(); else screen.removeAttribute("src");
}
// Long Poll - /screen.png blocks until the panel redraws, so one outstanding request keeps
// the image as fresh as the e-ink hardware itself, with no polling interval to tune.
let watching = false;
let generation = null;
async function watchScreen() {
if (watching) return;
watching = true;
try {
while (running) {
const url = "screen.png" + (generation === null ? "" : "?since=" + generation);
const response = await fetch(url);
if (response.status === 304) continue;
if (!response.ok) throw new Error(await response.text());
generation = response.headers.get("X-Frame-Generation");
const blob = await response.blob();
URL.revokeObjectURL(screen.src);
screen.src = URL.createObjectURL(blob);
}
} catch (error) {
report(String(error), true);
} finally {
watching = false;
}
}
async function refreshStatus() {
try {
applyStatus(JSON.parse(await call("status")));
} catch (error) {
report(String(error), true);
}
}
async function upload(name, input) {
if (!input.files.length) return;
const file = input.files[0];
report(`Uploading ${file.name}...`, false);
await call(`upload?name=${name}&filename=${encodeURIComponent(file.name)}`, {
method: "POST",
body: await file.arrayBuffer(),
});
}
$("boot-form").onsubmit = async (event) => {
event.preventDefault();
$("boot").disabled = true;
try {
await upload("firmware", $("firmware"));
await upload("sdcard", $("sdcard"));
report("Booting...", false);
applyStatus(JSON.parse(await call(`boot?variant=${$("variant").value}`, { method: "POST" })));
} catch (error) {
report(String(error), true);
} finally {
$("boot").disabled = false;
}
};
$("stop").onclick = async () => {
try {
applyStatus(JSON.parse(await call("stop", { method: "POST" })));
} catch (error) {
report(String(error), true);
}
};
for (const button of deviceButtons) {
button.onclick = async () => {
button.classList.add("active");
button.disabled = true;
const hold = button.dataset.holdMs ? `&hold-ms=${button.dataset.holdMs}` : "";
try {
await call(`button?name=${button.dataset.button}${hold}`, { method: "POST" });
} catch (error) {
report(String(error), true);
} finally {
button.classList.remove("active");
button.disabled = !running;
}
};
}
async function refreshLog() {
try {
log.textContent = await call("log");
if (follow.checked) log.scrollTop = log.scrollHeight;
} catch (error) {
report(String(error), true);
}
}
setInterval(refreshLog, 1000);
setInterval(refreshStatus, 5000);
refreshStatus();
refreshLog();
</script>
</body>
</html>