dune-weaver/modules/led/led_interface.py

"""
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
from modules.led.board_led_controller import BoardLEDController

LEDProviderType = Literal["wled", "board", "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 == "board":
            # The table's own LED ring, driven by the FluidNC firmware.
            self._controller = BoardLEDController()

    @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 == "board":
            self._controller = BoardLEDController()
        else:
            self._controller = None

    # NOTE: for the "board" provider the effect_* transition hooks below fall
    # through to False on purpose — the firmware switches run/idle effects
    # itself ($LED/RunEffect / $LED/IdleEffect); the host must not fight it.

    def effect_loading(self) -> bool:
        """Show loading effect"""
        if not self.is_configured:
            return False

        if self.provider == "wled":
            return wled_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)
        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)
        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)
        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 == "board":
            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)