diff --git a/examples/ClockKey.dui/layout.svg b/examples/ClockKey.dui/layout.svg new file mode 100644 index 0000000..59609a2 --- /dev/null +++ b/examples/ClockKey.dui/layout.svg @@ -0,0 +1,58 @@ + + + + + + + + + + + + + DeUX + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/examples/ClockKey.dui/manifest.yaml b/examples/ClockKey.dui/manifest.yaml new file mode 100644 index 0000000..e1a5266 --- /dev/null +++ b/examples/ClockKey.dui/manifest.yaml @@ -0,0 +1,54 @@ +name: ClockKey +type: Key +version: 1 +description: Analog Clock +author: Graphras.com +license: Apache-2.0 +category: utilities +layout: layout.svg + +bindings: + second_hand: + type: transform + node: second_hand + default: 0 + transforms: + - kind: rotate + from: 00 + to: 360 + origin: 0 0 + + minute_hand: + type: transform + node: minute_hand + default: 0 + transforms: + - kind: rotate + from: 00 + to: 360 + origin: 0 0 + + hour_hand: + type: transform + node: hour_hand + default: 0 + transforms: + - kind: rotate + from: 0 + to: 360 + origin: 0 0 + +events: + - name: click + source: key_press_release + max_duration_ms: 300 + + - name: hold + source: key_hold + hold_ms: 350 + + - name: press + source: key_press + + - name: release + source: key_release diff --git a/examples/streamdeck.py b/examples/streamdeck.py index d0a8f11..8957b20 100644 --- a/examples/streamdeck.py +++ b/examples/streamdeck.py @@ -74,6 +74,7 @@ import contextlib import datetime import logging +from collections.abc import Awaitable, Callable from pathlib import Path from typing import Any @@ -96,6 +97,7 @@ DeviceInfo, DuiCard, DuiKey, + KeyController, Theme, add_dui_path, ) @@ -861,6 +863,206 @@ async def on_detach(self) -> None: """Stop the background drift simulator.""" await self._svc.stop() +class ClockController(KeyController): + """Analog clock key -- ticking hour, minute, and second hands driven by system time. + + Loads ``ClockKey.dui`` and updates three transform bindings, + ``hour_hand``, ``minute_hand``, and ``second_hand``, with rotation + angles in degrees derived from the local system clock. All three + bindings are declared in the manifest as ``rotate`` transforms whose + ``from``/``to`` span the full ``0 -- 360`` degree range, so the + controller writes confirmed domain values (degrees) through + :meth:`~deux.DuiKey.set_range` with ``min_val=0`` and ``max_val=360``. + + Angle calculation + ~~~~~~~~~~~~~~~~~ + * **Second hand** -- ``6 degrees per second`` (``360 / 60``). At + ``:00`` the angle is ``0`` (12 o'clock position). + * **Minute hand** -- ``6 degrees per minute`` (``360 / 60``), with + sub-minute precision contributed by the seconds component + (``0.1 deg/s``). At 12 o'clock the angle is ``0``. + * **Hour hand** -- ``0.5 degrees per minute`` (``30 / 60``), i.e. + ``30 degrees per hour`` plus a smooth drift across the hour driven + by the minutes (and seconds). The hour value is taken modulo 12 + so that 12:00 and 00:00 both render at ``0``. + + Because this is a pure display (no user input changes the time), the + controller does not own a backend service. It runs a single + ``asyncio`` task that ticks once per second, recomputes the hand + angles, writes them to the key, and requests a refresh. The task is + started in :meth:`on_attach` and cancelled in :meth:`on_detach`, so + it is safe across reconnect cycles. + + Notes + ----- + The tick task only requests a refresh when at least one hand angle + actually changes since the last tick, avoiding redundant renders. + With a one-second tick the second hand changes every iteration, so + in practice a refresh is issued each tick while the controller is + attached. + """ + + TICK_INTERVAL_S = 1.0 + ANGLE_MIN = 0 + ANGLE_MAX = 360 + DEGREES_PER_SECOND_SECOND_HAND = 6.0 + DEGREES_PER_MINUTE_MINUTE_HAND = 6.0 + DEGREES_PER_MINUTE_HOUR_HAND = 0.5 + + def __init__(self) -> None: + self.key = DuiKey("ClockKey") + self._tick_task: asyncio.Task[None] | None = None + self._last_hour_angle: float | None = None + self._last_minute_angle: float | None = None + self._last_second_angle: float | None = None + + # Log every manifest input event at INFO so the clock key is + # easy to verify on a live device. These are pure observers -- + # they do not mutate any binding. + for event_name in ("press", "release", "click", "hold"): + self.key.on(event_name)(self._log_event(event_name)) + + @staticmethod + def _log_event(name: str) -> Callable[[], Awaitable[None]]: + """Build an async handler that logs *name* at INFO when invoked. + + Parameters + ---------- + name : str + The DUI event name to embed in the log message. + + Returns + ------- + Callable[[], Awaitable[None]] + An async, zero-argument handler suitable for + :meth:`~deux.DuiKey.on`. + """ + + async def _handler() -> None: + log.info("ClockKey event: %s", name) + + return _handler + + @classmethod + def compute_angles( + cls, now: datetime.datetime + ) -> tuple[float, float, float]: + """Compute the (hour, minute, second) hand angles in degrees for *now*. + + All angles are normalised so that ``0`` corresponds to the + 12 o'clock position and values increase clockwise. + + Parameters + ---------- + now : datetime.datetime + The point in time to render. Only the ``hour``, ``minute``, + and ``second`` fields are used. + + Returns + ------- + tuple[float, float, float] + ``(hour_angle, minute_angle, second_angle)`` all in the + range ``[0, 360)`` degrees. + """ + total_minutes = now.minute + now.second / 60.0 + second_angle = ( + now.second * cls.DEGREES_PER_SECOND_SECOND_HAND + ) % cls.ANGLE_MAX + minute_angle = ( + total_minutes * cls.DEGREES_PER_MINUTE_MINUTE_HAND + ) % cls.ANGLE_MAX + hour_angle = ( + (now.hour % 12) * 30.0 + + total_minutes * cls.DEGREES_PER_MINUTE_HOUR_HAND + ) % cls.ANGLE_MAX + return hour_angle, minute_angle, second_angle + + def _apply_now(self, now: datetime.datetime) -> bool: + """Update the key bindings for time *now*. + + Parameters + ---------- + now : datetime.datetime + The time to render. + + Returns + ------- + bool + ``True`` if any hand angle changed since the previous call, + ``False`` otherwise. + """ + hour_angle, minute_angle, second_angle = self.compute_angles(now) + if ( + hour_angle == self._last_hour_angle + and minute_angle == self._last_minute_angle + and second_angle == self._last_second_angle + ): + return False + self.key.set_range( + "hour_hand", + hour_angle, + min_val=self.ANGLE_MIN, + max_val=self.ANGLE_MAX, + ) + self.key.set_range( + "minute_hand", + minute_angle, + min_val=self.ANGLE_MIN, + max_val=self.ANGLE_MAX, + ) + self.key.set_range( + "second_hand", + second_angle, + min_val=self.ANGLE_MIN, + max_val=self.ANGLE_MAX, + ) + self._last_hour_angle = hour_angle + self._last_minute_angle = minute_angle + self._last_second_angle = second_angle + return True + + async def on_attach(self, deck: Deck) -> None: + """Render the current time and start the per-second tick task. + + Idempotent so reconnects do not stack background tasks. + + Parameters + ---------- + deck + The :class:`~deux.Deck` instance (unused -- the clock is + independent of deck state). + """ + del deck + self._apply_now(datetime.datetime.now()) + if self._tick_task is None or self._tick_task.done(): + self._tick_task = asyncio.create_task( + self._tick_loop(), name="clock-tick" + ) + + async def on_detach(self) -> None: + """Cancel the tick task and unsubscribe key events.""" + task = self._tick_task + self._tick_task = None + if task is not None and not task.done(): + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + await super().on_detach() + + async def _tick_loop(self) -> None: + """Drive hand updates once per second. + + Only requests a refresh when a visible angle actually changes, + so the renderer is not woken up unnecessarily. + """ + try: + while True: + await asyncio.sleep(self.TICK_INTERVAL_S) + if self._apply_now(datetime.datetime.now()): + await self.key.request_refresh() + except asyncio.CancelledError: + pass + class SceneController: """Scene-activation keys -- one :class:`DuiKey` per scene definition. @@ -1063,6 +1265,7 @@ def __init__( self.lights = LightsController() self.timer = TimerController() self.gauge = GaugeController(simulate=False) + self.clock = ClockController() self.dashboard = DashboardController() self.favorites = FavoritesController(catalog, self.audio) self.scenes = SceneController(scene_defs) @@ -1080,6 +1283,12 @@ def __init__( self.gauge, self.dashboard, ] + # KeyController-derived objects follow the same lifecycle but + # are tracked separately because they expose ``key`` instead of + # ``card`` and the typed list above is constrained to cards. + self._key_controllers: list[KeyController] = [ + self.clock, + ] async def on_connect(self, deck: Deck) -> None: """Configure screens for *deck* and start (or resume) the demo. @@ -1124,6 +1333,8 @@ async def _log_screen(name: str, screens: dict) -> None: # Drive the uniform lifecycle on every controller. for controller in self._controllers: await controller.on_attach(deck) + for key_controller in self._key_controllers: + await key_controller.on_attach(deck) self.nav.on_attach(deck) self.scenes.set_deck(deck) @@ -1147,11 +1358,17 @@ async def on_disconnect(self, info: DeviceInfo) -> None: ) for controller in self._controllers: await controller.on_detach() + for key_controller in self._key_controllers: + await key_controller.on_detach() # -- screen construction ------------------------------------------- def _build_main_screen(self, deck: Deck) -> None: - """Layout: favourites + scenes on keys, all four cards on the strip.""" + """Layout: clock on key 0, favourites + scenes on the remaining keys. + + Cards (audio, lights, gauge, dashboard) fill the touch strip + when the deck has one. + """ caps = deck.capabilities screen = deck.screen("Main") @@ -1161,13 +1378,31 @@ def _build_main_screen(self, deck: Deck) -> None: screen.set_card(2, self.gauge.card) screen.set_card(3, self.dashboard.card) - num_favs = min(len(self.favorites.keys), caps.key_count) - remaining = max(0, caps.key_count - num_favs) + # Reserve key 0 for the analog clock; lay favourites and scenes + # on the remaining keys in order. + clock_slot = 0 + if caps.key_count > clock_slot: + screen.set_key(clock_slot, self.clock.key) + next_slot = clock_slot + 1 + else: + next_slot = 0 + + remaining = max(0, caps.key_count - next_slot) + num_favs = min(len(self.favorites.keys), remaining) + remaining -= num_favs num_scenes = min(len(self.scenes.keys), remaining) - self.favorites.install(screen, list(range(num_favs))) + self.favorites.install( + screen, list(range(next_slot, next_slot + num_favs)) + ) self.scenes.install( - screen, list(range(num_favs, num_favs + num_scenes)) + screen, + list( + range( + next_slot + num_favs, + next_slot + num_favs + num_scenes, + ) + ), ) def _build_settings_screen(self, deck: Deck) -> None: