Merge pull request #16 from strohganoff/refactor/eventbase-subscripting

Refactor/eventbase subscripting

Author: strohganoff

README.md

@@ -292,7 +292,7 @@ The SDK allows you to create custom event listeners and events by extending the
 
 To create a custom event listener:
 
-1. Create new event model that inherits from `EventBase`.
+1. Create new event model that inherits from `EventBase["eventName"]`.
 2. Create a new class that inherits from `EventListener`.
     a. Implement the required `listen` and `stop` methods. The `listen` method should yield results as a json string that matches the new event model.
     b. List the new event classes in the `event_models` class variable of the new `EventListener` class.
@@ -307,11 +307,15 @@ from streamdeck.event_listener import EventListener
 from streamdeck.models.events import EventBase
 
 
-class MyCustomEvent(EventBase):
-    event: Literal["somethingHappened"]
-    ... # Define additional data attributes here
+class MyCustomEvent(EventBase["somethingHappened"]):
+    # The 'event' field's type annotation is internally set as Literal["somethingHappened"]
+    # Define additional data attributes here
+    result: str
+
 
 class MyCustomEventListener(EventListener):
+    event_models = [MyCustomEvent]
+    
     def listen(self) -> Generator[str | bytes, None, None]:
         ...
         # Listen/poll for something here in a loop, and yield the result.
@@ -320,7 +324,7 @@ class MyCustomEventListener(EventListener):
         # while self._running is True:
         #     result = module.check_status()
         #     if result is not None:
-        #         yield json.dumps({"event": "somethingHappend", "result": result})
+        #         yield json.dumps({"event": "somethingHappened", "result": result})
         #     time.sleep(1)
 
     def stop(self) -> None:
@@ -344,7 +348,7 @@ To use your custom event listener, add it to your `pyproject.toml` file:
     ]
 ```
 
-The `event_listeners` list should contain strings in module format for each module you want to use.
+The `event_listener_modules` list should contain strings in module format for each module you want to use.
 
 
 ## Creating and Packaging Plugins

streamdeck/__init__.py

@@ -3,7 +3,6 @@
     command_sender,
     manager,
     models,
-    types,
     utils,
     websocket,
 )
@@ -14,7 +13,6 @@
     "command_sender",
     "manager",
     "models",
-    "types",
     "utils",
     "websocket",
 ]

streamdeck/actions.py

@@ -9,10 +9,22 @@
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Generator
+    from typing import Protocol
+
+    from typing_extensions import ParamSpec, TypeVar
 
     from streamdeck.models.events import EventBase
-    from streamdeck.models.events.base import LiteralStrGenericAlias
-    from streamdeck.types import BaseEventHandlerFunc, EventHandlerFunc, EventNameStr, TEvent_contra
+    from streamdeck.types import ActionUUIDStr, EventNameStr
+
+
+
+    EventModel_contra = TypeVar("EventModel_contra", bound=EventBase, default=EventBase, contravariant=True)
+    InjectableParams = ParamSpec("InjectableParams", default=...)
+
+    class EventHandlerFunc(Protocol[EventModel_contra, InjectableParams]):
+        """Protocol for an event handler function that takes an event (of subtype of EventBase) and other parameters that are injectable."""
+        def __call__(self, event_data: EventModel_contra, *args: InjectableParams.args, **kwargs: InjectableParams.kwargs) -> None: ...
+
 
 
 logger = getLogger("streamdeck.actions")
@@ -22,14 +34,10 @@ class ActionBase(ABC):
     """Base class for all actions."""
 
     def __init__(self) -> None:
-        """Initialize an Action instance.
+        """Initialize an Action instance."""
+        self._events: dict[EventNameStr, set[EventHandlerFunc]] = defaultdict(set)
 
-        Args:
-            uuid (str): The unique identifier for the action.
-        """
-        self._events: dict[EventNameStr, set[BaseEventHandlerFunc]] = defaultdict(set)
-
-    def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[TEvent_contra] | BaseEventHandlerFunc], EventHandlerFunc[TEvent_contra] | BaseEventHandlerFunc]:
+    def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[EventModel_contra, InjectableParams]], EventHandlerFunc[EventModel_contra, InjectableParams]]:
         """Register an event handler for a specific event.
 
         Args:
@@ -41,15 +49,14 @@ def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[TEvent_c
         Raises:
             KeyError: If the provided event name is not available.
         """
-        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))
-
+        def _wrapper(func: EventHandlerFunc[EventModel_contra, InjectableParams]) -> EventHandlerFunc[EventModel_contra, InjectableParams]:
+            # Cast to EventHandlerFunc with default type arguments so that the storage type is consistent.
+            self._events[event_name].add(cast("EventHandlerFunc", func))
             return func
 
         return _wrapper
 
-    def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc[EventBase[LiteralStrGenericAlias]], None, None]:
+    def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc, None, None]:
         """Get all event handlers for a specific event.
 
         Args:
@@ -66,22 +73,23 @@ def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHand
 
         yield from self._events[event_name]
 
-    def get_registered_event_names(self) -> list[str]:
+    def get_registered_event_names(self) -> list[EventNameStr]:
         """Get all event names for which event handlers are registered.
 
         Returns:
             list[str]: The list of event names for which event handlers are registered.
         """
         return list(self._events.keys())
 
+
 class GlobalAction(ActionBase):
     """Represents an action that is performed at the plugin level, meaning it isn't associated with a specific device or action."""
 
 
 class Action(ActionBase):
     """Represents an action that can be performed for a specific action, with event handlers for specific event types."""
 
-    def __init__(self, uuid: str) -> None:
+    def __init__(self, uuid: ActionUUIDStr) -> None:
         """Initialize an Action instance.
 
         Args:
@@ -111,7 +119,7 @@ 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[EventBase[LiteralStrGenericAlias]], None, None]:
+    def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: ActionUUIDStr | None = None) -> Generator[EventHandlerFunc, None, None]:
         """Get all event handlers for a specific event from all registered actions.
 
         Args:

streamdeck/command_sender.py

@@ -7,6 +7,13 @@
 if TYPE_CHECKING:
     from typing import Any, Literal
 
+    from streamdeck.types import (
+        ActionInstanceUUIDStr,
+        ActionUUIDStr,
+        DeviceUUIDStr,
+        EventNameStr,
+        PluginDefinedData,
+    )
     from streamdeck.websocket import WebSocketClient
 
 
@@ -16,30 +23,31 @@
 
 class StreamDeckCommandSender:
     """Class for sending command event messages to the Stream Deck software through a WebSocket client."""
+
     def __init__(self, client: WebSocketClient, plugin_registration_uuid: str):
         self._client = client
         self._plugin_registration_uuid = plugin_registration_uuid
 
-    def _send_event(self, event: str, **kwargs: Any) -> None:
+    def _send_event(self, event: EventNameStr, **kwargs: Any) -> None:
         self._client.send_event({
             "event": event,
             **kwargs,
         })
 
-    def set_settings(self, context: str, payload: dict[str, Any]) -> None:
+    def set_settings(self, context: ActionInstanceUUIDStr, payload: PluginDefinedData) -> None:
         self._send_event(
             event="setSettings",
             context=context,
             payload=payload,
         )
 
-    def get_settings(self, context: str) -> None:
+    def get_settings(self, context: ActionInstanceUUIDStr) -> None:
         self._send_event(
             event="getSettings",
             context=context,
         )
 
-    def set_global_settings(self, payload: dict[str, Any]) -> None:
+    def set_global_settings(self, payload: PluginDefinedData) -> None:
         self._send_event(
             event="setGlobalSettings",
             context=self._plugin_registration_uuid,
@@ -52,14 +60,14 @@ def get_global_settings(self) -> None:
             context=self._plugin_registration_uuid,
         )
 
-    def open_url(self, context: str, url: str) -> None:
+    def open_url(self, context: ActionInstanceUUIDStr, url: str) -> None:
         self._send_event(
             event="openUrl",
             context=context,
             payload={"url": url},
         )
 
-    def log_message(self, context: str, message: str) -> None:
+    def log_message(self, context: ActionInstanceUUIDStr, message: str) -> None:
         self._send_event(
             event="logMessage",
             context=context,
@@ -68,10 +76,10 @@ def log_message(self, context: str, message: str) -> None:
 
     def set_title(
         self,
-        context: str,
+        context: ActionInstanceUUIDStr,
         state: int | None = None,
-        target: str | None = None,
-        title: str | None = None
+        target: Literal["hardware", "software", "both"] | None = None,
+        title: str | None = None,
     ) -> None:
         payload = {}
 
@@ -90,10 +98,10 @@ def set_title(
 
     def set_image(
         self,
-        context: str,
-        image: str,                                         # base64 encoded image,
-        target: Literal["hardware", "software", "both"],    # software, hardware, or both,
-        state: int,                                         # 0-based integer
+        context: ActionInstanceUUIDStr,
+        image: str,  # base64 encoded image,
+        target: Literal["hardware", "software", "both"],
+        state: int,
     ) -> None:
         """...
 
@@ -117,14 +125,26 @@ def set_image(
             },
         )
 
-    def set_feedback(self, context: str, payload: dict[str, Any]) -> None:
+    def set_feedback(self, context: ActionInstanceUUIDStr, payload: PluginDefinedData) -> None:
+        """Set's the feedback of an existing layout associated with an action instance.
+
+        Args:
+          context (str): Defines the context of the command, e.g. which action instance the command is intended for.
+          payload (PluginDefinedData): Additional information supplied as part of the command.
+        """
         self._send_event(
             event="setFeedback",
             context=context,
             payload=payload,
         )
 
-    def set_feedback_layout(self, context: str, layout: str) -> None:
+    def set_feedback_layout(self, context: ActionInstanceUUIDStr, layout: str) -> None:
+        """Sets the layout associated with an action instance.
+
+        Args:
+          context (str): Defines the context of the command, e.g. which action instance the command is intended for.
+          layout (str): Name of a pre-defined layout, or relative path to a custom one.
+        """
         self._send_event(
             event="setFeedbackLayout",
             context=context,
@@ -133,7 +153,7 @@ def set_feedback_layout(self, context: str, layout: str) -> None:
 
     def set_trigger_description(
         self,
-        context: str,
+        context: ActionInstanceUUIDStr,
         rotate: str | None = None,
         push: str | None = None,
         touch: str | None = None,
@@ -170,21 +190,21 @@ def set_trigger_description(
             },
         )
 
-    def show_alert(self, context: str) -> None:
+    def show_alert(self, context: ActionInstanceUUIDStr) -> None:
         """Temporarily show an alert icon on the image displayed by an instance of an action."""
         self._send_event(
             event="showAlert",
             context=context,
         )
 
-    def show_ok(self, context: str) -> None:
+    def show_ok(self, context: ActionInstanceUUIDStr) -> None:
         """Temporarily show an OK checkmark icon on the image displayed by an instance of an action."""
         self._send_event(
             event="showOk",
             context=context,
         )
 
-    def set_state(self, context: str, state: int) -> None:
+    def set_state(self, context: ActionInstanceUUIDStr, state: int) -> None:
         self._send_event(
             event="setState",
             context=context,
@@ -193,8 +213,8 @@ def set_state(self, context: str, state: int) -> None:
 
     def switch_to_profile(
         self,
-        context: str,
-        device: str,
+        context: ActionInstanceUUIDStr,
+        device: DeviceUUIDStr,
         profile: str | None = None,
         page: int = 0,
     ) -> None:
@@ -211,7 +231,7 @@ def switch_to_profile(
             page (int):  Page to show when switching to the profile; indexed from 0.
         """
         # TODO: Should validation happen that ensures the specified profile is declared in manifest.yaml?
-        payload = {}
+        payload: dict[str, str | int | None] = {}
 
         if profile is not None:
             payload = {
@@ -226,18 +246,17 @@ def switch_to_profile(
             payload=payload,
         )
 
-    def send_to_property_inspector(self, context: str, payload: dict[str, Any]) -> None:
+    def send_to_property_inspector(
+        self, context: ActionInstanceUUIDStr, payload: PluginDefinedData
+    ) -> None:
         self._send_event(
             event="sendToPropertyInspector",
             context=context,
             payload=payload,
         )
 
     def send_to_plugin(
-        self,
-        context: str,
-        action: str,
-        payload: dict[str, Any]
+        self, context: ActionInstanceUUIDStr, action: ActionUUIDStr, payload: PluginDefinedData
     ) -> None:
         """Send a payload to another plugin.
 

streamdeck/event_listener.py

@@ -14,7 +14,6 @@
     from typing_extensions import TypeIs
 
     from streamdeck.models.events import EventBase
-    from streamdeck.models.events.base import LiteralStrGenericAlias
 
 
 
@@ -120,7 +119,7 @@ class EventListener(ABC):
     Event listeners are classes that listen for events and simply yield them as they come.
     The EventListenerManager will handle the threading and pushing the events yielded into a shared queue.
     """
-    event_models: ClassVar[list[type[EventBase[LiteralStrGenericAlias]]]]
+    event_models: ClassVar[list[type[EventBase]]]
     """A list of event models that the listener can yield. Read in by the PluginManager to model the incoming event data off of.
 
     The plugin-developer must define this list in their subclass.