Lightshow status improvements

src/gunpla/base_gundam.py

@@ -64,6 +64,12 @@ def all_off(self) -> None:
         for led in self.get_all_leds():
             led.off()
 
+    def has_led(self, led_name: str) -> bool:
+        """
+        :return: True if led_name is a configured LED, False otherwise
+        """
+        return any(entry['name'] == led_name for entry in self.config['leds'])
+
     def get_all_leds(self, ignore_list: list[str] = None) -> list[LED]:
         """
         Returns all LEDs configured, enabled or disabled.  But not the board_led

src/server/Wrappers.py

@@ -1,6 +1,3 @@
-from src.server.RouteDecorator import lightshow_route
-
-
 def safe_execution(func):
     """
     Wraps an async route handler with a try/except block.
@@ -23,18 +20,19 @@ async def wrapper(*args, **kwargs):
     return wrapper
 
 
-def create_show_handler(func, show_manager):
+def create_show_handler(name, func, show_manager):
     """
-    Helper that when given a function, wraps it as a lighthow_route and safe_execution.
+    Builds a route handler that starts func as the named lightshow through the
+    show manager.  Errors inside the running show itself are recorded by the
+    manager's supervisor, not here — safe_execution only guards the request.
+    :param name: the lightshow name from the model config, shown in the UI
+    :param func: the async lightshow method on the gunpla
     :param show_manager: the LightshowManager that owns the running show
-    if needed we can add back in the request obj to show_handler and func(request)
-    :param func:
     :return:
     """
-    # note order matters for these
-    @lightshow_route(show_manager)
     @safe_execution
-    async def show_handler():
-        return await func()
+    async def show_handler(request):
+        await show_manager.start(name, func)
+        return {"status": "started", "show": name}, 202
 
     return show_handler

src/server/lightshow_manager.py

