diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml new file mode 100644 index 0000000..44bf216 --- /dev/null +++ b/.github/workflows/test.yaml @@ -0,0 +1,25 @@ +name: Test + +on: + push: + branches: + - main + pull_request: + branches: + - main + +jobs: + pytest: + name: Run tests on virtual hardware + runs-on: ubuntu-latest + steps: + - name: Check out code + uses: actions/checkout@v4 + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version-file: .python-version + - name: Install pytest + run: pip install pytest + - name: Run tests + run: python -m pytest tests/ -v diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..1e219a3 --- /dev/null +++ b/conftest.py @@ -0,0 +1 @@ +# Marks the repo root for pytest so tests can import the src package. diff --git a/docs/developer_setup.md b/docs/developer_setup.md index 51ea292..9ddbd01 100644 --- a/docs/developer_setup.md +++ b/docs/developer_setup.md @@ -50,16 +50,21 @@ entering the ```repl``` and importing the required classes. Light shows and other effects can be activated such as the following. ```python +import asyncio +from src.hardware import get_hardware from src.gunpla.nu_gundam import NuGundam -g = NuGundam() -g.activation(None) +g = NuGundam(get_hardware()) +asyncio.run(g.activation()) ``` Individual effects can be used as such on a individual LED directly ```python -from src.pi.LED import LED +import asyncio +from src.hardware import get_hardware from src.pi.led_effect import LEDEffects -led = LED(0, "0") -LEDEffects.fire(led) +hardware = get_hardware() +led = hardware.create_led(0, "head") +asyncio.run(LEDEffects.fire(led)) +asyncio.run(LEDEffects(hardware).brighten(led)) ``` diff --git a/main.py b/main.py index 713fb7f..ab3088b 100644 --- a/main.py +++ b/main.py @@ -1,16 +1,11 @@ -import asyncio - -import src +import src.hardware from src import settings -from src.hardware.Hardware import Hardware -from src.server.webserver import WebServer +from src.server.webserver import run_server def main(): - hardware: Hardware = src.hardware.get_hardware() - webserver = WebServer(settings.webserver, hardware) - asyncio.run( webserver.run()) + run_server(settings.webserver, src.hardware.get_hardware()) if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/src/config.py.template b/src/config.py.template index 7e1e846..b4fb1e5 100644 --- a/src/config.py.template +++ b/src/config.py.template @@ -4,5 +4,9 @@ webserver = { "ssid": '', "password": '', "hostname": 'gunpla.local', # replace with a desired hostname - "model": GenericGundam # The gunpla you have (class reference, not an instance) + "model": GenericGundam, # The gunpla you have (class reference, not an instance) + # Optional server settings, shown with their defaults: + # "host": '0.0.0.0', + # "port": 80, + # "debug": True, } diff --git a/src/gunpla/base_gundam.py b/src/gunpla/base_gundam.py index 430e5f1..ffad81b 100644 --- a/src/gunpla/base_gundam.py +++ b/src/gunpla/base_gundam.py @@ -1,8 +1,8 @@ -import asyncio import json from src.pi.disabled_LED import DisabledLED from src.pi.LED import LED +from src.pi.led_effect import LEDEffects class BaseGundam: @@ -17,8 +17,8 @@ def __init__(self, hardware, config: dict = None): """ from src.hardware.Hardware import Hardware self.hardware: Hardware = hardware + self.effects = LEDEffects(hardware) self._leds = {} - self.lightshow_lock = asyncio.Lock() if config is not None: self.config: json = config else: @@ -64,10 +64,11 @@ def all_off(self) -> None: for led in self.get_all_leds(): led.off() - def get_all_leds(self, ignore_list: list[str] = []) -> list[LED]: + def get_all_leds(self, ignore_list: list[str] = None) -> list[LED]: """ Returns all LEDs configured, enabled or disabled. But not the board_led """ + ignore_list = ignore_list or [] leds = [] for led_entry in self.config['leds']: led_name = led_entry['name'] diff --git a/src/gunpla/nu_gundam.py b/src/gunpla/nu_gundam.py index 165129c..9b96633 100644 --- a/src/gunpla/nu_gundam.py +++ b/src/gunpla/nu_gundam.py @@ -2,7 +2,6 @@ import random from src.gunpla.base_gundam import BaseGundam -from src.pi.led_effect import LEDEffects class NuGundam(BaseGundam): @@ -26,7 +25,7 @@ async def activation(self) -> None: await asyncio.sleep(0.1) head_led.off() await asyncio.sleep(0.5) - await LEDEffects.brighten(head_led) + await self.effects.brighten(head_led) async def fire_funnels(self) -> None: """ @@ -34,7 +33,7 @@ async def fire_funnels(self) -> None: """ for i in range(1, 7): funnel = self._get_led_from_name(f"fin_funnel_{i}") - await LEDEffects.fire(funnel) + await self.effects.fire(funnel) async def random_funnels(self) -> None: """ @@ -50,5 +49,5 @@ async def random_funnels(self) -> None: while True: funnel = random.choice(funnels) - await LEDEffects.charge_fire(funnel) + await self.effects.charge_fire(funnel) await asyncio.sleep(random.uniform(0, 3)) diff --git a/src/gunpla/unicorn_banshee.py b/src/gunpla/unicorn_banshee.py index 1b4ce80..7df33a9 100644 --- a/src/gunpla/unicorn_banshee.py +++ b/src/gunpla/unicorn_banshee.py @@ -1,7 +1,6 @@ import asyncio from src.gunpla.base_gundam import BaseGundam -from src.pi.led_effect import LEDEffects class UnicornBansheeGundam(BaseGundam): @@ -19,6 +18,6 @@ async def glow(self) -> None: """ Runs the glow lightshow """ - await LEDEffects.brighten_all(self.get_all_leds()) + await self.effects.brighten_all(self.get_all_leds()) await asyncio.sleep(3) self.all_off() diff --git a/src/hardware/Hardware.py b/src/hardware/Hardware.py index c8da0b3..6260470 100644 --- a/src/hardware/Hardware.py +++ b/src/hardware/Hardware.py @@ -14,11 +14,11 @@ def get_pin(self, pin_num, mode): def get_pwm(self, pin_obj): raise NotImplementedError - def reset_pin(self, pin_obj): + def reset_pin(self, pin): """ 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 + :param pin: The Pin object to reinitialize + :return: A Pin object ready for plain digital I/O on the same GPIO """ raise NotImplementedError @@ -33,6 +33,6 @@ def create_led(self, pin_number: int, name: str): Creates an LED instance appropriate for this hardware. :param pin_number: GPIO pin number :param name: LED name - :return: LED instance (LED or MockLED) + :return: LED instance appropriate for this hardware """ raise NotImplementedError diff --git a/src/hardware/PicoHardwre.py b/src/hardware/PicoHardwre.py index 978bb8b..878520c 100644 --- a/src/hardware/PicoHardwre.py +++ b/src/hardware/PicoHardwre.py @@ -32,10 +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_obj): - """Re-initializes the pin in place to clear PWM settings""" - pin_obj.init(self.Pin.OUT) - return pin_obj + def reset_pin(self, pin): + """Re-initializes the pin to clear PWM settings. machine.Pin accepts an + existing Pin object as the id, so cached LED pins are re-muxed in place.""" + return self.Pin(pin, self.Pin.OUT) 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 7dd8411..7883e04 100644 --- a/src/hardware/VirtualHardware.py +++ b/src/hardware/VirtualHardware.py @@ -1,90 +1,101 @@ -import src.hardware from src.hardware.Hardware import Hardware from src.hardware.Networking import Networking from src.pi.board_led import BoardLED +from src.pi.LED import LED -class VirtualHardware(Hardware): +class MockPin: """ - Virtual Hardware - Fake implementation so the webserver can run without a physical Raspberry Pi Pico connected to it. + Partial implementation of Pico Pin, only using the currently needed methods """ - class MockPin: - """ - Partial implementation of Pico Pin, only using the currently needed methods - """ - def __init__(self, num): - self.num = num + def __init__(self, num): + self.num = num - def on(self): - print(f"[SIM] Pin {self.num} ON") + def on(self): + print(f"[SIM] Pin {self.num} ON") - def off(self): - print(f"[SIM] Pin {self.num} OFF") + def off(self): + print(f"[SIM] Pin {self.num} OFF") - class MockPWM: - """ - Partial implementation of Pico PWM, only using the currently needed methods - """ - def __init__(self, p): - self.p = p +class MockPWM: + """ + Partial implementation of Pico PWM, only using the currently needed methods + """ - def freq(self, f): - pass + def __init__(self, p): + self.p = p - def duty_u16(self, d): - print(f"[SIM] PWM {self.p.num} @ {d}") + def freq(self, f): + pass - def deinit(self): - print(f"[SIM] PWM {self.p.num} De-initialized") + def duty_u16(self, d): + print(f"[SIM] PWM {self.p.num} @ {d}") - class NoOpNetworking(Networking): - """ - Networking implementation that does nothing - """ + def deinit(self): + print(f"[SIM] PWM {self.p.num} De-initialized") - def __init__(self): - pass - async def connect_to_wifi(self, ssid: str, password: str, attempts=10) -> str: - return "123.123.123.123" +class MockLED(LED): + """ + LED implementation for simulation that prints actions to console. + Used when running with VirtualHardware for testing without physical hardware. + """ + + def on(self): + """Turns on the LED with simulation output""" + print(f"[SIM] LED '{self._led_name}' (Pin {self._pin.num}) ON") + self._pin.on() + + def off(self): + """Turns off the LED with simulation output""" + print(f"[SIM] LED '{self._led_name}' (Pin {self._pin.num}) OFF") + self._pin.off() - def configure_host(self, host_name: str): - pass - class MockBoardLED(BoardLED): - """ - Fake implementation of the onboard led. - """ +class NoOpNetworking(Networking): + """ + Networking implementation that does nothing + """ - def __init__(self): - self._pin = src.hardware.VirtualHardware.MockPin(1) - self._led_name = "Mock Board LED" + def __init__(self): + pass + + async def connect_to_wifi(self, ssid: str, password: str, attempts=10) -> str: + return "123.123.123.123" + + def configure_host(self, host_name: str): + pass + + +class VirtualHardware(Hardware): + """ + Virtual Hardware + Fake implementation so the webserver can run without a physical Raspberry Pi Pico connected to it. + """ def __init__(self): - self.pin = self.MockPin - self.pwm = self.MockPWM + # Cached like PicoHardware, so both implementations behave the same + self._networking = NoOpNetworking() + self._board_led = BoardLED(MockPin("LED")) def get_pin(self, pin_num, mode="OUT"): - return self.pin(pin_num) + return MockPin(pin_num) def get_pwm(self, pin_obj): - return self.pwm(pin_obj) + return MockPWM(pin_obj) def board_led(self) -> BoardLED: - return self.MockBoardLED() + return self._board_led def networking(self) -> Networking: - return self.NoOpNetworking() + return self._networking - def reset_pin(self, pin_obj): - print(f"[SIM] Pin {pin_obj.num} reset to standard GPIO") - return pin_obj + def reset_pin(self, pin): + print(f"[SIM] Pin {pin.num} reset to standard GPIO") + return pin def create_led(self, pin_number: int, name: str): """Creates a mock LED for simulation""" - from src.pi.LED import MockLED - pin = self.get_pin(pin_number, mode="OUT") - return MockLED(pin, name) + return MockLED(self.get_pin(pin_number), name) diff --git a/src/hardware/__init__.py b/src/hardware/__init__.py index 29d8e73..7cf8804 100644 --- a/src/hardware/__init__.py +++ b/src/hardware/__init__.py @@ -1,14 +1,17 @@ import sys from src.hardware.Hardware import Hardware -from src.hardware.PicoHardwre import PicoHardware -from src.hardware.VirtualHardware import VirtualHardware def get_hardware() -> Hardware: """ :return: the appropriate hardware + + Imports are done lazily so the Pico never loads the mock implementation + (and the laptop never touches machine/network). """ if sys.platform == 'rp2': + from src.hardware.PicoHardwre import PicoHardware return PicoHardware() + from src.hardware.VirtualHardware import VirtualHardware return VirtualHardware() diff --git a/src/pi/LED.py b/src/pi/LED.py index 9c72fe6..7c611a1 100644 --- a/src/pi/LED.py +++ b/src/pi/LED.py @@ -42,26 +42,3 @@ def pin(self): :return: The underlying Raspberry Pi Pico Pin of the LED. """ 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): - """ - LED implementation for simulation that prints actions to console. - Used when running with VirtualHardware for testing without physical hardware. - """ - - def on(self): - """Turns on the LED with simulation output""" - print(f"[SIM] LED '{self._led_name}' (Pin {self._pin.num}) ON") - self._pin.on() - - def off(self): - """Turns off the LED with simulation output""" - print(f"[SIM] LED '{self._led_name}' (Pin {self._pin.num}) OFF") - self._pin.off() diff --git a/src/pi/board_led.py b/src/pi/board_led.py index 068eeed..d24a30e 100644 --- a/src/pi/board_led.py +++ b/src/pi/board_led.py @@ -6,8 +6,12 @@ class BoardLED(LED): Special Representation of the onboard Pico LED """ - def __init__(self): # pylint # pylint: disable=(super-init-not-called - from machine import Pin - - self._pin: Pin = Pin("LED", Pin.OUT) - self._led_name = "Board LED" + def __init__(self, pin=None): + """ + :param pin: The pin driving the onboard LED. Defaults to the real Pico pin; + virtual hardware injects a mock instead. + """ + if pin is None: + from machine import Pin + pin = Pin("LED", Pin.OUT) + super().__init__(pin, "Board LED") diff --git a/src/pi/disabled_LED.py b/src/pi/disabled_LED.py index afcd59c..57e9bf7 100644 --- a/src/pi/disabled_LED.py +++ b/src/pi/disabled_LED.py @@ -26,6 +26,3 @@ 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 8f00ed3..d1c5b42 100644 --- a/src/pi/led_effect.py +++ b/src/pi/led_effect.py @@ -1,14 +1,19 @@ import asyncio -import src.hardware -from src.pi import LED +from src.pi.LED import LED class LEDEffects: """ A collection of effects a LED can do. Things such as pulsate, breath, flash, etc. + PWM-based effects drive the pins through the hardware supplied at construction, + so the same effects run against real or virtual hardware. """ + def __init__(self, hardware): + from src.hardware.Hardware import Hardware + self.hardware: Hardware = hardware + @staticmethod async def blink(led: LED) -> None: """ @@ -33,36 +38,19 @@ async def fire(led: LED) -> None: await asyncio.sleep(.5) led.off() - @staticmethod - async def charge_fire(led: LED, charge_speed: int = 1) -> None: + async def charge_fire(self, led: LED, charge_speed: int = 1) -> None: """ A simple charging of a shot """ - await LEDEffects.brighten(led, start_percent=0, end_percent=75, speed=charge_speed) + await self.brighten(led, start_percent=0, end_percent=75, speed=charge_speed) led.off() await asyncio.sleep(0.5) - # LEDEffects.brighten(led, start_percent=75, end_percent=100, speed=1) + # self.brighten(led, start_percent=75, end_percent=100, speed=1) led.on() 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: + async def brighten(self, led: LED, start_percent: int = 0, end_percent: int = 100, speed: int = 10) -> None: """ Starting from start_pct goes to end_pct over the course of speed, brightens led :param led: @@ -71,52 +59,47 @@ async def brighten(led: LED, start_percent: int = 0, end_percent: int = 100, spe :param speed: :return: """ - timing = LEDEffects._step_timing(start_percent, end_percent, speed) - if not led.enabled() or timing is None: - return - 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) - 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())) + await self.brighten_all([led], start_percent, end_percent, speed) - @staticmethod - async def brighten_all(leds: list[LED], start_percent: int = 0, end_percent: int = 100, speed: int = 10) -> None: + async def brighten_all(self, leds: list[LED], start_percent: int = 0, end_percent: int = 100, speed: int = 10) -> None: """ - The current banshee amount of leds passed in causes it to I guess stack overflow and silently crash - 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. + Brightens all the given LEDs together from start_percent to end_percent over speed. + Note: on real hardware, driving the full banshee LED count at once has silently crashed + around 30% in the past — PWM here is not fully understood yet. """ - timing = LEDEffects._step_timing(start_percent, end_percent, speed) - if timing is None: + step_rate = 10 + + overall_change = end_percent - start_percent + if overall_change <= 0: + return + interval = overall_change / step_rate + sleep_time = speed / interval + + enabled_leds = [led for led in leds if led.enabled()] + if not enabled_leds: return - 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) - try: + for led in enabled_leds: + pwm = self.hardware.get_pwm(led.pin()) + pwm.freq(1000) + pwms.append(pwm) + 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) + + # range() stops short of end_percent, so land on the final brightness explicitly + duty = int((end_percent / 100) * 65_535) + for pwm in pwms: + pwm.duty_u16(duty) finally: + # Runs even when a show is cancelled mid-ramp: release the PWM and re-mux the + # pin back to plain GPIO, otherwise the LED's cached Pin stops responding. for pwm in pwms: pwm.deinit() - for led in active_leds: - led.set_pin(src.hardware.get_hardware().reset_pin(led.pin())) + for led in enabled_leds: + self.hardware.reset_pin(led.pin()) diff --git a/src/server/RouteDecorator.py b/src/server/RouteDecorator.py index 3a6ced0..8448d3c 100644 --- a/src/server/RouteDecorator.py +++ b/src/server/RouteDecorator.py @@ -1,49 +1,67 @@ import asyncio -def is_lightshow_running(gunpla, manager_attr="current_task") -> bool: +class LightshowManager: """ - :return: True if a lightshow task is currently tracked and hasn't finished yet + Owns the single running lightshow task for a gunpla. start() and stop() are + serialized with a lock so overlapping requests can't orphan a running show + or leave two shows fighting over the same LEDs. """ - existing_task = getattr(gunpla, manager_attr, None) - return existing_task is not None and not existing_task.done() + def __init__(self, gunpla): + self.gunpla = gunpla + self._task = None + self._lock = asyncio.Lock() -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) - if is_lightshow_running(gunpla, manager_attr): - existing_task.cancel() + 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, func) -> None: + """ + Cancels any running show (turning the LEDs off), then starts func as the + tracked lightshow task. + """ + async with self._lock: + if await self._cancel_current(): + self.gunpla.all_off() + self._task = asyncio.create_task(func()) + + async def stop(self) -> bool: + """ + Cancels any running lightshow and waits for it to clean up. + :return: True if a running show was cancelled, False if nothing was running + """ + async with self._lock: + return await self._cancel_current() + + async def _cancel_current(self) -> bool: + task = self._task + self._task = None + if task is None or task.done(): + return False + task.cancel() try: - await existing_task # Wait for cleanup + await task # Wait for cleanup except asyncio.CancelledError: pass - setattr(gunpla, manager_attr, None) + except Exception as e: + # A show that died on its own shouldn't fail the cancelling request + print(f"Lightshow ended with error: {e}") return True - setattr(gunpla, manager_attr, None) - return False -def lightshow_route(gunpla, manager_attr="current_task"): +def lightshow_route(manager: LightshowManager): """ A decorator factory that handles task management and standardized HTTP responses. """ def decorator(func): async def wrapper(request, *args, **kwargs): - 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) + # Replaces any running lightshow with this one. + await manager.start(func) # Return common HTTP response that the show started. return { diff --git a/src/server/Wrappers.py b/src/server/Wrappers.py index ff1f38e..2c268d3 100644 --- a/src/server/Wrappers.py +++ b/src/server/Wrappers.py @@ -23,16 +23,16 @@ async def wrapper(*args, **kwargs): return wrapper -def create_show_handler(func, gundam_instance): +def create_show_handler(func, show_manager): """ Helper that when given a function, wraps it as a lighthow_route and safe_execution. - :param gundam_instance: + :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(gunpla=gundam_instance) + @lightshow_route(show_manager) @safe_execution async def show_handler(): return await func() diff --git a/src/server/webserver.py b/src/server/webserver.py index 6d4e0cd..6fd92d5 100644 --- a/src/server/webserver.py +++ b/src/server/webserver.py @@ -6,10 +6,19 @@ 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, is_lightshow_running +from src.server.RouteDecorator import LightshowManager from src.server.Wrappers import create_show_handler, safe_execution +def run_server(settings: dict, hardware: Hardware) -> None: + """ + Composes and runs the webserver. The single entry point shared by the + on-device runner (main.py) and the local test server (tests/LocalServerTest.py). + """ + webserver = WebServer(settings, hardware) + asyncio.run(webserver.run()) + + class WebServer: """ Webserver that manages API routes and web pages for the Gunpla @@ -21,6 +30,7 @@ def __init__(self, configuration: dict, hardware: Hardware): # Instantiate the model class with hardware self.gundam: GenericGundam = configuration['model'](hardware) self.hardware: Hardware = hardware + self.show_manager = LightshowManager(self.gundam) Template.initialize(template_dir='src/templates') @safe_execution @@ -87,13 +97,16 @@ async def run(self): self._add_routes() - await self.app.start_server(host='0.0.0.0', port=80, debug=True) + await self.app.start_server( + host=self.settings.get('host', '0.0.0.0'), + 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 is_lightshow_running(self.gundam) + return self.show_manager.is_running() def _add_routes(self): """ @@ -121,7 +134,7 @@ 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.gundam)) + self.app.route(path)(create_show_handler(method_func, self.show_manager)) @self.app.route("/lightshow/stop") @safe_execution @@ -129,9 +142,8 @@ async def stop_lightshow(request): """ Stops any currently running lightshow task on the gundam instance. """ - async with self.gundam.lightshow_lock: - stopped = await cancel_lightshow(self.gundam) - self.gundam.all_off() + stopped = await self.show_manager.stop() + self.gundam.all_off() if stopped: return {"status": "stopped", "message": "Lightshow terminated"}, 200 diff --git a/src/test.py b/src/test.py index eed0021..62839b1 100644 --- a/src/test.py +++ b/src/test.py @@ -4,13 +4,12 @@ import time -from machine import Pin - def main(): """ Blinks the onboard Raspberry Pi Pico W LED several times. """ + from machine import Pin led = Pin("LED", Pin.OUT) led.on() time.sleep(0.5) diff --git a/tests/LocalServerTest.py b/tests/LocalServerTest.py index c9f5e67..34a078b 100644 --- a/tests/LocalServerTest.py +++ b/tests/LocalServerTest.py @@ -4,7 +4,7 @@ from src.gunpla.generic_gundam import GenericGundam from src.hardware.VirtualHardware import VirtualHardware -from src.server.webserver import WebServer +from src.server.webserver import run_server """ Sanity check class to run the webserver in local mode when a Raspberry pi is not needed. @@ -63,11 +63,11 @@ def main(): "ssid": "wifi", "password": 'wifi-pass', "hostname": 'virgo', - "model": lambda hardware: MobileDoll(hardware, model_config) + "model": lambda hardware: MobileDoll(hardware, model_config), + "port": 8080 # unprivileged, so no root needed on the laptop } - webserver = WebServer(test_settings, VirtualHardware()) - asyncio.run(webserver.run()) + run_server(test_settings, VirtualHardware()) if __name__ == "__main__": diff --git a/tests/test_virtual_hardware.py b/tests/test_virtual_hardware.py new file mode 100644 index 0000000..417a652 --- /dev/null +++ b/tests/test_virtual_hardware.py @@ -0,0 +1,118 @@ +""" +Tests that run the production code against VirtualHardware — no Pico needed. +""" +import asyncio + +from src.gunpla.generic_gundam import GenericGundam +from src.hardware import get_hardware +from src.hardware.VirtualHardware import VirtualHardware +from src.pi.disabled_LED import DisabledLED +from src.pi.led_effect import LEDEffects +from src.server.RouteDecorator import LightshowManager, lightshow_route + + +def test_get_hardware_selects_virtual_off_device(): + assert isinstance(get_hardware(), VirtualHardware) + + +def test_board_led_is_the_real_class_with_a_mock_pin(): + hardware = VirtualHardware() + led = hardware.board_led() + assert led.name() == "Board LED" + assert led.enabled() + led.on() + led.off() + + +def test_board_led_and_networking_are_cached_like_pico_hardware(): + hardware = VirtualHardware() + assert hardware.board_led() is hardware.board_led() + assert hardware.networking() is hardware.networking() + + +def test_brighten_drives_pwm_through_injected_hardware(): + hardware = VirtualHardware() + led = hardware.create_led(5, "head") + asyncio.run(LEDEffects(hardware).brighten(led, speed=0)) + + +def test_brighten_with_equal_percentages_is_a_noop(): + hardware = VirtualHardware() + led = hardware.create_led(5, "head") + asyncio.run(LEDEffects(hardware).brighten(led, start_percent=50, end_percent=50)) + + +def test_effects_skip_disabled_leds(): + hardware = VirtualHardware() + effects = LEDEffects(hardware) + asyncio.run(effects.brighten(DisabledLED("ghost"))) + asyncio.run(effects.brighten_all([hardware.create_led(1, "a"), DisabledLED("ghost")], speed=0)) + + +def test_gundam_caches_led_objects(): + gundam = GenericGundam(VirtualHardware()) + assert gundam._get_led_from_name("head") is gundam._get_led_from_name("head") + + +def test_all_on_and_off_run_on_virtual_hardware(): + gundam = GenericGundam(VirtualHardware()) + gundam.all_on() + gundam.all_off() + + +class FakeGunpla: + def all_off(self): + pass + + +def test_lightshow_lifecycle_tracks_and_clears_task(): + manager = LightshowManager(FakeGunpla()) + + async def scenario(): + started = asyncio.Event() + + async def show(): + started.set() + await asyncio.sleep(60) + + handler = lightshow_route(manager)(show) + _, status = await handler(None) + assert status == 202 + await started.wait() + assert manager.is_running() + + assert await manager.stop() is True + assert not manager.is_running() + assert await manager.stop() is False + + asyncio.run(scenario()) + + +def test_overlapping_start_requests_do_not_orphan_a_show(): + manager = LightshowManager(FakeGunpla()) + running = [] + + def make_show(name): + async def show(): + running.append(name) + try: + await asyncio.sleep(60) + finally: + running.remove(name) + return show + + async def scenario(): + await manager.start(make_show("first")) + await asyncio.sleep(0) + # Two replacement requests land while the first show is still being + # cancelled — without serialization one of these orphaned a show. + await asyncio.gather(manager.start(make_show("second")), manager.start(make_show("third"))) + await asyncio.sleep(0) + assert running == ["third"] + assert manager.is_running() + + await manager.stop() + assert running == [] + assert not manager.is_running() + + asyncio.run(scenario())