Merge pull request #7 from strohganoff/fix/event-handler-generic-type

Fix typing for events and event handlers

Author: strohganoff

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)

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)
+
+

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]