Merge pull request #475 from JnyJny/features/refactor-effects-taskmixin

Refactor effects system with TaskableMixin integration

Author: JnyJny

src/busylight/__main__.py

@@ -8,7 +8,6 @@
 from . import __version__
 from .busyserve import busyserve_cli
 from .callbacks import string_to_scaled_color
-from .effects import Effects
 from .global_options import GlobalOptions
 from .manager import LightManager
 from .speed import Speed

src/busylight/effects/blink.py

@@ -1,40 +1,43 @@
-"""change a light between two colors with a short interval."""
+"""Change a light between two colors with a short interval."""
 
-from typing import List, Optional, Tuple
+from typing import TYPE_CHECKING
+
+from busylight_core.mixins.taskable import TaskPriority
 
 from .effect import BaseEffect
 
+if TYPE_CHECKING:
+    from busylight_core import Light
+
 
 class Blink(BaseEffect):
     def __init__(
         self,
-        on_color: Tuple[int, int, int],
-        duty_cycle: float,
-        off_color: Optional[Tuple[int, int, int]] = None,
+        on_color: tuple[int, int, int],
+        off_color: tuple[int, int, int] | None = None,
         count: int = 0,
     ) -> None:
-        """This effect turns a light on and off with the specified color(s),
-        pausing for `duty_cycle` seconds in between each operation. If count is
-        given and greater than zero, the light will blink count times.
-
-        :param on_color: Tuple[int,int,int]
-        :param duty_cycle: float
-        :param off_color: Tuple[int,int,int] defaults to black.
-        :param count: int defaults to 0, indicating no limit.
+        """This effect alternates between on_color and off_color.
+
+        If count is given and greater than zero, the light will blink
+        count times.
+
+        :param on_color: RGB tuple for the "on" state
+        :param off_color: RGB tuple for the "off" state, defaults to black
+        :param count: Number of blink cycles, 0 means infinite
         """
         self.on_color = on_color
         self.off_color = off_color or (0, 0, 0)
-        self.duty_cycle = duty_cycle
         self.count = count
+        self.priority = TaskPriority.NORMAL
 
     def __repr__(self) -> str:
-        return f"{self.name}(on_color={self.on_color!r}, duty_cycle={self.duty_cycle!r}, off_color={self.off_color!r})"
+        return f"{self.name}(on_color={self.on_color!r}, off_color={self.off_color!r})"
+
+    @property
+    def colors(self) -> list[tuple[int, int, int]]:
+        return [self.on_color, self.off_color]
 
     @property
-    def colors(self) -> List[Tuple[int, int, int]]:
-        try:
-            return self._colors
-        except AttributeError:
-            pass
-        self._colors: List[Tuple[int, int, int]] = [self.on_color, self.off_color]
-        return self._colors
+    def default_interval(self) -> float:
+        return 0.5

src/busylight/effects/effect.py

@@ -1,19 +1,22 @@
-""" """
+"""Base class for all effects."""
 
 import abc
 import asyncio
-from functools import lru_cache
+from functools import cache
 from itertools import cycle, islice
-from typing import Dict, List, Tuple
+from typing import TYPE_CHECKING
 
-from busylight_core import Light
+from busylight_core.mixins.taskable import TaskPriority
 from loguru import logger
 
+if TYPE_CHECKING:
+    from busylight_core import Light
+
 
 class BaseEffect(abc.ABC):
     @classmethod
-    @lru_cache
-    def subclasses(cls) -> Dict[str, "BaseEffect"]:
+    @cache
+    def subclasses(cls) -> dict[str, "BaseEffect"]:
         """Returns a dictionary of Effect subclasses, keyed by name."""
         subclasses = {}
         if cls is BaseEffect:
@@ -33,7 +36,7 @@ def subclasses(cls) -> Dict[str, "BaseEffect"]:
 
     @classmethod
     def for_name(cls, name: str) -> "BaseEffect":
-        """Finds an effect subclass with the given name.
+        """Return an effect subclass with the given name.
 
         :param name: str
         :return: BaseEffect or subclass
@@ -47,26 +50,22 @@ def for_name(cls, name: str) -> "BaseEffect":
         except KeyError:
             raise ValueError(f"Unknown effect {name}") from None
 
+    @classmethod
+    def effects(cls) -> list[str]:
+        """Return a list of effect names."""
+        return list(cls.subclasses().keys())
+
     def __repr__(self) -> str:
         return f"{self.name}(...)"
 
     def __str__(self) -> str:
-        return f"{self.name} duty_cycle={self.duty_cycle}"
+        return f"{self.name} count={self.count}"
 
     @property
     def name(self) -> str:
         """The name of this effect."""
         return self.__class__.__name__
 
-    @property
-    def duty_cycle(self) -> float:
-        """Interval in seconds for current frame of the effect to be displayed."""
-        return getattr(self, "_duty_cycle", 0)
-
-    @duty_cycle.setter
-    def duty_cycle(self, seconds: float) -> None:
-        self._duty_cycle = seconds
-
     @property
     def count(self) -> int:
         """Number of cycles to run the effect.
