From ccf8310aaa5eee9b5dc6ab957099e2615efa0252 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Fri, 14 Feb 2025 15:34:56 -0600 Subject: [PATCH] Big fix up of typing for events and event handlers. --- streamdeck/actions.py | 48 +++++------------ streamdeck/manager.py | 30 +++++++---- streamdeck/models/events.py | 40 +++++++------- streamdeck/types.py | 83 +++++++++++++++++++++++++++-- streamdeck/utils/helper_actions.py | 3 +- tests/actions/test_action.py | 5 +- tests/actions/test_global_action.py | 3 +- tests/data/test_action1.py | 8 ++- tests/data/test_action2.py | 8 +-- 9 files changed, 150 insertions(+), 78 deletions(-) diff --git a/streamdeck/actions.py b/streamdeck/actions.py index c7aea1d..25d222c 100644 --- a/streamdeck/actions.py +++ b/streamdeck/actions.py @@ -4,42 +4,21 @@ from collections import defaultdict from functools import cached_property from logging import getLogger -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast + +from streamdeck.types import BaseEventHandlerFunc, available_event_names if TYPE_CHECKING: from collections.abc import Callable, Generator - from streamdeck.types import EventHandlerFunc, EventNameStr + from streamdeck.models.events import EventBase + from streamdeck.types import EventHandlerFunc, EventNameStr, TEvent_contra logger = getLogger("streamdeck.actions") - -available_event_names: set[EventNameStr] = { - "applicationDidLaunch", - "applicationDidTerminate", - "deviceDidConnect", - "deviceDidDisconnect", - "dialDown", - "dialRotate", - "dialUp", - "didReceiveGlobalSettings", - "didReceiveDeepLink", - "didReceiveSettings", - "keyDown", - "keyUp", - "propertyInspectorDidAppear", - "propertyInspectorDidDisappear", - "systemDidWakeUp", - "titleParametersDidChange", - "touchTap", - "willAppear", - "willDisappear", -} - - class ActionBase(ABC): """Base class for all actions.""" @@ -49,9 +28,9 @@ def __init__(self): Args: uuid (str): The unique identifier for the action. """ - self._events: dict[EventNameStr, set[EventHandlerFunc]] = defaultdict(set) + self._events: dict[EventNameStr, set[BaseEventHandlerFunc]] = defaultdict(set) - def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc], EventHandlerFunc]: + def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[TEvent_contra] | BaseEventHandlerFunc], EventHandlerFunc[TEvent_contra] | BaseEventHandlerFunc]: """Register an event handler for a specific event. Args: @@ -67,14 +46,15 @@ def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc], EventH msg = f"Provided event name for action handler does not exist: {event_name}" raise KeyError(msg) - def _wrapper(func: EventHandlerFunc) -> EventHandlerFunc: - self._events[event_name].add(func) + def _wrapper(func: EventHandlerFunc[TEvent_contra]) -> EventHandlerFunc[TEvent_contra]: + # Cast to BaseEventHandlerFunc so that the storage type is consistent. + self._events[event_name].add(cast(BaseEventHandlerFunc, func)) return func return _wrapper - def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc, None, None]: + def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc[EventBase], None, None]: """Get all event handlers for a specific event. Args: @@ -133,12 +113,12 @@ def register(self, action: ActionBase) -> None: """ self._plugin_actions.append(action) - def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | None = None) -> Generator[EventHandlerFunc, None, None]: + def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | None = None) -> Generator[EventHandlerFunc[EventBase], None, None]: """Get all event handlers for a specific event from all registered actions. Args: event_name (EventName): The name of the event to retrieve handlers for. - event_action_uuid (str | None): The action UUID to get handlers for. + event_action_uuid (str | None): The action UUID to get handlers for. If None (i.e., the event is not action-specific), get all handlers for the event. Yields: @@ -148,7 +128,7 @@ def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | # If the event is action-specific (i.e is not a GlobalAction and has a UUID attribute), # only get handlers for that action, as we don't want to trigger # and pass this event to handlers for other actions. - if event_action_uuid is not None and (hasattr(action, "uuid") and action.uuid != event_action_uuid): + if event_action_uuid is not None and (isinstance(action, Action) and action.uuid != event_action_uuid): continue yield from action.get_event_handlers(event_name) diff --git a/streamdeck/manager.py b/streamdeck/manager.py index 3f8bc3b..ac31335 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -1,20 +1,24 @@ from __future__ import annotations import functools -import inspect -import logging from logging import getLogger from typing import TYPE_CHECKING from streamdeck.actions import ActionRegistry from streamdeck.command_sender import StreamDeckCommandSender from streamdeck.models.events import ContextualEventMixin, event_adapter +from streamdeck.types import ( + EventHandlerBasicFunc, + EventHandlerFunc, + TEvent_contra, + is_bindable_handler, + is_valid_event_name, +) from streamdeck.utils.logging import configure_streamdeck_logger from streamdeck.websocket import WebSocketClient if TYPE_CHECKING: - from collections.abc import Callable from typing import Any, Literal from streamdeck.actions import Action @@ -70,7 +74,7 @@ def register_action(self, action: Action) -> None: self._registry.register(action) - def _inject_command_sender(self, handler: Callable[..., None], command_sender: StreamDeckCommandSender) -> Callable[..., None]: + def _inject_command_sender(self, handler: EventHandlerFunc[TEvent_contra], command_sender: StreamDeckCommandSender) -> EventHandlerBasicFunc[TEvent_contra]: """Inject command_sender into handler if it accepts it as a parameter. Args: @@ -80,10 +84,7 @@ def _inject_command_sender(self, handler: Callable[..., None], command_sender: S Returns: The handler with command_sender injected if needed """ - args: dict[str, inspect.Parameter] = inspect.signature(handler).parameters - - # Check dynamically if the `command_sender`'s name is in the handler's arguments. - if "command_sender" in args: + if is_bindable_handler(handler): return functools.partial(handler, command_sender=command_sender) return handler @@ -101,13 +102,20 @@ def run(self) -> None: for message in client.listen_forever(): data: EventBase = event_adapter.validate_json(message) + + if not is_valid_event_name(data.event): + logger.error("Received event name is not valid: %s", data.event) + continue + logger.debug("Event received: %s", data.event) # If the event is action-specific, we'll pass the action's uuid to the handler to ensure only the correct action is triggered. event_action_uuid = data.action if isinstance(data, ContextualEventMixin) else None - for handler in self._registry.get_action_handlers(event_name=data.event, event_action_uuid=event_action_uuid): - handler = self._inject_command_sender(handler, command_sender) + for event_handler in self._registry.get_action_handlers(event_name=data.event, event_action_uuid=event_action_uuid): + processed_handler = self._inject_command_sender(event_handler, command_sender) # TODO: from contextual event occurences, save metadata to the action's properties. - handler(data) + processed_handler(data) + + diff --git a/streamdeck/models/events.py b/streamdeck/models/events.py index 60cde3b..bad3cd2 100644 --- a/streamdeck/models/events.py +++ b/streamdeck/models/events.py @@ -29,82 +29,82 @@ class DeviceSpecificEventMixin: class ApplicationDidLaunch(EventBase): - event: Literal["applicationDidLaunch"] + event: Literal["applicationDidLaunch"] # type: ignore[override] payload: dict[Literal["application"], str] """Payload containing the name of the application that triggered the event.""" class ApplicationDidTerminate(EventBase): - event: Literal["applicationDidTerminate"] + event: Literal["applicationDidTerminate"] # type: ignore[override] payload: dict[Literal["application"], str] """Payload containing the name of the application that triggered the event.""" class DeviceDidConnect(EventBase, DeviceSpecificEventMixin): - event: Literal["deviceDidConnect"] + event: Literal["deviceDidConnect"] # type: ignore[override] deviceInfo: dict[str, Any] """Information about the newly connected device.""" class DeviceDidDisconnect(EventBase, DeviceSpecificEventMixin): - event: Literal["deviceDidDisconnect"] + event: Literal["deviceDidDisconnect"] # type: ignore[override] class DialDown(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["dialDown"] + event: Literal["dialDown"] # type: ignore[override] payload: dict[str, Any] class DialRotate(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["dialRotate"] + event: Literal["dialRotate"] # type: ignore[override] payload: dict[str, Any] class DialUp(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["dialUp"] + event: Literal["dialUp"] # type: ignore[override] payload: dict[str, Any] class DidReceiveDeepLink(EventBase): - event: Literal["didReceiveDeepLink"] + event: Literal["didReceiveDeepLink"] # type: ignore[override] payload: dict[Literal["url"], str] class DidReceiveGlobalSettings(EventBase): - event: Literal["didReceiveGlobalSettings"] + event: Literal["didReceiveGlobalSettings"] # type: ignore[override] payload: dict[Literal["settings"], dict[str, Any]] class DidReceivePropertyInspectorMessage(EventBase, ContextualEventMixin): - event: Literal["sendToPlugin"] + event: Literal["sendToPlugin"] # type: ignore[override] payload: dict[str, Any] class DidReceiveSettings(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["didReceiveSettings"] + event: Literal["didReceiveSettings"] # type: ignore[override] payload: dict[str, Any] class KeyDown(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["keyDown"] + event: Literal["keyDown"] # type: ignore[override] payload: dict[str, Any] class KeyUp(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["keyUp"] + event: Literal["keyUp"] # type: ignore[override] payload: dict[str, Any] class PropertyInspectorDidAppear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["propertyInspectorDidAppear"] + event: Literal["propertyInspectorDidAppear"] # type: ignore[override] class PropertyInspectorDidDisappear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["propertyInspectorDidDisappear"] + event: Literal["propertyInspectorDidDisappear"] # type: ignore[override] class SystemDidWakeUp(EventBase): - event: Literal["systemDidWakeUp"] + event: Literal["systemDidWakeUp"] # type: ignore[override] class TitleParametersDict(TypedDict): @@ -127,24 +127,24 @@ class TitleParametersDidChangePayload(TypedDict): class TitleParametersDidChange(EventBase, DeviceSpecificEventMixin): - event: Literal["titleParametersDidChange"] + event: Literal["titleParametersDidChange"] # type: ignore[override] context: str """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" payload: TitleParametersDidChangePayload class TouchTap(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["touchTap"] + event: Literal["touchTap"] # type: ignore[override] payload: dict[str, Any] class WillAppear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["willAppear"] + event: Literal["willAppear"] # type: ignore[override] payload: dict[str, Any] class WillDisappear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["willDisappear"] + event: Literal["willDisappear"] # type: ignore[override] payload: dict[str, Any] diff --git a/streamdeck/types.py b/streamdeck/types.py index efc0ee4..52a6f71 100644 --- a/streamdeck/types.py +++ b/streamdeck/types.py @@ -1,16 +1,42 @@ from __future__ import annotations -from collections.abc import Callable -from typing import Literal +import inspect +from typing import TYPE_CHECKING, Literal, Protocol, TypeVar, Union -from typing_extensions import TypeAlias # noqa: UP035 +from typing_extensions import TypeAlias, TypeIs # noqa: UP035 from streamdeck.models.events import EventBase -EventHandlerFunc: TypeAlias = Callable[[EventBase], None] # noqa: UP040 +if TYPE_CHECKING: + from streamdeck.command_sender import StreamDeckCommandSender +available_event_names: set[EventNameStr] = { + "applicationDidLaunch", + "applicationDidTerminate", + "deviceDidConnect", + "deviceDidDisconnect", + "dialDown", + "dialRotate", + "dialUp", + "didReceiveGlobalSettings", + "didReceiveDeepLink", + "didReceiveSettings", + "sendToPlugin", # DidReceivePropertyInspectorMessage event + "keyDown", + "keyUp", + "propertyInspectorDidAppear", + "propertyInspectorDidDisappear", + "systemDidWakeUp", + "titleParametersDidChange", + "touchTap", + "willAppear", + "willDisappear", +} + + +# For backwards compatibility with older versions of Python, we can't just use available_event_names as the values of the Literal EventNameStr. EventNameStr: TypeAlias = Literal[ # noqa: UP040 "applicationDidLaunch", "applicationDidTerminate", @@ -22,6 +48,7 @@ "didReceiveGlobalSettings", "didReceiveDeepLink", "didReceiveSettings", + "sendToPlugin", # DidReceivePropertyInspectorMessage event "keyDown", "keyUp", "propertyInspectorDidAppear", @@ -32,3 +59,51 @@ "willAppear", "willDisappear" ] + + +def is_valid_event_name(event_name: str) -> TypeIs[EventNameStr]: + """Check if the event name is one of the available event names.""" + return event_name in available_event_names + + +### Event Handler Type Definitions ### + +## Protocols for event handler functions that act on subtypes of EventBase instances in a Generic way. + +# A type variable for a subtype of EventBase +TEvent_contra = TypeVar("TEvent_contra", bound=EventBase, contravariant=True) + + +class EventHandlerBasicFunc(Protocol[TEvent_contra]): + """Protocol for a basic event handler function that takes just an event (of subtype of EventBase).""" + def __call__(self, event_data: TEvent_contra) -> None: ... + + +class EventHandlerBindableFunc(Protocol[TEvent_contra]): + """Protocol for an event handler function that takes an event (of subtype of EventBase) and a command sender.""" + def __call__(self, event_data: TEvent_contra, command_sender: StreamDeckCommandSender) -> None: ... + + +# Type alias for an event handler function that takes an event (of subtype of EventBase), and optionally a command sender. +EventHandlerFunc = Union[EventHandlerBasicFunc[TEvent_contra], EventHandlerBindableFunc[TEvent_contra]] # noqa: UP007 + + +## Protocols for event handler functions that act on EventBase instances. + +class BaseEventHandlerBasicFunc(EventHandlerBasicFunc[EventBase]): + """Protocol for a basic event handler function that takes just an EventBase.""" + +class BaseEventHandlerBindableFunc(EventHandlerBindableFunc[EventBase]): + """Protocol for an event handler function that takes an event (of subtype of EventBase) and a command sender.""" + + +# Type alias for a base event handler function that expects an actual EventBase instance (and optionally a command sender) — used for type hinting internal storage of event handlers. +BaseEventHandlerFunc = Union[BaseEventHandlerBasicFunc, BaseEventHandlerBindableFunc] # noqa: UP007 + + + +# def is_bindable_handler(handler: EventHandlerFunc[TEvent_contra] | BaseEventHandlerFunc) -> TypeIs[EventHandlerBindableFunc[TEvent_contra] | BaseEventHandlerBindableFunc]: +def is_bindable_handler(handler: EventHandlerBasicFunc[TEvent_contra] | EventHandlerBindableFunc[TEvent_contra]) -> TypeIs[EventHandlerBindableFunc[TEvent_contra]]: + """Check if the handler is prebound with the `command_sender` parameter.""" + # Check dynamically if the `command_sender`'s name is in the handler's arguments. + return "command_sender" in inspect.signature(handler).parameters diff --git a/streamdeck/utils/helper_actions.py b/streamdeck/utils/helper_actions.py index 2a42568..9255bd2 100644 --- a/streamdeck/utils/helper_actions.py +++ b/streamdeck/utils/helper_actions.py @@ -3,7 +3,8 @@ from logging import getLogger from typing import TYPE_CHECKING -from streamdeck.actions import Action, available_event_names +from streamdeck.actions import Action +from streamdeck.types import available_event_names if TYPE_CHECKING: diff --git a/tests/actions/test_action.py b/tests/actions/test_action.py index 25cf76c..c5751b8 100644 --- a/tests/actions/test_action.py +++ b/tests/actions/test_action.py @@ -3,7 +3,8 @@ from typing import TYPE_CHECKING import pytest -from streamdeck.actions import Action, available_event_names +from streamdeck.actions import Action +from streamdeck.types import available_event_names if TYPE_CHECKING: @@ -85,4 +86,4 @@ def test_action_get_event_handlers_invalid_event(): action = Action("test.uuid.for.action") with pytest.raises(KeyError): - list(action.get_event_handlers("invalidEvent")) \ No newline at end of file + list(action.get_event_handlers("invalidEvent")) diff --git a/tests/actions/test_global_action.py b/tests/actions/test_global_action.py index 39c4bc0..8654a96 100644 --- a/tests/actions/test_global_action.py +++ b/tests/actions/test_global_action.py @@ -3,7 +3,8 @@ from typing import TYPE_CHECKING import pytest -from streamdeck.actions import GlobalAction, available_event_names +from streamdeck.actions import GlobalAction +from streamdeck.types import available_event_names if TYPE_CHECKING: diff --git a/tests/data/test_action1.py b/tests/data/test_action1.py index ccab56f..f2d366b 100644 --- a/tests/data/test_action1.py +++ b/tests/data/test_action1.py @@ -1,5 +1,6 @@ from streamdeck.actions import Action -from streamdeck.models.events import KeyDown +from streamdeck.command_sender import StreamDeckCommandSender +from streamdeck.models.events import KeyDown, WillAppear test_action1 = Action("my-first-test-action") @@ -8,3 +9,8 @@ @test_action1.on("keyDown") def handle_key_down_event(event_data: KeyDown) -> None: print("KeyDown event handled.") + + +@test_action1.on("willAppear") +def handle_will_appear_event(event_data: WillAppear, command_sender: StreamDeckCommandSender) -> None: + print("WillAppear event handled.", command_sender) diff --git a/tests/data/test_action2.py b/tests/data/test_action2.py index 4ede7fd..3c1142a 100644 --- a/tests/data/test_action2.py +++ b/tests/data/test_action2.py @@ -1,10 +1,10 @@ from streamdeck.actions import Action -from streamdeck.models.events import ApplicationDidTerminate +from streamdeck.models.events import ApplicationDidLaunch test_action2 = Action("my-first-test-action") -@test_action2.on("applicationDidTerminate") -def handle_application_terminated_event(event_data: ApplicationDidTerminate) -> None: - print("ApplicationDidTerminate event handled.") +@test_action2.on("applicationDidLaunch") +def handle_application_launched_event(event_data: ApplicationDidLaunch) -> None: + print("ApplicationDidLaunch event handled.")