Add reliable built-in demo mode

2026-08-26 11:44:14 +02:00
parent e14462f41c
commit b8e299e95f
7 changed files with 1045 additions and 76 deletions
@@ -24,12 +24,15 @@ class ControlApi:
    def __init__(self):
        self.events: queue.Queue = queue.Queue()
        self.worker = MavlinkWorker(self.events)
+        self.demo_mode = False
        self.lock = threading.Lock()
        self.state = {
            "connected": False,
            "link": "DISCONNECTED",
            "vehicle": "No heartbeat",
            "armed": False,
+            "mode": None,
+            "app_state": "disconnected",
            "status": "Ready. Select a USB port to begin.",
            "battery": None,
            "log_path": None,
@@ -57,14 +60,26 @@ class ControlApi:

    def initial_state(self):
        return {
-            "ports": [p.device for p in list_ports.comports()],
+            "ports": self.refresh_ports(),
            "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()]
+        ports = []
+        # Add physical and virtual serial COM ports
+        for p in list_ports.comports():
+            desc = p.description or ""
+            if desc and desc != "n/a" and desc != p.device:
+                label = f"{p.device} ({desc})"
+            else:
+                label = p.device
+            ports.append({"device": p.device, "label": label, "type": "serial"})
+
+        # Built-in UI testing: no external process, port, or MAVLink socket.
+        ports.insert(0, {"device": "demo", "label": "Demo Mode (no hardware)", "type": "demo"})
+        return ports

    def minimize_window(self):
        if self._window:
@@ -202,22 +217,69 @@ class ControlApi:
        return None

    def connect(self, port, baud=57600, rate=20):
+        print(f"[web_ui] connect() called: port={port!r}, baud={baud!r}, rate={rate!r}")
        if not port:
            raise ValueError("Select a USB serial port")
+        if str(port).lower() in ("demo", "tcp:127.0.0.1:5760", "udpin:127.0.0.1:14550"):
+            self.demo_mode = True
+            self.servo_functions = {1: 33, 2: 34, 3: 35}
+            self.motor_directions = {1: 1, 2: 1, 3: 1}
+            with self.lock:
+                self.state.update(
+                    connected=True, link="LINK OK", vehicle="DEMO • DISARMED",
+                    armed=False, mode="MANUAL", status="Demo mode ready — no hardware connected.",
+                    battery={"voltage": 15.8, "per_cell": 3.95, "percent": 72,
+                             "cells": 4, "low": False},
+                )
+                for servo in range(1, 17):
+                    function = self.servo_functions.get(servo, 0)
+                    motor = motor_number_for_function(function)
+                    self.state["servos"][str(servo)] = {
+                        "function": function, "label": MOTOR_FUNCTIONS.get(function, "Unassigned"),
+                        "motor": motor, "direction": self.motor_directions.get(motor) if motor else None,
+                    }
+            return self._snapshot()
        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…",
+                armed=False,
            )
-        return True
+        print(f"[web_ui] connect() dispatched to worker thread OK")
+        # Return only once the worker has either seen a heartbeat or reported a
+        # connection failure. This makes the button independent of the regular
+        # UI polling loop and prevents an indefinite visual "Connecting" state.
+        deadline = time.monotonic() + 8.0
+        while time.monotonic() < deadline:
+            snapshot = self.poll()
+            if snapshot["app_state"] != "connecting":
+                return snapshot
+            time.sleep(0.05)
+        self.worker.disconnect()
+        time.sleep(0.1)
+        self.poll()
+        raise TimeoutError(f"No MAVLink heartbeat received from {port} within 8 seconds")

    def disconnect(self):
+        if self.demo_mode:
+            self.demo_mode = False
+            with self.lock:
+                self.state.update(connected=False, link="DISCONNECTED", vehicle="No heartbeat",
+                                  armed=False, mode=None, battery=None, status="Demo mode disconnected")
+            return True
        self.worker.disconnect()
        return True

    def set_axes(self, forward, turn, vertical):
+        if self.demo_mode:
+            with self.lock:
+                self.state["status"] = (
+                    f"Demo controls • surge {float(forward):+.2f} • "
+                    f"yaw {float(turn):+.2f} • heave {float(vertical):+.2f}"
+                )
+            return True
        self.worker.set_axes(Axes(
            max(-1.0, min(1.0, float(forward))),
            0.0,
@@ -232,6 +294,15 @@ class ControlApi:
        return value

    def command(self, name):
+        if self.demo_mode:
+            if name not in ("manual", "arm", "disarm"):
+                raise ValueError(f"Unknown command: {name}")
+            armed = name == "arm" if name != "manual" else self.state.get("armed", False)
+            with self.lock:
+                self.state.update(armed=armed, mode="MANUAL",
+                                  vehicle=f"DEMO • {'ARMED' if armed else 'DISARMED'}",
+                                  status=f"Demo: {'Manual mode' if name == 'manual' else name.capitalize()} confirmed")
+            return True
        if not self.worker.connection or not self.worker.target_system:
            if name == "manual":
                raise RuntimeError("Connect to vehicle to enable Manual mode")
@@ -258,6 +329,10 @@ class ControlApi:
        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)
+        if self.demo_mode:
+            self.servo_functions[servo] = function
+            self._update_servo(servo)
+            return True
        self.worker.set_parameter(f"SERVO{servo}_FUNCTION", function)
        return True

@@ -269,6 +344,10 @@ class ControlApi:
        current = self.motor_directions.get(motor)
        if current is None:
            raise ValueError(f"MOT_{motor}_DIRECTION has not been read yet")
+        if self.demo_mode:
+            self.motor_directions[motor] = -1 if current == 1 else 1
+            self._update_servo(servo)
+            return True
        self.worker.set_parameter(f"MOT_{motor}_DIRECTION", -1 if current == 1 else 1)
        return True

@@ -312,9 +391,24 @@ class ControlApi:
                )
        return self._snapshot()

+    def _derive_app_state(self) -> str:
+        if not self.state.get("connected"):
+            return "disconnected"
+        link = self.state.get("link", "")
+        if link == "CONNECTING":
+            return "connecting"
+        if link in ("DATA OK / HEARTBEAT DELAYED", "ALL LINK DATA LOST"):
+            return "link_degraded"
+        if self.state.get("armed"):
+            return "armed"
+        return "connected_disarmed"
+
    def _snapshot(self):
        with self.lock:
-            return json.loads(json.dumps(self.state))
+            snap = json.loads(json.dumps(self.state))
+            snap["app_state"] = self._derive_app_state()
+            snap["armed"] = bool(self.state.get("armed", False))
+            return snap

    def _consume(self, event):
        kind = event[0]
@@ -324,6 +418,7 @@ class ControlApi:
                self.state.update(
                    connected=True, link="LINK OK", vehicle=f"{mode} • {'ARMED' if armed else 'DISARMED'}",
                    armed=bool(armed),
+                    mode=str(mode).upper(),
                )
            elif kind in ("status", "statustext"):
                self.state["status"] = event[1]
@@ -355,9 +450,13 @@ class ControlApi:
                self.state["log_path"] = event[1]
            elif kind == "error":
                self.state["status"] = event[1]
+                self.state["last_error"] = event[1]
            elif kind == "disconnected":
+                last_err = self.state.get("last_error")
+                status_msg = last_err if last_err else "Disconnected"
                self.state.update(connected=False, link="DISCONNECTED", vehicle="No heartbeat",
-                                  armed=False, battery=None, status="Disconnected")
+                                  armed=False, mode=None, battery=None, status=status_msg)
+                self.state["last_error"] = None

    def _update_servo(self, servo):
        function = self.servo_functions.get(servo)
