from __future__ import annotations import json import os import queue import threading import time import base64 import shutil import http.server import socketserver import urllib.parse from pathlib import Path import webview from serial.tools import list_ports from app import Axes, MOTOR_FUNCTIONS, MavlinkWorker, motor_number_for_function class ControlApi: """Small, local-only bridge between the CSS UI and the MAVLink worker.""" def __init__(self): self.events: queue.Queue = queue.Queue() self.worker = MavlinkWorker(self.events) self.lock = threading.Lock() self.state = { "connected": False, "link": "DISCONNECTED", "vehicle": "No heartbeat", "armed": False, "status": "Ready. Select a USB port to begin.", "battery": None, "log_path": None, "servos": {str(i): {"function": None, "direction": None} for i in range(1, 17)}, } self.servo_functions: dict[int, int] = {} self.motor_directions: dict[int, int] = {} self.battery_path = Path(__file__).resolve().parent / "battery_settings.json" self.model_path = Path(__file__).resolve().parent / "model_settings.json" self.model_cache = Path(__file__).resolve().parent / "runtime" / "vehicle_model.stl" self.model_cache_name = Path(__file__).resolve().parent / "runtime" / "vehicle_model_name.txt" # Leading underscore keeps the native WebView object out of js_api # serialization. Only the small public methods below are exposed. self._window = None self.maximized = False self.battery_config = { "cells": 4, "capacity_mah": 5500, "c_rating": 35, "full_v_per_cell": 4.2, "empty_v_per_cell": 3.3, "warning_percent": 20, } try: self.battery_config.update(json.loads(self.battery_path.read_text(encoding="utf-8"))) except (OSError, ValueError, TypeError): pass def initial_state(self): return { "ports": [p.device for p in list_ports.comports()], "battery_config": self.battery_config, "state": self._snapshot(), "model_info": self.get_saved_model_info(), } def refresh_ports(self): return [p.device for p in list_ports.comports()] def minimize_window(self): if self._window: self._window.minimize() return True def toggle_maximize_window(self): if not self._window: return False if self.maximized: self._window.restore() else: self._window.maximize() self.maximized = not self.maximized return self.maximized def close_window(self): if self._window: self._window.destroy() return True def choose_model(self): selected = None # 1. Native Windows topmost open file dialog (instant, reliable) try: import tkinter as tk from tkinter import filedialog root = tk.Tk() root.withdraw() root.attributes("-topmost", True) file_path = filedialog.askopenfilename( title="Select 3D STL Vehicle Model", initialdir=str(Path.home() / "Downloads"), filetypes=[("STL 3D Meshes (*.stl)", "*.stl"), ("All files (*.*)", "*.*")], ) root.destroy() if file_path: selected = (file_path,) except Exception: pass # 2. Fallback to pywebview create_file_dialog if not selected and self._window: try: selected = self._window.create_file_dialog( webview.FileDialog.OPEN, directory=str(Path.home()), allow_multiple=False, file_types=("STL mesh (*.stl)", "All files (*.*)"), ) except Exception: selected = None if not selected: return None path = Path(selected[0]) if not path.is_file() or not path.suffix.lower().endswith(".stl"): raise ValueError("Choose an STL mesh") if path.stat().st_size > 30 * 1024 * 1024: raise ValueError("The STL is larger than 30 MB; export a lighter mesh") content = path.read_bytes() self.model_cache.parent.mkdir(exist_ok=True) self.model_cache.write_bytes(content) self.model_cache_name.write_text(path.name, encoding="utf-8") try: self.model_path.write_text(json.dumps({"path": str(path)}, indent=2) + "\n", encoding="utf-8") except Exception: pass return {"name": path.name, "data": base64.b64encode(content).decode("ascii")} def save_model(self, name, data): safe_name = Path(str(name)).name if not safe_name.lower().endswith(".stl"): raise ValueError("Choose an STL mesh") try: content = base64.b64decode(data, validate=True) except (ValueError, TypeError) as exc: raise ValueError("Invalid STL data") from exc if not content or len(content) > 30 * 1024 * 1024: raise ValueError("STL must be between 1 byte and 30 MB") self.model_cache.parent.mkdir(exist_ok=True) self.model_cache.write_bytes(content) self.model_cache_name.write_text(safe_name, encoding="utf-8") return True def get_saved_model_info(self): self._migrate_legacy_model() try: if self.model_cache.is_file() and self.model_cache.stat().st_size <= 30 * 1024 * 1024: name = self.model_cache_name.read_text(encoding="utf-8").strip() or "Main Assembly V1 (2).stl" return {"name": name, "size": self.model_cache.stat().st_size} except (OSError, ValueError): pass return None def read_saved_model_chunk(self, offset, size=262144): offset, size = int(offset), min(int(size), 524288) if offset < 0 or size < 1 or not self.model_cache.is_file(): raise ValueError("Saved model is unavailable") with self.model_cache.open("rb") as stream: stream.seek(offset) content = stream.read(size) return {"offset": offset, "data": base64.b64encode(content).decode("ascii"), "eof": offset + len(content) >= self.model_cache.stat().st_size} def _migrate_legacy_model(self): if not self.model_cache.is_file(): default_asset = Path(__file__).resolve().parent / "assets" / "Main Assembly V1 (2).stl" if default_asset.is_file(): try: self.model_cache.parent.mkdir(exist_ok=True) shutil.copyfile(default_asset, self.model_cache) self.model_cache_name.write_text(default_asset.name, encoding="utf-8") return except Exception: pass try: if self.model_path.is_file(): path = Path(json.loads(self.model_path.read_text(encoding="utf-8"))["path"]) if path.is_file() and path.suffix.lower() == ".stl" and path.stat().st_size <= 30 * 1024 * 1024: self.model_cache.parent.mkdir(exist_ok=True) shutil.copyfile(path, self.model_cache) self.model_cache_name.write_text(path.name, encoding="utf-8") except Exception: pass def saved_model(self): self._migrate_legacy_model() try: if self.model_cache.is_file() and self.model_cache.stat().st_size <= 30 * 1024 * 1024: name = self.model_cache_name.read_text(encoding="utf-8").strip() or "Main Assembly V1 (2).stl" return {"name": name, "data": base64.b64encode(self.model_cache.read_bytes()).decode("ascii")} except (OSError, ValueError): pass return None def connect(self, port, baud=57600, rate=20): if not port: raise ValueError("Select a USB serial port") self.worker.rate_hz = int(rate) self.worker.connect(str(port), int(baud)) with self.lock: self.state.update( connected=True, link="CONNECTING", status=f"Opening {port} at {baud} baud…", ) return True def disconnect(self): self.worker.disconnect() return True def set_axes(self, forward, turn, vertical): self.worker.set_axes(Axes( max(-1.0, min(1.0, float(forward))), 0.0, max(-1.0, min(1.0, float(vertical))), max(-1.0, min(1.0, float(turn))), )) return True def set_gain(self, value): value = max(5, min(50, int(value))) self.worker.gain = value / 100.0 return value def command(self, name): if not self.worker.connection or not self.worker.target_system: if name == "manual": raise RuntimeError("Connect to vehicle to enable Manual mode") elif name == "arm": raise RuntimeError("Connect to vehicle before arming") elif name == "disarm": raise RuntimeError("Connect to vehicle before disarming") raise RuntimeError("No autopilot heartbeat") if name == "manual": self.worker.log("OPERATOR", "clicked SET MANUAL") self.worker.set_manual_mode() elif name == "arm": self.worker.log("OPERATOR", "clicked ARM") self.worker.command_arm(True) elif name == "disarm": self.worker.log("OPERATOR", "clicked DISARM") self.worker.command_arm(False) else: raise ValueError(f"Unknown command: {name}") return True def set_servo_function(self, servo, motor): servo, motor = int(servo), int(motor) if not 1 <= servo <= 16 or not 0 <= motor <= 12: raise ValueError("Invalid servo or motor") function = 0 if motor == 0 else (32 + motor if motor <= 8 else 73 + motor) self.worker.set_parameter(f"SERVO{servo}_FUNCTION", function) return True def reverse_motor(self, servo): servo = int(servo) motor = motor_number_for_function(self.servo_functions.get(servo)) if motor is None: raise ValueError(f"SERVO{servo} is not assigned to a motor") current = self.motor_directions.get(motor) if current is None: raise ValueError(f"MOT_{motor}_DIRECTION has not been read yet") self.worker.set_parameter(f"MOT_{motor}_DIRECTION", -1 if current == 1 else 1) return True def save_battery(self, config): clean = { "cells": int(config["cells"]), "capacity_mah": int(config["capacity_mah"]), "c_rating": float(config["c_rating"]), "full_v_per_cell": float(config["full_v_per_cell"]), "empty_v_per_cell": float(config["empty_v_per_cell"]), "warning_percent": int(config["warning_percent"]), } if clean["cells"] < 1 or clean["capacity_mah"] <= 0 or clean["c_rating"] <= 0: raise ValueError("Battery cells, capacity and C rating must be positive") if not clean["full_v_per_cell"] > clean["empty_v_per_cell"] > 0: raise ValueError("Full voltage per cell must be greater than empty voltage") if not 0 <= clean["warning_percent"] <= 100: raise ValueError("Warning percentage must be between 0 and 100") self.battery_config = clean self.battery_path.write_text(json.dumps(clean, indent=2) + "\n", encoding="utf-8") return clean def open_logs(self): path = Path(__file__).resolve().parent / "logs" path.mkdir(exist_ok=True) os.startfile(path) return True def poll(self): while True: try: event = self.events.get_nowait() except queue.Empty: break self._consume(event) if self.worker.target_system and time.monotonic() - self.worker.last_heartbeat > 5: with self.lock: self.state["link"] = ( "DATA OK / HEARTBEAT DELAYED" if time.monotonic() - self.worker.last_rx < 2 else "ALL LINK DATA LOST" ) return self._snapshot() def _snapshot(self): with self.lock: return json.loads(json.dumps(self.state)) def _consume(self, event): kind = event[0] with self.lock: if kind == "heartbeat": _, system, autopilot, mode, armed = event self.state.update( connected=True, link="LINK OK", vehicle=f"{mode} • {'ARMED' if armed else 'DISARMED'}", armed=bool(armed), ) elif kind in ("status", "statustext"): self.state["status"] = event[1] elif kind == "ack": self.state["status"] = f"Command {event[1]} ACK: {event[2]}" elif kind == "battery": voltage = float(event[1]) cells = self.battery_config["cells"] per_cell = voltage / cells low, high = self.battery_config["empty_v_per_cell"], self.battery_config["full_v_per_cell"] percent = max(0, min(100, (per_cell - low) * 100 / (high - low))) self.state["battery"] = { "voltage": voltage, "per_cell": per_cell, "percent": percent, "cells": cells, "low": percent <= self.battery_config["warning_percent"], } elif kind == "servo_param": _, param_id, value = event servo = int(param_id[5:param_id.index("_")]) self.servo_functions[servo] = int(value) self._update_servo(servo) elif kind == "motor_direction": _, motor, value = event self.motor_directions[int(motor)] = int(value) for servo in range(1, 17): self._update_servo(servo) elif kind == "param_saved": self.state["status"] = f"Confirmed: {event[1]}={event[2]}" elif kind == "log_path": self.state["log_path"] = event[1] elif kind == "error": self.state["status"] = event[1] elif kind == "disconnected": self.state.update(connected=False, link="DISCONNECTED", vehicle="No heartbeat", armed=False, battery=None, status="Disconnected") def _update_servo(self, servo): function = self.servo_functions.get(servo) motor = motor_number_for_function(function) self.state["servos"][str(servo)] = { "function": function, "label": MOTOR_FUNCTIONS.get(function, "Unassigned") if function is not None else "Waiting…", "motor": motor, "direction": self.motor_directions.get(motor) if motor else None, } def close(self): self.worker.disconnect() def _apply_dwm_styling(window): try: hwnd = None if hasattr(window, "native") and window.native: hwnd = int(window.native.Handle.ToInt64()) if hwnd: import ctypes corner_pref = ctypes.c_int(2) # DWMWCP_ROUND ctypes.windll.dwmapi.DwmSetWindowAttribute( ctypes.c_void_p(hwnd), 33, ctypes.byref(corner_pref), ctypes.sizeof(corner_pref) ) dark_mode = ctypes.c_int(1) # DWMWA_USE_IMMERSIVE_DARK_MODE ctypes.windll.dwmapi.DwmSetWindowAttribute( ctypes.c_void_p(hwnd), 20, ctypes.byref(dark_mode), ctypes.sizeof(dark_mode) ) except Exception: pass class AppHTTPHandler(http.server.SimpleHTTPRequestHandler): def log_message(self, format, *args): pass # Suppress HTTP request logs in console def do_GET(self): unquoted = urllib.parse.unquote(self.path) if self.path in ("/model", "/model.stl", "/runtime/vehicle_model.stl") or "Main Assembly" in unquoted or "vehicle_model" in unquoted: project_root = Path(__file__).resolve().parent model_file = project_root / "assets" / "Main Assembly V1 (2).stl" if not model_file.is_file(): model_file = project_root / "runtime" / "vehicle_model.stl" if model_file.is_file(): data = model_file.read_bytes() self.send_response(200) self.send_header("Content-Type", "model/stl") self.send_header("Content-Length", str(len(data))) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(data) return return super().do_GET() def start_local_server(directory: Path) -> int: handler = lambda *args, **kwargs: AppHTTPHandler(*args, directory=str(directory), **kwargs) server = socketserver.TCPServer(("127.0.0.1", 0), handler) port = server.server_address[1] threading.Thread(target=server.serve_forever, daemon=True).start() return port def main(): api = ControlApi() api._migrate_legacy_model() project_root = Path(__file__).resolve().parent port = start_local_server(project_root) app_url = f"http://127.0.0.1:{port}/web/index.html" window = webview.create_window( "SUB Control", url=app_url, js_api=api, width=1400, height=808, min_size=(1125, 690), background_color="#111719", frameless=True, easy_drag=False, shadow=True, ) api._window = window window.events.shown += lambda: _apply_dwm_styling(window) window.events.maximized += lambda: setattr(api, "maximized", True) window.events.restored += lambda: setattr(api, "maximized", False) try: webview.start(debug=True) finally: api.close() if __name__ == "__main__": main()