Fix/code review findings

src/gunpla/base_gundam.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 
 from src.pi.disabled_LED import DisabledLED
@@ -9,12 +10,20 @@ class BaseGundam:
     Base Gunpla.
     """
 
-    def __init__(self, hardware):
+    def __init__(self, hardware, config: dict = None):
+        """
+        :param hardware: The hardware abstraction to drive LEDs with.
+        :param config: An in-memory config to use instead of reading get_config_file() from disk.
+        """
         from src.hardware.Hardware import Hardware
         self.hardware: Hardware = hardware
         self._leds = {}
-        with open(self.get_config_file()) as config_contents:
-            self.config: json = json.loads(config_contents.read())
+        self.lightshow_lock = asyncio.Lock()
+        if config is not None:
+            self.config: json = config
+        else:
+            with open(self.get_config_file()) as config_contents:
+                self.config: json = json.loads(config_contents.read())
 
     def get_config_file(self) -> str:
         """

src/hardware/Hardware.py

@@ -14,7 +14,12 @@ def get_pin(self, pin_num, mode):
     def get_pwm(self, pin_obj):
         raise NotImplementedError
 
-    def reset_pin(self, pin_num):
+    def reset_pin(self, pin_obj):
+        """
+        Reinitializes a pin previously used for PWM back to plain digital output.
+        :param pin_obj: The Pin object to reinitialize
+        :return: The same Pin object, ready for plain digital I/O
+        """
         raise NotImplementedError
 
     def networking(self) -> Networking:

src/hardware/PicoHardwre.py

@@ -32,9 +32,10 @@ def get_pin(self, pin_num, mode="OUT"):
     def get_pwm(self, pin_obj):
         return self.PWM(pin_obj)
 
-    def reset_pin(self, pin_num):
-        """Re-initializes the pin to clear PWM settings"""
-        return self.get_pin(pin_num, mode="OUT")
+    def reset_pin(self, pin_obj):
+        """Re-initializes the pin in place to clear PWM settings"""
+        pin_obj.init(self.Pin.OUT)
+        return pin_obj
 
     def create_led(self, pin_number: int, name: str):
         """Creates a real LED with actual GPIO pin"""

src/hardware/VirtualHardware.py

@@ -79,8 +79,9 @@ def board_led(self) -> BoardLED:
     def networking(self) -> Networking:
         return self.NoOpNetworking()
 
-    def reset_pin(self, pin_num):
-        print(f"[SIM] Pin {pin_num} reset to standard GPIO")
+    def reset_pin(self, pin_obj):
+        print(f"[SIM] Pin {pin_obj.num} reset to standard GPIO")
+        return pin_obj
 
     def create_led(self, pin_number: int, name: str):
         """Creates a mock LED for simulation"""

src/pi/LED.py

@@ -43,6 +43,12 @@ def pin(self):
         """
         return self._pin
 
+    def set_pin(self, pin) -> None:
+        """
+        Replaces the underlying pin, e.g. after LEDEffects reinitializes it post-PWM use.
+        """
+        self._pin = pin
+
 
 class MockLED(LED):
     """

src/pi/disabled_LED.py

@@ -26,3 +26,6 @@ def on(self):
 
     def off(self):
         pass
+
+    def set_pin(self, pin) -> None:
+        pass

src/pi/led_effect.py

@@ -46,6 +46,21 @@ async def charge_fire(led: LED, charge_speed: int = 1) -> None:
         await asyncio.sleep(2)
         led.off()
 
+    @staticmethod
+    def _step_timing(start_percent: int, end_percent: int, speed: int):
+        """
+        Computes the step rate and per-step sleep time for a brighten effect.
+        :return: (step_rate, sleep_time), or None if the range is degenerate (nothing to animate)
+        """
+        step_rate = 10
+        overall_change = end_percent - start_percent
+        if overall_change <= 0:
+            return None
+        interval = overall_change / step_rate
+        sleep_time = speed / interval
+        # print(f"overall[{overall_change}] interval[{interval}] sleep[{sleep_time}]")
+        return step_rate, sleep_time
+
     @staticmethod
     async def brighten(led: LED, start_percent: int = 0, end_percent: int = 100, speed: int = 10) -> None:
         """
@@ -56,24 +71,21 @@ async def brighten(led: LED, start_percent: int = 0, end_percent: int = 100, spe
         :param speed:
         :return:
         """
-        if not led.enabled():
-            return
-        step_rate = 10
-
-        overall_change = end_percent - start_percent
-        if overall_change <= 0:
+        timing = LEDEffects._step_timing(start_percent, end_percent, speed)
+        if not led.enabled() or timing is None:
             return
-        interval = overall_change / step_rate
-        sleep_time = speed / interval
-        # print(f"overall[{overall_change}] interval[{interval}] sleep[{sleep_time}]")
+        step_rate, sleep_time = timing
         # todo: use interval as the loop counter and just increment percent until end_percent
         pwm = src.hardware.get_hardware().get_pwm(led.pin())
         pwm.freq(1000)
-        for percent in range(start_percent, end_percent, step_rate):
-            duty = int((percent / 100) * 65_535)
-            pwm.duty_u16(duty)
-            await asyncio.sleep(sleep_time)
-        pwm.deinit()
+        try:
+            for percent in range(start_percent, end_percent, step_rate):
+                duty = int((percent / 100) * 65_535)
+                pwm.duty_u16(duty)
+                await asyncio.sleep(sleep_time)
+        finally:
+            pwm.deinit()
+            led.set_pin(src.hardware.get_hardware().reset_pin(led.pin()))
 
     @staticmethod
     async def brighten_all(leds: list[LED], start_percent: int = 0, end_percent: int = 100, speed: int = 10) -> None:
@@ -82,27 +94,29 @@ async def brighten_all(leds: list[LED], start_percent: int = 0, end_percent: int
         around 30%  so this method should not be used until that's addressed.  I also don't think i understand all there
         is to PWM.
         """
-        step_rate = 10
-
-        overall_change = end_percent - start_percent
-        if overall_change <= 0:
+        timing = LEDEffects._step_timing(start_percent, end_percent, speed)
+        if timing is None:
             return
-        interval = overall_change / step_rate
-        sleep_time = speed / interval
+        step_rate, sleep_time = timing
 
         pwms = []
+        active_leds = []
         for led in leds:
             if not led.enabled():
                 continue
             pwm = src.hardware.get_hardware().get_pwm(led.pin())
             pwm.freq(1000)
             pwms.append(pwm)
+            active_leds.append(led)
 
-        for percent in range(start_percent, end_percent, step_rate):
-            duty = int((percent / 100) * 65_535)
+        try:
+            for percent in range(start_percent, end_percent, step_rate):
+                duty = int((percent / 100) * 65_535)
+                for pwm in pwms:
+                    pwm.duty_u16(duty)
+                await asyncio.sleep(sleep_time)
+        finally:
             for pwm in pwms:
-                pwm.duty_u16(duty)
-            await asyncio.sleep(sleep_time)
-
-        for pwm in pwms:
-            pwm.deinit()
+                pwm.deinit()
+            for led in active_leds:
+                led.set_pin(src.hardware.get_hardware().reset_pin(led.pin()))

src/server/RouteDecorator.py

@@ -1,20 +1,31 @@
 import asyncio
 
 
+def is_lightshow_running(gunpla, manager_attr="current_task") -> bool:
+    """
+    :return: True if a lightshow task is currently tracked and hasn't finished yet
+    """
+    existing_task = getattr(gunpla, manager_attr, None)
+    return existing_task is not None and not existing_task.done()
+
+
 async def cancel_lightshow(gunpla, manager_attr="current_task"):
     """
     Cancels any running lightshow task on the gunpla and clears the tracked task.
+    Callers must hold gunpla.lightshow_lock: this check-then-act sequence needs to stay
+    atomic with respect to other requests reading/writing the tracked task.
     :return: True if a running show was cancelled, False if nothing was running
     """
     existing_task = getattr(gunpla, manager_attr, None)
-    setattr(gunpla, manager_attr, None)
-    if existing_task and not existing_task.done():
+    if is_lightshow_running(gunpla, manager_attr):
         existing_task.cancel()
         try:
             await existing_task  # Wait for cleanup
         except asyncio.CancelledError:
             pass
+        setattr(gunpla, manager_attr, None)
         return True
+    setattr(gunpla, manager_attr, None)
     return False
 
 
@@ -25,13 +36,14 @@ def lightshow_route(gunpla, manager_attr="current_task"):
     """
     def decorator(func):
         async def wrapper(request, *args, **kwargs):
-            # If any existing lightshow is running, cancel it and turn off all the LEDs.
-            if await cancel_lightshow(gunpla, manager_attr):
-                gunpla.all_off()
+            async with gunpla.lightshow_lock:
+                # If any existing lightshow is running, cancel it and turn off all the LEDs.
+                if await cancel_lightshow(gunpla, manager_attr):
+                    gunpla.all_off()
 
-            # Start the new show and track it
-            task = asyncio.create_task(func())
-            setattr(gunpla, manager_attr, task)
+                # Start the new show and track it
+                task = asyncio.create_task(func())
+                setattr(gunpla, manager_attr, task)
 
             # Return common HTTP response that the show started.
             return {

src/server/webserver.py

@@ -6,7 +6,7 @@
 from src.pi.led_effect import LEDEffects
 from src.server.microdot.Microdot import Microdot, Request
 from src.server.microdot.utemplate import Template
-from src.server.RouteDecorator import cancel_lightshow
+from src.server.RouteDecorator import cancel_lightshow, is_lightshow_running
 from src.server.Wrappers import create_show_handler, safe_execution
 
 
@@ -93,8 +93,7 @@ def _is_lightshow_running(self):
         """
         :return: True if lightshow is running, False otherwise
         """
-        existing_task = getattr(self.gundam, "current_task", None)
-        return existing_task is not None and not existing_task.done()
+        return is_lightshow_running(self.gundam)
 
     def _add_routes(self):
         """
@@ -130,8 +129,9 @@ async def stop_lightshow(request):
             """
             Stops any currently running lightshow task on the gundam instance.
             """
-            stopped = await cancel_lightshow(self.gundam)
-            self.gundam.all_off()
+            async with self.gundam.lightshow_lock:
+                stopped = await cancel_lightshow(self.gundam)
+                self.gundam.all_off()
 
             if stopped:
                 return {"status": "stopped", "message": "Lightshow terminated"}, 200

tests/LocalServerTest.py

@@ -17,9 +17,7 @@ class MobileDoll(GenericGundam):
     """
 
     def __init__(self, hardware, model_config: json = None):
-        super().__init__(hardware)
-        if model_config:
-            self.config = model_config
+        super().__init__(hardware, config=model_config)
 
     def get_config_file(self) -> str:
         return "tests/config/virgo.json"