Merge pull request #480 from JnyJny/features/led-parameter-support

feat: add LED parameter support for multi-LED devices

Author: JnyJny

src/busylight/api/busylight_api.py

@@ -26,13 +26,11 @@
 [Source](https://github.com/JnyJny/busylight.git)
 """
 
-# FastAPI Security Scheme
 busylightapi_security = HTTPBasic()
 
 
 class BusylightAPI(FastAPI):
     def __init__(self):
-        # Get and save the debug flag
         debug = environ.get("BUSYLIGHT_DEBUG", False)
         logger.info(f"Debug: {debug}")
 
@@ -49,13 +47,11 @@ def __init__(self):
             self.username = None
             self.password = None
 
-        # Get and save the CORS Access-Control-Allow-Origin header
         logger.info(
             "Set up CORS Access-Control-Allow-Origin header, if environment variable BUSYLIGHT_API_CORS_ORIGINS_LIST is set.",
         )
         self.origins = json_loads(environ.get("BUSYLIGHT_API_CORS_ORIGINS_LIST", "[]"))
 
-        # Validate that BUSYLIGHT_API_CORS_ORIGINS_LIST is a list of strings
         if (not isinstance(self.origins, list)) or any(
             not isinstance(item, str) for item in self.origins
         ):
@@ -90,7 +86,6 @@ def lights(self) -> list[Light]:
         return self.controller.lights
 
     def update(self) -> None:
-        # Force refresh of lights in controller
         _ = self.controller.lights
 
     def release(self) -> None:
@@ -111,10 +106,6 @@ async def apply_effect(self, effect: Effects, light_id: int = None) -> None:
     def get(self, path: str, **kwargs) -> Callable:
         self.endpoints.append(path)
 
-        # CORS allowed origins (for the Access-Control-Allow-Origin header)
-        # are set through an environment variable BUSYLIGHT_API_CORS_ORIGINS_LIST
-        # e.g.: export BUSYLIGHT_API_CORS_ORIGINS_LIST='["http://localhost", "http://localhost:8080"]'
-        # (see https://fastapi.tiangolo.com/tutorial/cors/ for details)
         if self.origins:
             self.add_middleware(
                 CORSMiddleware,
@@ -141,8 +132,6 @@ def authenticate_user(
 busylightapi = BusylightAPI()
 
 
-## Startup & Shutdown
-##
 @busylightapi.on_event("startup")
 async def startup() -> None:
     busylightapi.update()
@@ -157,8 +146,6 @@ async def shutdown() -> None:
         logger.debug("problem during shutdown: {error}")
 
 
-## Exception Handlers
-##
 @busylightapi.exception_handler(LightUnavailableError)
 async def light_unavailable_handler(
     request: Request,
@@ -207,8 +194,6 @@ async def color_lookup_error_handler(
     )
 
 
-## Middleware Handlers
-##
 @busylightapi.middleware("http")
 async def light_manager_update(request: Request, call_next: Callable) -> Response:
     """Check for plug/unplug events and update the light manager."""
@@ -217,8 +202,6 @@ async def light_manager_update(request: Request, call_next: Callable) -> Respons
     return await call_next(request)
 
 
-## GET API Routes
-##
 @busylightapi.get("/", response_model=list[EndPoint])
 async def available_endpoints() -> list[dict[str, str]]:
     """API endpoint listing.
@@ -303,6 +286,7 @@ async def light_on(
     light_id: int = Path(..., title="Numeric light identifier", ge=0),
     color: str = "green",
     dim: float = 1.0,
+    led: int = 0,
 ) -> dict[str, Any]:
     """Turn on the specified light with the given `color`.
 
@@ -311,17 +295,22 @@ async def light_on(
 
     `color` can be a color name or a hexadecimal string e.g. "red",
     "#ff0000", "#f00", "0xff0000", "0xf00", "f00", "ff0000"
+
+    `led` parameter targets specific LEDs on multi-LED devices:
+    - 0 = all LEDs (default)
+    - 1+ = specific LED (1=first/top, 2=second/bottom, etc.)
     """
     rgb = parse_color_string(color, dim)
-    steady = Effects.for_name("steady")(rgb)
-    await busylightapi.apply_effect(steady, light_id)
+
+    busylightapi.controller.by_index(light_id).turn_on(rgb, led=led)
 
     return {
         "action": "on",
         "light_id": light_id,
         "color": color,
         "rgb": rgb,
         "dim": dim,
+        "led": led,
     }
 
 
@@ -332,22 +321,28 @@ async def light_on(
 async def lights_on(
     color: str = "green",
     dim: float = 1.0,
+    led: int = 0,
 ) -> dict[str, Any]:
     """Turn on all lights with the given `color`.
 
     `color` can be a color name or a hexadecimal string e.g. "red",
     "#ff0000", "#f00", "0xff0000", "0xf00", "f00", "ff0000"
+
+    `led` parameter targets specific LEDs on multi-LED devices:
+    - 0 = all LEDs (default)
+    - 1+ = specific LED (1=first/top, 2=second/bottom, etc.)
     """
     rgb = parse_color_string(color, dim)
-    steady = Effects.for_name("steady")(rgb)
-    await busylightapi.apply_effect(steady)
+
+    busylightapi.controller.all().turn_on(rgb, led=led)
 
     return {
         "action": "on",
         "light_id": "all",
         "color": color,
         "rgb": rgb,
         "dim": dim,
+        "led": led,
     }
 
 
@@ -397,6 +392,7 @@ async def blink_light(
     speed: Speed = Speed.Slow,
     dim: float = 1.0,
     count: int = 0,
+    led: int = 0,
 ) -> dict[str, Any]:
     """Start blinking the specified light: color and off.
 
@@ -407,12 +403,15 @@ async def blink_light(
     #ff0000, #f00, 0xff0000, 0xf00, f00, ff0000
 
     `count` is the number of times to blink the light.
+
+    `led` parameter targets specific LEDs on multi-LED devices:
+    - 0 = all LEDs (default)
+    - 1+ = specific LED (1=first/top, 2=second/bottom, etc.)
     """
     rgb = parse_color_string(color, dim)
 
-    # Use controller's fluent API
     selection = busylightapi.controller.by_index(light_id)
-    selection.blink(rgb, count=count, speed=speed.name.lower())
+    selection.blink(rgb, count=count, speed=speed.name.lower(), led=led)
 
     return {
         "action": "blink",
@@ -422,6 +421,7 @@ async def blink_light(
         "speed": speed,
         "dim": dim,
         "count": count,
+        "led": led,
     }
 
 
@@ -434,15 +434,19 @@ async def blink_lights(
     speed: Speed = Speed.Slow,
     dim: float = 1.0,
     count: int = 0,
+    led: int = 0,
 ) -> dict[str, Any]:
     """Start blinking all the lights: red and off
     <p>Note: lights will not be synchronized.</p>
+
+    `led` parameter targets specific LEDs on multi-LED devices:
+    - 0 = all LEDs (default)
+    - 1+ = specific LED (1=first/top, 2=second/bottom, etc.)
     """
     rgb = parse_color_string(color, dim)
 
-    # Use controller's fluent API
     selection = busylightapi.controller.all()
-    selection.blink(rgb, count=count, speed=speed.name.lower())
+    selection.blink(rgb, count=count, speed=speed.name.lower(), led=led)
 
     return {
         "action": "blink",
@@ -452,6 +456,7 @@ async def blink_lights(
         "speed": speed,
         "dim": dim,
         "count": count,
+        "led": led,
     }
 
 

src/busylight/controller.py

@@ -69,20 +69,29 @@ def __iter__(self) -> Iterator[Light]:
         """Return an iterator over the lights in this selection."""
         return iter(self.lights)
 
-    def turn_on(self, color: tuple[int, int, int]) -> LightSelection:
+    def turn_on(self, color: tuple[int, int, int], led: int = 0) -> LightSelection:
         """Turn on all lights in the selection with the specified color.
 
         :param color: RGB color tuple with values 0-255
+        :param led: Target LED index. 0 affects all LEDs, 1+ targets specific LEDs
+
+        For devices with multiple LEDs (like Blink1 mk2), use led parameter to control
+        individual LEDs. Single-LED devices ignore this parameter. LED indexing is
+        device-specific but typically: 0=all, 1=first/top, 2=second/bottom, etc.
 
         Example:
-            Turn lights red::
+            Turn all LEDs red::
+
+                selection.turn_on((255, 0, 0))  # led=0 default
+
+            Turn only top LED of Blink1 mk2 blue::
 
-                selection.turn_on((255, 0, 0))
+                selection.turn_on((0, 0, 255), led=1)
         """
 
         for light in self.lights:
             try:
-                light.on(color)
+                light.on(color, led=led)
             except LightUnavailableError as error:
                 logger.debug(f"Light unavailable during turn_on: {error}")
 
@@ -115,25 +124,37 @@ def blink(
         color: tuple[int, int, int],
         count: int = 0,
         speed: str = "slow",
+        led: int = 0,
     ) -> LightSelection:
         """Apply blink effect to all lights in the selection.
 
         :param color: RGB color tuple with values 0-255
         :param count: Number of blinks. 0 means infinite blinking
         :param speed: Blink speed - "slow", "medium", or "fast"
+        :param led: Target LED index. 0 affects all LEDs, 1+ targets specific LEDs
+
+        For devices with multiple LEDs, use led parameter to blink specific LEDs.
+        Single-LED devices ignore this parameter.
 
         Example:
-            Blink lights green 5 times at fast speed::
+            Blink all LEDs green 5 times at fast speed::
 
                 selection.blink((0, 255, 0), count=5, speed="fast")
+
+            Blink only bottom LED of Blink1 mk2::
+
+                selection.blink((255, 0, 0), led=2)
         """
         try:
             speed_obj = Speed(speed)
         except ValueError:
             speed_obj = Speed.slow
 
-        effect = Effects.for_name("blink")(color, count=count)
-        return self.apply_effect(effect, interval=speed_obj.duty_cycle)
+        if led == 0:
+            effect = Effects.for_name("blink")(color, count=count)
+            return self.apply_effect(effect, interval=speed_obj.duty_cycle)
+        else:
+            return self._apply_led_blink(color, count, speed_obj.duty_cycle, led)
 
     def apply_effect(
         self,
@@ -194,6 +215,68 @@ async def effect_task(target_light=light):
 
         return self
 
+    def _apply_led_blink(
+        self,
+        color: tuple[int, int, int],
+        count: int,
+        interval: float,
+        led: int,
+    ) -> LightSelection:
+        """Apply LED-aware blink effect to all lights in the selection.
+
+        :param color: RGB color tuple for the blink effect
+        :param count: Number of blinks, 0 means infinite
+        :param interval: Interval between blinks in seconds
+        :param led: Target LED index for multi-LED devices
+        """
+
+        async def led_blink_supervisor():
+            tasks = []
+            for light in self.lights:
+                light.cancel_tasks()
+
+                async def led_blink_task(target_light=light):
+                    blink_count = 0
+                    try:
+                        while count == 0 or blink_count < count:
+                            target_light.on(color, led=led)
+                            await asyncio.sleep(interval)
+                            target_light.on((0, 0, 0), led=led)
+                            await asyncio.sleep(interval)
+                            blink_count += 1
+                    finally:
+                        target_light.on((0, 0, 0), led=led)
+
+                task = light.add_task(
+                    name="led_blink",
+                    func=led_blink_task,
+                    priority=1,
+                    replace=True,
+                )
+                tasks.append(task)
+
+            if tasks:
+                if count > 0:
+                    await asyncio.wait(tasks)
+                else:
+                    try:
+                        await asyncio.wait(tasks)
+                    except KeyboardInterrupt:
+                        for task in tasks:
+                            task.cancel()
+                        raise
+
+        try:
+            loop = asyncio.get_running_loop()
+            asyncio.create_task(led_blink_supervisor())
+        except RuntimeError:
+            try:
+                asyncio.run(led_blink_supervisor())
+            except KeyboardInterrupt:
+                pass
+
+        return self
+
 
 class LightController:
     """Light controller with fluent interface for managing USB LED lights.

src/busylight/subcommands/blink.py

@@ -56,27 +56,41 @@ def blink_lights(
         show_default=True,
         help="Number of blinks. 0 means infinite.",
     ),
+    led: int = typer.Option(
+        0,
+        "--led",
+        help="Target LED index (0=all LEDs, 1+=specific LED for multi-LED devices)",
+        show_default=True,
+    ),
 ) -> None:
     """Create a blinking effect on selected lights.
 
     :param ctx: Typer context containing global options and controller
     :param color: Color specification for the blink effect
     :param speed: Blink speed (slow, medium, or fast)
     :param count: Number of blinks. 0 means infinite blinking
+    :param led: Target LED index for multi-LED devices
 
     This command applies a blinking effect to the selected lights using
-    the specified color, speed, and count. The effect runs asynchronously
-    and can be interrupted with Ctrl+C.
+    the specified color, speed, and count. For devices with multiple LEDs
+    (like Blink1 mk2), use --led to target specific LEDs.
 
+    The effect runs asynchronously and can be interrupted with Ctrl+C.
     For infinite blinking (count=0), the command will continue until
     interrupted. For finite counts, the command exits after the specified
     number of blinks complete.
+
+    Example:
+        Target specific LEDs::
+
+            busylight blink red --led 1 --count 3    # Top LED only
+            busylight blink green --led 2            # Bottom LED infinite
     """
     logger.info("Blinking lights with color: {}", color)
 
     try:
         selection = get_light_selection(ctx)
-        selection.blink(color, count=count, speed=speed.name.lower())
+        selection.blink(color, count=count, speed=speed.name.lower(), led=led)
     except (KeyboardInterrupt, TimeoutError):
         selection = get_light_selection(ctx)
         selection.turn_off()