@@ -0,0 +1,140 @@
+import asyncio
+
+# Status states reported to the UI (MicroPython has no enum module)
+IDLE = "idle"            # nothing has run yet
+RUNNING = "running"      # a lightshow is currently running
+COMPLETED = "completed"  # the last lightshow ran to its natural end
+STOPPED = "stopped"      # the last lightshow was cancelled by the user or replaced
+ERRORED = "errored"      # the last lightshow raised an exception
+FINISHED = "finished"    # the last activity was a one-shot action like "All LEDs on"
+
+
+class LightshowManager:
+    """
+    Owns the single running lightshow task for a gunpla and the status shown in
+    the UI.  start/stop/run_action are serialized with a lock so overlapping
+    requests can't orphan a running show or leave two shows fighting over the
+    same LEDs.  Each show runs under a supervisor coroutine that records how it
+    ended (completed, stopped, or errored) — outcomes MicroPython's Task object
+    can't report on its own.
+    """
+
+    def __init__(self, gunpla):
+        self.gunpla = gunpla
+        self._task = None
+        self._lock = asyncio.Lock()
+        self._state = IDLE
+        self._show = None
+        self._error = None
+
+    def status(self) -> dict:
+        """
+        :return: the current status as a dict, suitable for JSON responses
+        """
+        return {"state": self._state, "show": self._show, "error": self._error}
+
+    def describe(self) -> str:
+        """
+        :return: a human readable one-liner of the current status for the UI
+        """
+        if self._state == IDLE:
+            return "No lightshow has run yet."
+        if self._state == RUNNING:
+            return f"Lightshow '{self._show}' is running."
+        if self._state == COMPLETED:
+            return f"Lightshow '{self._show}' completed."
+        if self._state == STOPPED:
+            return f"Lightshow '{self._show}' was stopped."
+        if self._state == ERRORED:
+            return f"Lightshow '{self._show}' failed: {self._error}"
+        return f"{self._show} — finished."
+
+    def is_running(self) -> bool:
+        """
+        :return: True if a lightshow is currently running, False otherwise
+        """
+        return self._task is not None and not self._task.done()
+
+    async def start(self, name: str, func) -> None:
+        """
+        Cancels any running show (turning the LEDs off), then starts func as
+        the tracked lightshow task under a supervisor that records its outcome.
+        """
+        async with self._lock:
+            await self._cancel_current()
+            self._set_status(RUNNING, name)
+            self._task = asyncio.create_task(self._supervise(name, func))
+
+    async def stop(self):
+        """
+        Cancels any running lightshow, waits for it to clean up and turns the
+        LEDs off — whether or not a show was actually running.
+        :return: the name of the cancelled show, or None if nothing was running
+        """
+        async with self._lock:
+            stopped_show = await self._cancel_current()
+            if stopped_show is None:
+                self.gunpla.all_off()
+            return stopped_show
+
+    async def run_action(self, description: str, action) -> None:
+        """
+        Runs a one-shot LED action (e.g. "All LEDs on"), cancelling any running
+        show first so the show can't fight the user over the LEDs.  If the
+        action itself fails, that's recorded as the current status instead of
+        leaving the (now-cancelled) show's stale status in place.
+        :param description: what the action is, shown in the UI status
+        :param action: a plain callable that manipulates the LEDs
+        """
+        async with self._lock:
+            await self._cancel_current()
+            try:
+                action()
+            except Exception as e:
+                self._set_status(ERRORED, description, str(e))
+                raise
+            self._set_status(FINISHED, description)
+
+    def _set_status(self, state: str, show: str, error: str = None) -> None:
+        self._state = state
+        self._show = show
+        self._error = error
+
+    async def _supervise(self, name: str, func) -> None:
+        try:
+            await func()
+        except asyncio.CancelledError:
+            self._set_status(STOPPED, name)
+            raise
+        except Exception as e:
+            print(f"Lightshow '{name}' failed: {e}")
+            self._set_status(ERRORED, name, str(e))
+        else:
+            self._set_status(COMPLETED, name)
+
+    async def _cancel_current(self):
+        """
+        Cancels the tracked show if one is still running, waits for its cleanup
+        and turns all LEDs off.  Callers must hold the lock.
+
+        self._task is kept set until the cancelled task is fully done, so
+        is_running() keeps reporting True for the outgoing show throughout its
+        cleanup instead of flipping to False before the supervisor has updated
+        status() to match — the two would otherwise disagree for however long
+        the show's own cleanup takes to unwind.
+
+        :return: the name of the cancelled show, or None if nothing was running
+        """
+        task = self._task
+        if task is None or task.done():
+            self._task = None
+            return None
+        name = self._show
+        task.cancel()
+        try:
+            await task  # Wait for cleanup; the supervisor records the outcome
+        except asyncio.CancelledError:
+            pass
+        self._task = None
+        self.gunpla.all_off()
+        return name

src/server/webserver.py

@@ -4,9 +4,9 @@
 from src.gunpla.generic_gundam import GenericGundam
 from src.hardware.Hardware import Hardware
 from src.pi.led_effect import LEDEffects
+from src.server.lightshow_manager import LightshowManager
 from src.server.microdot.Microdot import Microdot, Request
 from src.server.microdot.utemplate import Template
-from src.server.RouteDecorator import LightshowManager
 from src.server.Wrappers import create_show_handler, safe_execution
 
 
@@ -40,12 +40,11 @@ async def index(self, request: Request):
         """
         led_list = [{"name": led.name()} for led in self.gundam.get_all_leds()]
         show_list = self.gundam.config['lightshow']
-        running_show = self._is_lightshow_running()
         return await Template('index.html').render_async(
             name_of_title="Gundam LED Control",
             all_leds=led_list,
             lightshows=show_list,
-            running_show=running_show
+            status_message=self.show_manager.describe()
         ), 200, {'Content-Type': 'text/html'}
 
     @safe_execution
@@ -56,22 +55,22 @@ async def canary(self, request: Request):
         asyncio.create_task(LEDEffects.blink(self.hardware.board_led()))
         return "chirp", 202
 
-    def all_on(self, request: Request):
+    async def all_on(self, request: Request):
         """
