WuYongJie/ESP32
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
ESP32/tools/preview-bridge/server.py
Codex 035f1e8cea Initial ESP32 RLCD metrics workspace
2026-06-23 20:38:23 +08:00

352 lines
13 KiB
Python
Raw Blame History

from __future__ import annotations
import json
import os
import re
import subprocess
import sys
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import unquote, urlparse
PROJECT_ROOT = Path(__file__).resolve().parents[2]
FRONTEND_DIR = PROJECT_ROOT / "frontend-preview"
MAIN_DIR = PROJECT_ROOT / "main"
GENERATED_DIR = MAIN_DIR / "generated"
ACTIVE_SCREEN_PATH = FRONTEND_DIR / "active-screen.json"
TEMPLATES_PATH = FRONTEND_DIR / "templates.json"
FIRMWARES_PATH = FRONTEND_DIR / "firmwares.json"
UI_CONFIG_PATH = GENERATED_DIR / "ui_config.h"
DEFAULT_PORT = 8788
IDF_VERSION = "v6.0.1"
EIM_EXE = Path(
r"C:\Users\54972\AppData\Local\Microsoft\WinGet\Packages"
r"\Espressif.EIM-CLI_Microsoft.Winget.Source_8wekyb3d8bbwe\eim.exe"
)
TEMPLATE_IDS = {
"clock": "UI_TEMPLATE_CLOCK",
"dual-metric": "UI_TEMPLATE_DUAL_METRIC",
"meter": "UI_TEMPLATE_METER",
}
def read_json(path: Path, fallback: Any) -> Any:
try:
return json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
return fallback
def write_json(path: Path, data: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def response_bytes(data: Any) -> bytes:
return json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")
def c_string(value: Any, limit: int = 48) -> str:
text = str(value or "")[:limit]
text = text.replace("\\", "\\\\").replace('"', '\\"')
text = text.replace("\r", " ").replace("\n", " ")
return f'"{text}"'
def c_int(value: Any, default: int = 0, minimum: int = -999, maximum: int = 999) -> int:
try:
number = int(float(str(value).strip()))
except (TypeError, ValueError):
number = default
return max(minimum, min(maximum, number))
def normalize_screen(screen: dict[str, Any]) -> dict[str, Any]:
screen = dict(screen or {})
fields = dict(screen.get("fields") or {})
effects = dict(screen.get("effects") or {})
template_id = screen.get("templateId") or "dual-metric"
if template_id not in TEMPLATE_IDS:
template_id = "dual-metric"
orientation = screen.get("orientation") if screen.get("orientation") in {"landscape", "portrait"} else "landscape"
density = screen.get("density") if screen.get("density") in {"comfortable", "compact"} else "comfortable"
return {
"templateId": template_id,
"title": str(screen.get("title") or "RLCD STATUS"),
"orientation": orientation,
"density": density,
"effects": {
"ghost": bool(effects.get("ghost", True)),
"grid": bool(effects.get("grid", True)),
},
"fields": fields,
}
def firmware_catalog() -> list[dict[str, Any]]:
data = read_json(FIRMWARES_PATH, {"firmwares": []})
result: list[dict[str, Any]] = []
for item in data.get("firmwares", []):
firmware = dict(item)
if firmware.get("type") == "whole-image":
path = Path(firmware.get("path", ""))
firmware["ready"] = path.exists() and path.is_file()
firmware["size"] = path.stat().st_size if firmware["ready"] else 0
elif firmware.get("type") == "idf-project":
project_path = Path(firmware.get("projectPath", ""))
artifact = project_path / firmware.get("artifact", "")
firmware["ready"] = artifact.exists() and artifact.is_file()
firmware["size"] = artifact.stat().st_size if firmware["ready"] else 0
else:
firmware["ready"] = False
result.append(firmware)
return result
def find_firmware(firmware_id: str) -> dict[str, Any]:
for firmware in firmware_catalog():
if firmware.get("id") == firmware_id:
return firmware
raise ValueError(f"Unknown firmware id: {firmware_id}")
def ensure_known_path(path: Path, allowed_roots: list[Path]) -> Path:
resolved = path.resolve()
for root in allowed_roots:
try:
resolved.relative_to(root.resolve())
return resolved
except ValueError:
continue
raise ValueError(f"Path is not allowlisted: {resolved}")
def validate_port(port: str) -> str:
port = (port or "COM3").strip().upper()
if not re.fullmatch(r"COM\d{1,3}", port):
raise ValueError("Only COM ports are allowed on this Windows workbench.")
return port
def run_eim(command: str, timeout: int) -> dict[str, Any]:
if not EIM_EXE.exists():
raise RuntimeError(f"EIM CLI not found: {EIM_EXE}")
completed = subprocess.run(
[str(EIM_EXE), "run", command, IDF_VERSION],
cwd=str(PROJECT_ROOT),
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
timeout=timeout,
check=False,
)
output = completed.stdout[-12000:] if completed.stdout else ""
return {
"ok": completed.returncode == 0,
"returncode": completed.returncode,
"output": output,
}
def generate_ui_config(screen: dict[str, Any]) -> str:
screen = normalize_screen(screen)
fields = screen["fields"]
template_macro = TEMPLATE_IDS[screen["templateId"]]
orientation = 1 if screen["orientation"] == "portrait" else 0
density = 1 if screen["density"] == "compact" else 0
primary_value = c_int(fields.get("primaryValue"), 0, -999, 999)
secondary_value = c_int(fields.get("secondaryValue"), 0, -999, 999)
time_text = fields.get("time") or "12:00"
lines = [
"/* Generated by tools/preview-bridge/server.py. */",
"#pragma once",
"",
"#define UI_TEMPLATE_CLOCK 1",
"#define UI_TEMPLATE_DUAL_METRIC 2",
"#define UI_TEMPLATE_METER 3",
f"#define UI_ACTIVE_TEMPLATE {template_macro}",
"",
"#define UI_ORIENTATION_LANDSCAPE 0",
"#define UI_ORIENTATION_PORTRAIT 1",
f"#define UI_ACTIVE_ORIENTATION {orientation}",
"",
"#define UI_DENSITY_COMFORTABLE 0",
"#define UI_DENSITY_COMPACT 1",
f"#define UI_ACTIVE_DENSITY {density}",
"",
f"#define UI_TITLE_TEXT {c_string(screen.get('title'), 32)}",
f"#define UI_STATUS_TEXT {c_string(fields.get('status') or 'READY', 18)}",
f"#define UI_TIME_TEXT {c_string(time_text, 12)}",
f"#define UI_DATE_TEXT {c_string(fields.get('date') or '2026-06-20', 24)}",
f"#define UI_PRIMARY_LABEL_TEXT {c_string(fields.get('primaryLabel') or 'A', 18)}",
f"#define UI_PRIMARY_VALUE_TEXT {c_string(fields.get('primaryValue') or '0', 10)}",
f"#define UI_PRIMARY_VALUE_INT {primary_value}",
f"#define UI_PRIMARY_UNIT_TEXT {c_string(fields.get('primaryUnit') or '', 8)}",
f"#define UI_SECONDARY_LABEL_TEXT {c_string(fields.get('secondaryLabel') or 'B', 18)}",
f"#define UI_SECONDARY_VALUE_TEXT {c_string(fields.get('secondaryValue') or '0', 10)}",
f"#define UI_SECONDARY_VALUE_INT {secondary_value}",
f"#define UI_SECONDARY_UNIT_TEXT {c_string(fields.get('secondaryUnit') or '', 8)}",
f"#define UI_FOOTER_LEFT_TEXT {c_string(fields.get('footerLeft') or '', 32)}",
f"#define UI_FOOTER_RIGHT_TEXT {c_string(fields.get('footerRight') or '', 32)}",
"",
]
return "\n".join(lines)
def save_generated_config(screen: dict[str, Any]) -> None:
GENERATED_DIR.mkdir(parents=True, exist_ok=True)
UI_CONFIG_PATH.write_text(generate_ui_config(screen), encoding="utf-8")
class WorkbenchHandler(SimpleHTTPRequestHandler):
server_version = "RLCDWorkbench/1.0"
def end_headers(self) -> None:
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
super().end_headers()
def do_OPTIONS(self) -> None:
self.send_response(204)
self.end_headers()
def do_GET(self) -> None:
parsed = urlparse(self.path)
if parsed.path.startswith("/api/"):
self.handle_api_get(parsed.path)
return
self.serve_frontend(parsed.path)
def do_POST(self) -> None:
parsed = urlparse(self.path)
if parsed.path.startswith("/api/"):
self.handle_api_post(parsed.path)
return
self.send_error(404)
def read_body_json(self) -> Any:
length = int(self.headers.get("Content-Length", "0"))
if length <= 0:
return {}
raw = self.rfile.read(length).decode("utf-8")
return json.loads(raw or "{}")
def send_json(self, data: Any, status: int = 200) -> None:
body = response_bytes(data)
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def send_error_json(self, message: str, status: int = 400) -> None:
self.send_json({"ok": False, "error": message}, status=status)
def handle_api_get(self, path: str) -> None:
try:
if path == "/api/health":
self.send_json({"ok": True, "projectRoot": str(PROJECT_ROOT)})
elif path == "/api/templates":
self.send_json(read_json(TEMPLATES_PATH, {"templates": []}))
elif path == "/api/screen":
self.send_json({"screen": normalize_screen(read_json(ACTIVE_SCREEN_PATH, {}))})
elif path == "/api/firmwares":
self.send_json({"firmwares": firmware_catalog()})
else:
self.send_error_json("Unknown API endpoint", 404)
except Exception as exc: # noqa: BLE001
self.send_error_json(str(exc), 500)
def handle_api_post(self, path: str) -> None:
try:
body = self.read_body_json()
if path == "/api/screen":
screen = normalize_screen(body)
write_json(ACTIVE_SCREEN_PATH, screen)
self.send_json({"ok": True, "message": "Screen config saved."})
elif path == "/api/generate":
screen = normalize_screen(body.get("screen") or read_json(ACTIVE_SCREEN_PATH, {}))
write_json(ACTIVE_SCREEN_PATH, screen)
save_generated_config(screen)
self.send_json({"ok": True, "message": f"Generated {UI_CONFIG_PATH}"})
elif path == "/api/build":
screen = normalize_screen(read_json(ACTIVE_SCREEN_PATH, {}))
save_generated_config(screen)
result = run_eim("idf.py build", timeout=900)
self.send_json(result, status=200 if result["ok"] else 500)
elif path == "/api/flash":
self.flash_firmware(body)
else:
self.send_error_json("Unknown API endpoint", 404)
except subprocess.TimeoutExpired:
self.send_error_json("Command timed out.", 504)
except Exception as exc: # noqa: BLE001
self.send_error_json(str(exc), 500)
def flash_firmware(self, body: dict[str, Any]) -> None:
if body.get("confirmed") is not True:
raise ValueError("Flash requires explicit confirmation.")
port = validate_port(str(body.get("port") or "COM3"))
firmware = find_firmware(str(body.get("firmwareId") or ""))
firmware_type = firmware.get("type")
if firmware_type == "whole-image":
path = ensure_known_path(
Path(firmware.get("path", "")),
[Path(r"D:\esp\waveshare-rlcd-4.2\03_Firmware")],
)
if not path.exists():
raise FileNotFoundError(path)
offset = firmware.get("offset") or "0x0"
command = f'esptool.py --chip esp32s3 -p {port} -b 460800 write_flash {offset} "{path}"'
result = run_eim(command, timeout=900)
elif firmware_type == "idf-project":
project_path = ensure_known_path(Path(firmware.get("projectPath", "")), [PROJECT_ROOT])
if project_path != PROJECT_ROOT:
raise ValueError("Only the current ESP-IDF project can be flashed.")
result = run_eim(f"idf.py -p {port} flash", timeout=900)
else:
raise ValueError(f"Unsupported firmware type: {firmware_type}")
self.send_json(result, status=200 if result["ok"] else 500)
def serve_frontend(self, raw_path: str) -> None:
path = unquote(raw_path or "/")
if path == "/":
path = "/index.html"
candidate = (FRONTEND_DIR / path.lstrip("/")).resolve()
try:
candidate.relative_to(FRONTEND_DIR.resolve())
except ValueError:
self.send_error(403)
return
if not candidate.exists() or not candidate.is_file():
self.send_error(404)
return
self.path = "/" + str(candidate.relative_to(FRONTEND_DIR)).replace(os.sep, "/")
self.directory = str(FRONTEND_DIR)
return super().do_GET()
def main() -> int:
port = int(os.environ.get("RLCD_WORKBENCH_PORT", DEFAULT_PORT))
address = ("127.0.0.1", port)
httpd = ThreadingHTTPServer(address, WorkbenchHandler)
print(f"RLCD workbench bridge: http://127.0.0.1:{port}")
print(f"Project root: {PROJECT_ROOT}")
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\nStopping bridge.")
return 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink