From c20d7da9b2b78d01561eea8212098f2c98244dcc Mon Sep 17 00:00:00 2001 From: frozenwizard <172203+frozenwizard@users.noreply.github.com> Date: Sun, 2 Aug 2026 21:40:42 -0500 Subject: [PATCH] more fixes --- src/gunpla/base_gundam.py | 15 ++++++-- src/hardware/Hardware.py | 7 +++- src/hardware/PicoHardwre.py | 7 ++-- src/hardware/VirtualHardware.py | 5 ++- src/pi/LED.py | 6 +++ src/pi/disabled_LED.py | 3 ++ src/pi/led_effect.py | 68 ++++++++++++++++++++------------- src/server/RouteDecorator.py | 28 ++++++++++---- src/server/webserver.py | 10 ++--- tests/LocalServerTest.py | 4 +- 10 files changed, 101 insertions(+), 52 deletions(-) diff --git a/src/gunpla/base_gundam.py b/src/gunpla/base_gundam.py index c325b60..430e5f1 100644 --- a/src/gunpla/base_gundam.py +++ b/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: """ diff --git a/src/hardware/Hardware.py b/src/hardware/Hardware.py index 5ea98a1..c8da0b3 100644 --- a/src/hardware/Hardware.py +++ b/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: diff --git a/src/hardware/PicoHardwre.py b/src/hardware/PicoHardwre.py index 12849b4..978bb8b 100644 --- a/src/hardware/PicoHardwre.py +++ b/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""" diff --git a/src/hardware/VirtualHardware.py b/src/hardware/VirtualHardware.py index fd5a342..7dd8411 100644 --- a/src/hardware/VirtualHardware.py +++ b/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""" diff --git a/src/pi/LED.py b/src/pi/LED.py index 4ba1bdc..9c72fe6 100644 --- a/src/pi/LED.py +++ b/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): """ diff --git a/src/pi/disabled_LED.py b/src/pi/disabled_LED.py index 57e9bf7..afcd59c 100644 --- a/src/pi/disabled_LED.py +++ b/src/pi/disabled_LED.py @@ -26,3 +26,6 @@ def on(self): def off(self): pass + + def set_pin(self, pin) -> None: + pass diff --git a/src/pi/led_effect.py b/src/pi/led_effect.py index 07a3e86..8f00ed3 100644 --- a/src/pi/led_effect.py +++ b/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())) diff --git a/src/server/RouteDecorator.py b/src/server/RouteDecorator.py index a65f68b..3a6ced0 100644 --- a/src/server/RouteDecorator.py +++ b/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 { diff --git a/src/server/webserver.py b/src/server/webserver.py index 93dd894..6d4e0cd 100644 --- a/src/server/webserver.py +++ b/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 diff --git a/tests/LocalServerTest.py b/tests/LocalServerTest.py index 6d13566..c9f5e67 100644 --- a/tests/LocalServerTest.py +++ b/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"