@@ -79,23 +78,50 @@ def count(self) -> int:
     def count(self, count: int) -> None:
         self._count = int(count)
 
+    @property
+    def priority(self) -> TaskPriority:
+        """Task priority for this effect."""
+        return getattr(self, "_priority", TaskPriority.NORMAL)
+
+    @priority.setter
+    def priority(self, priority: TaskPriority) -> None:
+        self._priority = priority
+
     @property
     @abc.abstractmethod
-    def colors(self) -> List[Tuple[int, int, int]]:
+    def colors(self) -> list[tuple[int, int, int]]:
         """A list of color tuples."""
 
-    async def __call__(self, light: Light) -> None:
-        """Apply this effect to the given light.
+    @property
+    @abc.abstractmethod
+    def default_interval(self) -> float:
+        """Default interval between color changes in seconds."""
+
+    async def execute(self, light: "Light", interval: float | None = None) -> None:
+        """Execute this effect on the given light.
 
-        :param light: Light
+        This method runs the full effect cycle similar to the original
+        generator-based approach but as a single long-running async function.
+
+        :param light: Light instance with TaskableMixin capabilities
+        :param interval: Override default interval between color changes
         """
+        sleep_interval = interval if interval is not None else self.default_interval
+        
         if self.count > 0:
             cycle_count = self.count * len(self.colors)
+            color_iterator = islice(cycle(self.colors), cycle_count)
         else:
-            cycle_count = None
-
-        for color in islice(cycle(self.colors), cycle_count):
-            light.on(color)
-            await asyncio.sleep(self.duty_cycle)
+            color_iterator = cycle(self.colors)
 
-        light.off()
+        try:
+            for color in color_iterator:
+                light.on(color)
+                await asyncio.sleep(sleep_interval)
+        finally:
+            light.off()
+
+    def reset(self) -> None:
+        """Reset the effect's internal state."""
+        if hasattr(self, "_color_cycle"):
+            delattr(self, "_color_cycle")

src/busylight/effects/gradient.py

@@ -1,53 +1,61 @@
-"""a smooth color gradient for a given color."""
+"""A smooth color gradient for a given color."""
 
-from typing import List, Tuple
+from typing import TYPE_CHECKING
+
+from busylight_core.mixins.taskable import TaskPriority
 
 from .effect import BaseEffect
 
+if TYPE_CHECKING:
+    from busylight_core import Light
+
 
 class Gradient(BaseEffect):
-    """This effect will produce a color range from black to the given
-    color and then back to black again with the given number of steps
-    between off and on. If count is given and is greater than zero the
-    light will cycle thru the sequence count times.
+    """This effect produces a smooth color gradient from black to the given
+    color and then back to black again with the given number of steps.
+    If count is given and is greater than zero the light will cycle through
+    the sequence count times.
     """
 
     def __init__(
         self,
-        color: Tuple[int, int, int],
-        duty_cycle: float,
+        color: tuple[int, int, int],
         step: int = 1,
+        step_max: int = 255,
         count: int = 0,
     ) -> None:
-        """:param color: Tuple[int,int,int]
-        :param duty_cycle: float
-        :param step: int defaults to 1.
-        :param count: int defaults to 0, indicating no limit.
+        """Initialize gradient effect.
+
+        :param color: Target RGB color for the gradient
+        :param step: Step size for gradient calculation
+        :param step_max: Maximum step value, determines gradient smoothness
+        :param count: Number of gradient cycles, 0 means infinite
         """
         self.color = color
-        self.duty_cycle = duty_cycle
-        # XXX need to choose steps that make sense for scaled colors
-        #     where the max(color) << 255
-        self.step = max(0, min(step, 255))
+        self.step = max(1, min(step, step_max))
+        self.step_max = step_max
         self.count = count
+        self.priority = TaskPriority.LOW
 
     @property
-    def colors(self) -> List[Tuple[int, int, int]]:
-        try:
+    def colors(self) -> list[tuple[int, int, int]]:
+        if hasattr(self, "_colors"):
             return self._colors
-        except AttributeError:
-            pass
 
         red, green, blue = self.color
-
         colors = []
-        for i in range(1, 256, self.step):
-            scale = i / 255
+
+        for i in range(self.step, self.step_max + 1, self.step):
+            scale = i / self.step_max
             r = round(scale * red)
             g = round(scale * green)
             b = round(scale * blue)
             colors.append((r, g, b))
 
-        self._colors: List[Tuple[int, int, int]] = colors + list(reversed(colors[:-1]))
-
+        ramp_down = list(reversed(colors[:-1]))
+        self._colors: list[tuple[int, int, int]] = colors + ramp_down
         return self._colors
+
+    @property
+    def default_interval(self) -> float:
+        return 0.1