-          Turns on all LEDs.
+          Turns on all LEDs, halting any running lightshow first.
           :param request:  Ignored.
           :return: HTTP 202 and message
           """
-        self.gundam.all_on()
+        await self.show_manager.run_action("All LEDs on", self.gundam.all_on)
         return "All leds are on", 202
 
-    def all_off(self, request: Request):
+    async def all_off(self, request: Request):
         """
-        Turns off all LEDs.
+        Turns off all LEDs, halting any running lightshow first.
         :param request:  Ignored.
         :return: HTTP 202 and message
         """
-        self.gundam.all_off()
+        await self.show_manager.run_action("All LEDs off", self.gundam.all_off)
         return "All leds are off", 202
 
     async def _connect_to_wifi(self):
@@ -102,12 +101,6 @@ async def run(self):
             port=self.settings.get('port', 80),
             debug=self.settings.get('debug', True))
 
-    def _is_lightshow_running(self):
-        """
-        :return: True if lightshow is running, False otherwise
-        """
-        return self.show_manager.is_running()
-
     def _add_routes(self):
         """
            Given a server adds all endpoints for Leds and lightshows
@@ -119,12 +112,18 @@ def _add_routes(self):
         @self.app.route("/led/<led_name>/on")
         @safe_execution
         async def led_on_handler(request, led_name):
-            return self.gundam.led_on(led_name)
+            # Validate before run_action, which cancels any running show — an
+            # unknown led_name shouldn't destroy a show just to then 500.
+            if not self.gundam.has_led(led_name):
+                raise Exception(f"Entry '{led_name}' not found")
+            await self.show_manager.run_action(f"LED '{led_name}' on", lambda: self.gundam.led_on(led_name))
 
         @self.app.route("/led/<led_name>/off")
         @safe_execution
         async def led_off_handler(request, led_name):
-            return self.gundam.led_off(led_name)
+            if not self.gundam.has_led(led_name):
+                raise Exception(f"Entry '{led_name}' not found")
+            await self.show_manager.run_action(f"LED '{led_name}' off", lambda: self.gundam.led_off(led_name))
 
         self.app.route("/all/on")(self.all_on)
         self.app.route("/all/off")(self.all_off)
@@ -134,21 +133,30 @@ async def led_off_handler(request, led_name):
             path = f"/lightshow/{lightshow['path']}"
             method_func = getattr(self.gundam, lightshow['method'])
 
-            self.app.route(path)(create_show_handler(method_func, self.show_manager))
+            self.app.route(path)(create_show_handler(lightshow['name'], method_func, self.show_manager))
 
         @self.app.route("/lightshow/stop")
         @safe_execution
         async def stop_lightshow(request):
             """
             Stops any currently running lightshow task on the gundam instance.
+            Stop always leaves all the LEDs off, running show or not.
             """
-            stopped = await self.show_manager.stop()
-            self.gundam.all_off()
-
-            if stopped:
-                return {"status": "stopped", "message": "Lightshow terminated"}, 200
+            stopped_show = await self.show_manager.stop()
+            if stopped_show:
+                return {"status": "stopped", "message": f"Lightshow '{stopped_show}' terminated"}, 200
             return {"status": "idle", "message": "No active lightshow to stop"}, 200
 
+        @self.app.route("/lightshow/status")
+        @safe_execution
+        async def lightshow_status(request):
+            """
+            Reports the current lightshow status as JSON.
+            """
+            status = self.show_manager.status()
+            status["message"] = self.show_manager.describe()
+            return status, 200
+
         # 404 Handler
         @self.app.errorhandler(404)
         async def not_found(request):