diff --git a/docs/configuring_gunpla.md b/docs/configuring_gunpla.md index 9974118..b9d80a8 100644 --- a/docs/configuring_gunpla.md +++ b/docs/configuring_gunpla.md @@ -20,7 +20,7 @@ webserver = { "ssid": "My house wifi", "password": 'password', "hostname": 'nu-gundam.local', - "model": NuGundam() + "model": NuGundam } ``` diff --git a/src/config.py.template b/src/config.py.template index 593ce35..7e1e846 100644 --- a/src/config.py.template +++ b/src/config.py.template @@ -1,7 +1,8 @@ -from gunpla.GenericGundam import GenericGundam +from src.gunpla.generic_gundam import GenericGundam + webserver = { "ssid": '', "password": '', - "hostname": 'gunpla.local' #replace with a desired hostname, - "model": GenericGundam() #The gunpla you have -} \ No newline at end of file + "hostname": 'gunpla.local', # replace with a desired hostname + "model": GenericGundam # The gunpla you have (class reference, not an instance) +} diff --git a/src/gunpla/base_gundam.py b/src/gunpla/base_gundam.py index 648ba9d..c325b60 100644 --- a/src/gunpla/base_gundam.py +++ b/src/gunpla/base_gundam.py @@ -2,7 +2,6 @@ from src.pi.disabled_LED import DisabledLED from src.pi.LED import LED -from src.server.Wrappers import safe_execution class BaseGundam: @@ -13,6 +12,7 @@ class BaseGundam: def __init__(self, hardware): 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()) @@ -44,44 +44,16 @@ def all_on(self) -> None: Turns all configured LED's on. """ print("turning on all leds") - self._all_leds_on() - - def _all_leds_on(self) -> str: - """ - Turns all LEDs on - """ - leds: str = "" - for led_entry in self.config['leds']: - led_name = led_entry['name'] - led = self._get_led_from_name(led_name) + for led in self.get_all_leds(): led.on() - if isinstance(led, DisabledLED): - leds += f"{led_name}: disabled\n" - else: - leds += f"{led_name}: on\n" - return leds def all_off(self) -> None: """ Turns all configured LED's off """ print("turning off all leds") - self._all_leds_off() - - def _all_leds_off(self) -> str: - """" - Turns all LEDs off - """ - leds: str = "" - for led_entry in self.config['leds']: - led_name = led_entry['name'] - led = self._get_led_from_name(led_name) + for led in self.get_all_leds(): led.off() - if isinstance(led, DisabledLED): - leds += f"{led_name}: disabled\n" - else: - leds += f"{led_name}: off\n" - return leds def get_all_leds(self, ignore_list: list[str] = []) -> list[LED]: """ @@ -98,16 +70,21 @@ def get_all_leds(self, ignore_list: list[str] = []) -> list[LED]: def _get_led_from_name(self, led_name: str) -> LED: """ - Given a name of an LED, returns the LED object for it. + Given a name of an LED, returns the LED object for it, creating and caching it on first use. Throws an exception if it's not found :param led_name: :return: """ - entry = self.__get_entry_from_name(led_name) - if 'disabled' in entry and entry['disabled']: - print(f"{led_name} is disabled") - return DisabledLED(led_name) - return self.hardware.create_led(entry['pin'], led_name) + led = self._leds.get(led_name) + if led is None: + entry = self.__get_entry_from_name(led_name) + if 'disabled' in entry and entry['disabled']: + print(f"{led_name} is disabled") + led = DisabledLED(led_name) + else: + led = self.hardware.create_led(entry['pin'], led_name) + self._leds[led_name] = led + return led def __get_entry_from_name(self, led_name: str) -> json: """ diff --git a/src/gunpla/unicorn_banshee.py b/src/gunpla/unicorn_banshee.py index e4ee1cc..1b4ce80 100644 --- a/src/gunpla/unicorn_banshee.py +++ b/src/gunpla/unicorn_banshee.py @@ -21,4 +21,4 @@ async def glow(self) -> None: """ await LEDEffects.brighten_all(self.get_all_leds()) await asyncio.sleep(3) - self._all_leds_off() + self.all_off() diff --git a/src/hardware/VirtualHardware.py b/src/hardware/VirtualHardware.py index 584ea2c..fd5a342 100644 --- a/src/hardware/VirtualHardware.py +++ b/src/hardware/VirtualHardware.py @@ -61,7 +61,7 @@ class MockBoardLED(BoardLED): def __init__(self): self._pin = src.hardware.VirtualHardware.MockPin(1) - self.led_name = "Mock Board LED" + self._led_name = "Mock Board LED" def __init__(self): self.pin = self.MockPin diff --git a/src/pi/led_effect.py b/src/pi/led_effect.py index 0a7ac92..07a3e86 100644 --- a/src/pi/led_effect.py +++ b/src/pi/led_effect.py @@ -1,5 +1,4 @@ import asyncio -import time import src.hardware from src.pi import LED @@ -16,7 +15,7 @@ async def blink(led: LED) -> None: Blinks the onboard LED twice """ led.on() - time.sleep(0.5) + await asyncio.sleep(0.5) led.off() await asyncio.sleep(0.5) led.on() @@ -57,15 +56,19 @@ async def brighten(led: LED, start_percent: int = 0, end_percent: int = 100, spe :param speed: :return: """ - pwm = src.hardware.get_hardware().get_pwm(led.pin) - pwm.freq(1000) + if not led.enabled(): + return step_rate = 10 overall_change = end_percent - start_percent + if overall_change <= 0: + return interval = overall_change / step_rate sleep_time = speed / interval # print(f"overall[{overall_change}] interval[{interval}] sleep[{sleep_time}]") # 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) @@ -79,20 +82,23 @@ 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. """ - pwms = [] - for led in leds: - pwm = src.hardware.get_hardware().get_pwm(led.pin) - pwm.freq(1000) - pwms.append(pwm) - step_rate = 10 overall_change = end_percent - start_percent + if overall_change <= 0: + return interval = overall_change / step_rate sleep_time = speed / interval + pwms = [] + for led in leds: + if not led.enabled(): + continue + pwm = src.hardware.get_hardware().get_pwm(led.pin()) + pwm.freq(1000) + pwms.append(pwm) + for percent in range(start_percent, end_percent, step_rate): - print(percent) duty = int((percent / 100) * 65_535) for pwm in pwms: pwm.duty_u16(duty) diff --git a/src/server/RouteDecorator.py b/src/server/RouteDecorator.py index e9f2a19..a65f68b 100644 --- a/src/server/RouteDecorator.py +++ b/src/server/RouteDecorator.py @@ -1,6 +1,23 @@ import asyncio +async def cancel_lightshow(gunpla, manager_attr="current_task"): + """ + Cancels any running lightshow task on the gunpla and clears 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(): + existing_task.cancel() + try: + await existing_task # Wait for cleanup + except asyncio.CancelledError: + pass + return True + return False + + def lightshow_route(gunpla, manager_attr="current_task"): """ A decorator factory that handles task management and @@ -9,14 +26,8 @@ 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. - existing_task = getattr(gunpla, manager_attr, None) - if existing_task and not existing_task.done(): - existing_task.cancel() - try: - gunpla.all_off() - await existing_task # Wait for cleanup - except asyncio.CancelledError: - pass + if await cancel_lightshow(gunpla, manager_attr): + gunpla.all_off() # Start the new show and track it task = asyncio.create_task(func()) diff --git a/src/server/microdot/Microdot.py b/src/server/microdot/Microdot.py index 24f1b98..04da2ec 100644 --- a/src/server/microdot/Microdot.py +++ b/src/server/microdot/Microdot.py @@ -491,7 +491,7 @@ def form(self): if self.content_type is None: return None mime_type = self.content_type.split(';')[0] - if mime_type != 'application/x-templates-form-urlencoded': + if mime_type != 'application/x-www-form-urlencoded': return None self._form = self._parse_urlencoded(self.body) return self._form diff --git a/src/server/webserver.py b/src/server/webserver.py index ba1fca9..93dd894 100644 --- a/src/server/webserver.py +++ b/src/server/webserver.py @@ -6,6 +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.Wrappers import create_show_handler, safe_execution @@ -93,7 +94,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 + return existing_task is not None and not existing_task.done() def _add_routes(self): """ @@ -129,20 +130,11 @@ async def stop_lightshow(request): """ Stops any currently running lightshow task on the gundam instance. """ - existing_task = getattr(self.gundam, "current_task", None) - - if existing_task and not existing_task.done(): - existing_task.cancel() - try: - # Wait for the task to acknowledge the cancellation - await existing_task - except asyncio.CancelledError: - pass + stopped = await cancel_lightshow(self.gundam) + self.gundam.all_off() - self.gundam.all_off() + if stopped: return {"status": "stopped", "message": "Lightshow terminated"}, 200 - - self.gundam.all_off() return {"status": "idle", "message": "No active lightshow to stop"}, 200 # 404 Handler diff --git a/src/test.py b/src/test.py index daa3890..eed0021 100644 --- a/src/test.py +++ b/src/test.py @@ -4,6 +4,8 @@ import time +from machine import Pin + def main(): """ diff --git a/tests/LocalServerTest.py b/tests/LocalServerTest.py index 0d2f610..6d13566 100644 --- a/tests/LocalServerTest.py +++ b/tests/LocalServerTest.py @@ -17,8 +17,9 @@ class MobileDoll(GenericGundam): """ def __init__(self, hardware, model_config: json = None): - self.hardware = hardware - self.config = model_config if model_config else {} + super().__init__(hardware) + if model_config: + self.config = model_config def get_config_file(self) -> str: return "tests/config/virgo.json"