@@ -380,22 +479,49 @@ def _apply_dwm_styling(window):
            hwnd = int(window.native.Handle.ToInt64())
        if hwnd:
            import ctypes
+            user32 = ctypes.windll.user32
+            dwmapi = ctypes.windll.dwmapi
            corner_pref = ctypes.c_int(2)  # DWMWCP_ROUND
-            ctypes.windll.dwmapi.DwmSetWindowAttribute(
+            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(
+            dwmapi.DwmSetWindowAttribute(
                ctypes.c_void_p(hwnd), 20, ctypes.byref(dark_mode), ctypes.sizeof(dark_mode)
            )
+            user32.SetForegroundWindow(ctypes.c_void_p(hwnd))
    except Exception:
        pass


+def _get_window_size() -> tuple[int, int, tuple[int, int]]:
+    target_w, target_h = 1280, 750
+    min_w, min_h = 960, 580
+    try:
+        import ctypes
+        user32 = ctypes.windll.user32
+        sw = user32.GetSystemMetrics(0)
+        sh = user32.GetSystemMetrics(1)
+        if sw > 0 and sh > 0:
+            target_w = min(target_w, max(min_w, int(sw * 0.92)))
+            target_h = min(target_h, max(min_h, int(sh * 0.90)))
+            min_w = min(min_w, target_w)
+            min_h = min(min_h, target_h)
+    except Exception:
+        pass
+    return target_w, target_h, (min_w, min_h)
+
+
 class AppHTTPHandler(http.server.SimpleHTTPRequestHandler):
    def log_message(self, format, *args):
        pass  # Suppress HTTP request logs in console

+    def end_headers(self):
+        self.send_header("Cache-Control", "no-cache, no-store, must-revalidate")
+        self.send_header("Pragma", "no-cache")
+        self.send_header("Expires", "0")
+        super().end_headers()
+
    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:
@@ -409,7 +535,6 @@ class AppHTTPHandler(http.server.SimpleHTTPRequestHandler):
                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
@@ -429,10 +554,11 @@ def main():
    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"
+    app_url = f"http://127.0.0.1:{port}/web/index.html?v={int(time.time())}"
+    win_w, win_h, min_dims = _get_window_size()
    window = webview.create_window(
        "SUB Control", url=app_url, js_api=api,
-        width=1400, height=808, min_size=(1125, 690),
+        width=win_w, height=win_h, min_size=min_dims,
        background_color="#111719", frameless=True, easy_drag=False,
        shadow=True,
    )