""" Unified LED interface for different LED control systems Provides a common abstraction layer for pattern manager integration. """ import asyncio from typing import Optional, Literal from modules.led.led_controller import LEDController, effect_loading as wled_loading, effect_idle as wled_idle, effect_connected as wled_connected, effect_playing as wled_playing # Try to import DW LED controller - it requires RPi-specific dependencies try: from modules.led.dw_led_controller import DWLEDController, effect_loading as dw_led_loading, effect_idle as dw_led_idle, effect_connected as dw_led_connected, effect_playing as dw_led_playing DW_LEDS_AVAILABLE = True except ImportError: # Running on non-RPi platform - DW LEDs not available DWLEDController = None dw_led_loading = None dw_led_idle = None dw_led_connected = None dw_led_playing = None DW_LEDS_AVAILABLE = False LEDProviderType = Literal["wled", "dw_leds", "none"] class LEDInterface: """ Unified interface for LED control that works with multiple backends. Automatically delegates to the appropriate controller based on configuration. """ def __init__(self, provider: LEDProviderType = "none", ip_address: Optional[str] = None, num_leds: Optional[int] = None, gpio_pin: Optional[int] = None, pixel_order: Optional[str] = None, brightness: Optional[float] = None, speed: Optional[int] = None, intensity: Optional[int] = None): self.provider = provider self._controller = None if provider == "wled" and ip_address: self._controller = LEDController(ip_address) elif provider == "dw_leds": if not DW_LEDS_AVAILABLE: raise ImportError("DW LED controller requires Raspberry Pi GPIO libraries. Install with: pip install -r requirements.txt") # DW LEDs uses local GPIO, no IP needed num_leds = num_leds or 60 gpio_pin = gpio_pin or 12 pixel_order = pixel_order or "GRB" brightness = brightness if brightness is not None else 0.35 speed = speed if speed is not None else 128 intensity = intensity if intensity is not None else 128 self._controller = DWLEDController(num_leds, gpio_pin, brightness, pixel_order=pixel_order, speed=speed, intensity=intensity) @property def is_configured(self) -> bool: """Check if LED controller is configured""" return self._controller is not None def update_config(self, provider: LEDProviderType, ip_address: Optional[str] = None, num_leds: Optional[int] = None, gpio_pin: Optional[int] = None, pixel_order: Optional[str] = None, brightness: Optional[float] = None, speed: Optional[int] = None, intensity: Optional[int] = None): """Update LED provider configuration""" self.provider = provider # Stop existing controller if switching providers if self._controller and hasattr(self._controller, 'stop'): try: self._controller.stop() except: pass if provider == "wled" and ip_address: self._controller = LEDController(ip_address) elif provider == "dw_leds": if not DW_LEDS_AVAILABLE: raise ImportError("DW LED controller requires Raspberry Pi GPIO libraries. Install with: pip install -r requirements.txt") num_leds = num_leds or 60 gpio_pin = gpio_pin or 12 pixel_order = pixel_order or "GRB" brightness = brightness if brightness is not None else 0.35 speed = speed if speed is not None else 128 intensity = intensity if intensity is not None else 128 self._controller = DWLEDController(num_leds, gpio_pin, brightness, pixel_order=pixel_order, speed=speed, intensity=intensity) else: self._controller = None def effect_loading(self) -> bool: """Show loading effect""" if not self.is_configured: return False if self.provider == "wled": return wled_loading(self._controller) elif self.provider == "dw_leds": return dw_led_loading(self._controller) return False def effect_idle(self, effect_name: Optional[str] = None) -> bool: """Show idle effect""" if not self.is_configured: return False if self.provider == "wled": return wled_idle(self._controller) elif self.provider == "dw_leds": return dw_led_idle(self._controller, effect_name) return False def effect_connected(self) -> bool: """Show connected effect""" if not self.is_configured: return False if self.provider == "wled": return wled_connected(self._controller) elif self.provider == "dw_leds": return dw_led_connected(self._controller) return False def effect_playing(self, effect_name: Optional[str] = None) -> bool: """Show playing effect""" if not self.is_configured: return False if self.provider == "wled": return wled_playing(self._controller) elif self.provider == "dw_leds": return dw_led_playing(self._controller, effect_name) return False def set_power(self, state: int) -> dict: """Set power state (0=Off, 1=On, 2=Toggle)""" if not self.is_configured: return {"connected": False, "message": "No LED controller configured"} return self._controller.set_power(state) def check_status(self) -> dict: """Check controller status""" if not self.is_configured: return {"connected": False, "message": "No LED controller configured"} if self.provider == "wled": return self._controller.check_wled_status() elif self.provider == "dw_leds": return self._controller.check_status() return {"connected": False, "message": "Unknown provider"} def get_controller(self): """Get the underlying controller instance (for advanced usage)""" return self._controller # Async versions of methods for non-blocking calls from async context # These use asyncio.to_thread() to avoid blocking the event loop async def effect_loading_async(self) -> bool: """Show loading effect (non-blocking)""" return await asyncio.to_thread(self.effect_loading) async def effect_idle_async(self, effect_name: Optional[str] = None) -> bool: """Show idle effect (non-blocking)""" return await asyncio.to_thread(self.effect_idle, effect_name) async def effect_connected_async(self) -> bool: """Show connected effect (non-blocking)""" return await asyncio.to_thread(self.effect_connected) async def effect_playing_async(self, effect_name: Optional[str] = None) -> bool: """Show playing effect (non-blocking)""" return await asyncio.to_thread(self.effect_playing, effect_name) async def set_power_async(self, state: int) -> dict: """Set power state (non-blocking)""" return await asyncio.to_thread(self.set_power, state) async def check_status_async(self) -> dict: """Check controller status (non-blocking)""" return await asyncio.to_thread(self.check_status)