From 84eeb0e7457fb71733b64a41fb141e0b798a79b3 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Tue, 4 Feb 2025 16:05:03 -0600 Subject: [PATCH 1/5] Add code and tests for new global actions feature --- streamdeck/actions.py | 44 ++-- streamdeck/models/events.py | 10 + tests/actions/__init__.py | 0 tests/{ => actions}/test_action.py | 0 tests/{ => actions}/test_action_registry.py | 0 tests/actions/test_event_handler_filtering.py | 204 ++++++++++++++++++ tests/actions/test_global_action.py | 91 ++++++++ 7 files changed, 335 insertions(+), 14 deletions(-) create mode 100644 tests/actions/__init__.py rename tests/{ => actions}/test_action.py (100%) rename tests/{ => actions}/test_action_registry.py (100%) create mode 100644 tests/actions/test_event_handler_filtering.py create mode 100644 tests/actions/test_global_action.py diff --git a/streamdeck/actions.py b/streamdeck/actions.py index ec17a41..a71c529 100644 --- a/streamdeck/actions.py +++ b/streamdeck/actions.py @@ -39,24 +39,17 @@ } -class Action: - """Represents an action that can be performed, with event handlers for specific event types.""" +class ActionBase: + """Base class for all actions.""" - def __init__(self, uuid: str): + def __init__(self): """Initialize an Action instance. Args: uuid (str): The unique identifier for the action. """ - self.uuid = uuid - self._events: dict[EventNameStr, set[EventHandlerFunc]] = defaultdict(set) - @cached_property - def name(self) -> str: - """The name of the action, derived from the last part of the UUID.""" - return self.uuid.split(".")[-1] - def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc], EventHandlerFunc]: """Register an event handler for a specific event. @@ -99,14 +92,36 @@ def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHand yield from self._events[event_name] +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): + """Initialize an Action instance. + + Args: + uuid (str): The unique identifier for the action. + """ + super().__init__() + self.uuid = uuid + + @cached_property + def name(self) -> str: + """The name of the action, derived from the last part of the UUID.""" + return self.uuid.split(".")[-1] + + class ActionRegistry: """Manages the registration and retrieval of actions and their event handlers.""" def __init__(self) -> None: """Initialize an ActionRegistry instance.""" - self._plugin_actions: list[Action] = [] + self._plugin_actions: list[ActionBase] = [] - def register(self, action: Action) -> None: + def register(self, action: ActionBase) -> None: """Register an action with the registry. Args: @@ -126,9 +141,10 @@ def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | EventHandlerFunc: The event handler functions for the specified event. """ for action in self._plugin_actions: - # If the event is action-specific, only get handlers for that action, as we don't want to trigger + # 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 event_action_uuid != action.uuid: + if event_action_uuid is not None and (hasattr(action, "uuid") and action.uuid != event_action_uuid): continue yield from action.get_event_handlers(event_name) diff --git a/streamdeck/models/events.py b/streamdeck/models/events.py index dc6eae7..2cd6bf7 100644 --- a/streamdeck/models/events.py +++ b/streamdeck/models/events.py @@ -21,6 +21,16 @@ def is_action_specific(cls) -> bool: """Check if the event is specific to an action instance (i.e. the event has an "action" field).""" return "action" in cls.model_fields + @classmethod + def is_device_specific(cls) -> bool: + """Check if the event is specific to a device instance (i.e. the event has a "device" field).""" + return "device" in cls.model_fields + + @classmethod + def is_action_instance_specific(cls) -> bool: + """Check if the event is specific to an action instance (i.e. the event has a "context" field).""" + return "context" in cls.model_fields + class ApplicationDidLaunchEvent(EventBase): event: Literal["applicationDidLaunch"] diff --git a/tests/actions/__init__.py b/tests/actions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_action.py b/tests/actions/test_action.py similarity index 100% rename from tests/test_action.py rename to tests/actions/test_action.py diff --git a/tests/test_action_registry.py b/tests/actions/test_action_registry.py similarity index 100% rename from tests/test_action_registry.py rename to tests/actions/test_action_registry.py diff --git a/tests/actions/test_event_handler_filtering.py b/tests/actions/test_event_handler_filtering.py new file mode 100644 index 0000000..8bf7c09 --- /dev/null +++ b/tests/actions/test_event_handler_filtering.py @@ -0,0 +1,204 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import create_autospec + +import pytest +from polyfactory.factories.pydantic_factory import ModelFactory +from streamdeck.actions import Action, ActionRegistry, GlobalAction +from streamdeck.models import events + + +if TYPE_CHECKING: + from unittest.mock import Mock + + + +@pytest.fixture +def mock_event_handler() -> Mock: + def dummy_handler(event: events.EventBase) -> None: + """Dummy event handler function that matches the EventHandlerFunc TypeAlias.""" + + return create_autospec(dummy_handler, spec_set=True) + + +class ApplicationDidLaunchEventFactory(ModelFactory[events.ApplicationDidLaunchEvent]): + """Polyfactory factory for creating fake applicationDidLaunch event message based on our Pydantic model. + + ApplicationDidLaunchEvent's hold no unique identifier properties, besides the almost irrelevant `event` name property. + """ + +class DeviceDidConnectFactory(ModelFactory[events.DeviceDidConnectEvent]): + """Polyfactory factory for creating fake deviceDidConnect event message based on our Pydantic model. + + DeviceDidConnectEvent's have `device` unique identifier property. + """ + +class KeyDownEventFactory(ModelFactory[events.KeyDownEvent]): + """Polyfactory factory for creating fake keyDown event message based on our Pydantic model. + + KeyDownEvent's have the unique identifier properties: + `device`: Identifies the Stream Deck device that this event is associated with. + `action`: Identifies the action that caused the event. + `context`: Identifies the *instance* of an action that caused the event. + """ + + +@pytest.mark.parametrize(("event_name","event_factory"), [ + ("keyDown", KeyDownEventFactory), + ("deviceDidConnect", DeviceDidConnectFactory), + ("applicationDidLaunch", ApplicationDidLaunchEventFactory) +]) +def test_global_action_gets_triggered_by_event( + mock_event_handler: Mock, + event_name: str, + event_factory: ModelFactory[events.EventBase], +): + """Test that a global action's event handlers are triggered by an event. + + Global actions should be triggered by any event type that is registered with them, + regardless of the event's unique identifier properties (or whether they're even present). + """ + fake_event_data = event_factory.build() + + global_action = GlobalAction() + + global_action.on(event_name)(mock_event_handler) + + for handler in global_action.get_event_handlers(event_name): + handler(fake_event_data) + + assert mock_event_handler.call_count == 1 + assert fake_event_data in mock_event_handler.call_args.args + + +@pytest.mark.parametrize(("event_name","event_factory"), [ + ("keyDown", KeyDownEventFactory), + ("deviceDidConnect", DeviceDidConnectFactory), + ("applicationDidLaunch", ApplicationDidLaunchEventFactory) +]) +def test_action_gets_triggered_by_event(mock_event_handler: Mock, event_name: str, event_factory: ModelFactory[events.EventBase]): + # Create a fake event model instance + fake_event_data: events.EventBase = event_factory.build() + # Extract the action UUID from the fake event data, or use a default value + action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else "my-fake-action-uuid" + + action = Action(uuid=action_uuid) + + # Register the mock event handler with the action + action.on(event_name)(mock_event_handler) + + # Get the action's event handlers for the event and call them + for handler in action.get_event_handlers(event_name): + handler(fake_event_data) + + # For some reason, assert_called_once() and assert_called_once_with() are returning False here... + # assert mock_event_handler.assert_called_once(fake_event_data) + assert mock_event_handler.call_count == 1 + assert fake_event_data in mock_event_handler.call_args.args + + + +@pytest.mark.parametrize(("event_name","event_factory"), [ + ("keyDown", KeyDownEventFactory), + ("deviceDidConnect", DeviceDidConnectFactory), + ("applicationDidLaunch", ApplicationDidLaunchEventFactory) +]) +def test_global_action_registry_get_action_handlers_filtering(mock_event_handler: Mock, event_name: str, event_factory: ModelFactory[events.EventBase]): + # Create a fake event model instance + fake_event_data: events.EventBase = event_factory.build() + # Extract the action UUID from the fake event data, or use a default value + action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else None + + registry = ActionRegistry() + # Create an Action instance, without an action UUID as global actions aren't associated with a specific action + global_action = GlobalAction() + + global_action.on(event_name)(mock_event_handler) + + # Register the global action with the registry + registry.register(global_action) + + for handler in registry.get_action_handlers( + event_name=fake_event_data.event, + event_action_uuid=action_uuid, + ): + handler(fake_event_data) + + assert mock_event_handler.call_count == 1 + assert fake_event_data in mock_event_handler.call_args.args + + + +@pytest.mark.parametrize(("event_name","event_factory"), [ + ("keyDown", KeyDownEventFactory), + ("deviceDidConnect", DeviceDidConnectFactory), + ("applicationDidLaunch", ApplicationDidLaunchEventFactory) +]) +def test_action_registry_get_action_handlers_filtering(mock_event_handler: Mock, event_name: str, event_factory: ModelFactory[events.EventBase]): + # Create a fake event model instance + fake_event_data: events.EventBase = event_factory.build() + # Extract the action UUID from the fake event data, or use a default value + action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else None + + registry = ActionRegistry() + # Create an Action instance, using either the fake event's action UUID or a default value + action = Action(uuid=action_uuid or "my-fake-action-uuid") + + action.on(event_name)(mock_event_handler) + + # Register the action with the registry + registry.register(action) + + for handler in registry.get_action_handlers( + event_name=fake_event_data.event, + event_action_uuid=action_uuid, # This will be None if the event is not action-specific (i.e. doesn't have an action UUID property) + ): + handler(fake_event_data) + + assert mock_event_handler.call_count == 1 + assert fake_event_data in mock_event_handler.call_args.args + + + +def test_multiple_actions_filtering(): + registry = ActionRegistry() + action = Action("my-fake-action-uuid-1") + global_action = GlobalAction() + + global_action_event_handler_called = False + action_event_handler_called = False + + @global_action.on("applicationDidLaunch") + def _global_app_did_launch_action_handler(event: events.EventBase): + nonlocal global_action_event_handler_called + global_action_event_handler_called = True + + @action.on("keyDown") + def _action_key_down_event_handler(event: events.EventBase): + nonlocal action_event_handler_called + action_event_handler_called = True + + # Register both actions with the registry + registry.register(global_action) + registry.register(action) + + # Create a fake event model instances + fake_app_did_launch_event_data: events.ApplicationDidLaunchEvent = ApplicationDidLaunchEventFactory.build() + fake_key_down_event_data: events.KeyDownEvent = KeyDownEventFactory.build(action=action.uuid) + + for handler in registry.get_action_handlers(event_name=fake_app_did_launch_event_data.event): + handler(fake_app_did_launch_event_data) + + assert global_action_event_handler_called + assert not action_event_handler_called + + # Reset the flag for global action event handler + global_action_event_handler_called = False + + # Get the action handlers for the event and call them + for handler in registry.get_action_handlers(event_name=fake_key_down_event_data.event, event_action_uuid=fake_key_down_event_data.action): + handler(fake_key_down_event_data) + + assert action_event_handler_called + assert not global_action_event_handler_called diff --git a/tests/actions/test_global_action.py b/tests/actions/test_global_action.py new file mode 100644 index 0000000..5c96169 --- /dev/null +++ b/tests/actions/test_global_action.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from polyfactory.factories.pydantic_factory import ModelFactory +from streamdeck.actions import Action, GlobalAction, available_event_names +from streamdeck.models import events + + +if TYPE_CHECKING: + from streamdeck.models.events import EventBase + from streamdeck.types import EventNameStr + + +def test_global_action_register_event_handler(): + """Test that an event handler can be registered for each valid event name.""" + global_action = GlobalAction() + + for event_name in available_event_names: + @global_action.on(event_name) + def handler(event: EventBase) -> None: + pass + + # Ensure the handler is registered for the correct event name + assert len(global_action._events[event_name]) == 1 + assert handler in global_action._events[event_name] + + +def test_global_action_register_invalid_event_handler(): + """Test that attempting to register an invalid event handler raises an exception.""" + global_action = GlobalAction() + + with pytest.raises(Exception): + @global_action.on("InvalidEvent") + def handler(event: EventBase): + pass + + +def test_global_action_get_event_handlers(): + """Test that the correct event handlers are retrieved for each event name.""" + global_action = GlobalAction() + + for event_name in available_event_names: + # Register a handler for the event name + @global_action.on(event_name) + def handler(event: EventBase): + pass + + # Retrieve the handlers using the generator + handlers = list(global_action.get_event_handlers(event_name)) + + # Ensure that the correct handler is retrieved + assert len(handlers) == 1 + assert handlers[0] == handler + + +def test_global_action_get_event_handlers_no_event_registered(): + """Test that attempting to get handlers for an event with no registered handlers raises an exception.""" + global_action = GlobalAction() + + with pytest.raises(Exception): + list(global_action.get_event_handlers("InvalidEvent")) + + +def test_global_action_register_multiple_handlers_for_event(): + """Test that multiple handlers can be registered for an event.""" + global_action = GlobalAction() + + @global_action.on("keyDown") + def handler_one(event: EventBase): + pass + + @global_action.on("keyDown") + def handler_two(event: EventBase): + pass + + handlers = list(global_action.get_event_handlers("keyDown")) + + # Ensure both handlers are registered for the event + assert len(handlers) == 2 + assert handler_one in handlers + assert handler_two in handlers + + +def test_global_action_get_event_handlers_invalid_event(): + """Test that attempting to get handlers for an invalid event raises a KeyError.""" + global_action = GlobalAction() + + with pytest.raises(KeyError): + list(global_action.get_event_handlers("InvalidEvent")) From 1c966fa2fe625c241e7dc2b193565e491809488c Mon Sep 17 00:00:00 2001 From: strohganoff Date: Tue, 4 Feb 2025 16:31:38 -0600 Subject: [PATCH 2/5] Fix Environment classifiers for pypi --- pyproject.toml | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 3688cc7..ae8e30b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,13 +47,11 @@ "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Development Status :: 4 - Beta", - "Environment :: Desktop", + "Environment :: MacOS X", + "Intended Audience :: Developers", "License :: OSI Approved :: MIT License", - "Topic :: CLI", - "Topic :: Stream Deck", - "Topic :: Stream Deck :: Plugin", - "Topic :: Stream Deck :: Plugin :: SDK", - "Topic :: Stream Deck :: SDK", + "Topic :: Software Development :: Libraries :: Application Frameworks", + "Topic :: Software Development :: Libraries :: Python Modules", "Typing :: Typed", ] requires-python = ">=3.9" From 3dd0955442b6f4dfb35d2884165af6e0e79866af Mon Sep 17 00:00:00 2001 From: strohganoff Date: Wed, 12 Feb 2025 21:45:32 -0600 Subject: [PATCH 3/5] Minor changes for handling new GlobalAction class --- streamdeck/__main__.py | 10 +++++----- streamdeck/actions.py | 6 +++++- streamdeck/manager.py | 4 ++-- 3 files changed, 12 insertions(+), 8 deletions(-) diff --git a/streamdeck/__main__.py b/streamdeck/__main__.py index 3953225..6832762 100644 --- a/streamdeck/__main__.py +++ b/streamdeck/__main__.py @@ -10,7 +10,7 @@ import tomli as toml -from streamdeck.actions import Action +from streamdeck.actions import ActionBase from streamdeck.cli.errors import ( DirectoryNotFoundError, NotAFileError, @@ -146,7 +146,7 @@ def read_streamdeck_config_from_pyproject(plugin_dir: Path) -> StreamDeckConfigD class ActionLoader: @classmethod - def load_actions(cls: type[Self], plugin_dir: Path, files: list[str]) -> Generator[Action, None, None]: + def load_actions(cls: type[Self], plugin_dir: Path, files: list[str]) -> Generator[ActionBase, None, None]: # Ensure the parent directory of the plugin modules is in `sys.path`, # so that import statements in the plugin module will work as expected. if str(plugin_dir) not in sys.path: @@ -191,12 +191,12 @@ def _load_module_from_file(filepath: Path) -> ModuleType: return module @staticmethod - def _get_actions_from_loaded_module(module: ModuleType) -> Generator[Action, None, None]: + def _get_actions_from_loaded_module(module: ModuleType) -> Generator[ActionBase, None, None]: # Iterate over all attributes in the module to find Action subclasses for attribute_name in dir(module): attribute = getattr(module, attribute_name) - # Check if the attribute is an instance of the Action class - if isinstance(attribute, Action): + # Check if the attribute is an instance of the Action class or GlobalAction class. + if issubclass(type(attribute), ActionBase): yield attribute diff --git a/streamdeck/actions.py b/streamdeck/actions.py index a71c529..c7aea1d 100644 --- a/streamdeck/actions.py +++ b/streamdeck/actions.py @@ -1,5 +1,6 @@ from __future__ import annotations +from abc import ABC from collections import defaultdict from functools import cached_property from logging import getLogger @@ -39,7 +40,7 @@ } -class ActionBase: +class ActionBase(ABC): """Base class for all actions.""" def __init__(self): @@ -89,6 +90,9 @@ def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHand msg = f"Provided event name for pulling handlers from action does not exist: {event_name}" raise KeyError(msg) + if event_name not in self._events: + return + yield from self._events[event_name] diff --git a/streamdeck/manager.py b/streamdeck/manager.py index fe7074b..c276c73 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -61,8 +61,8 @@ def register_action(self, action: Action) -> None: Args: action (Action): The action to register. """ - # First, configure a logger for the action, giving it the last part of its uuid as name. - action_component_name = action.uuid.split(".")[-1] + # First, configure a logger for the action, giving it the last part of its uuid as name (if it has one). + action_component_name = action.uuid.split(".")[-1] if hasattr(action, "uuid") else "global" configure_streamdeck_logger(name=action_component_name, plugin_uuid=self.uuid) self._registry.register(action) From 72c526ad3d2169ed506afd4b364b23d292c78b26 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Wed, 12 Feb 2025 22:05:35 -0600 Subject: [PATCH 4/5] Remove 'Event' suffix that was on some event class names. Create mixin classes for event models that have action-specific or device-specific attributes --- streamdeck/manager.py | 4 +- streamdeck/models/events.py | 175 +++++------------- tests/actions/test_action_registry.py | 12 +- tests/actions/test_event_handler_filtering.py | 19 +- tests/data/test_action1.py | 4 +- tests/data/test_action2.py | 4 +- tests/test_plugin_manager.py | 6 +- 7 files changed, 73 insertions(+), 151 deletions(-) diff --git a/streamdeck/manager.py b/streamdeck/manager.py index c276c73..a9f89b8 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -6,7 +6,7 @@ from streamdeck.actions import ActionRegistry from streamdeck.command_sender import StreamDeckCommandSender -from streamdeck.models.events import event_adapter +from streamdeck.models.events import ContextualEventMixin, event_adapter from streamdeck.utils.logging import configure_streamdeck_logger from streamdeck.websocket import WebSocketClient @@ -83,7 +83,7 @@ def run(self) -> None: 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: str | None = cast(str, data.action) if data.is_action_specific() else None + 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): # TODO: from contextual event occurences, save metadata to the action's properties. diff --git a/streamdeck/models/events.py b/streamdeck/models/events.py index 2cd6bf7..60cde3b 100644 --- a/streamdeck/models/events.py +++ b/streamdeck/models/events.py @@ -16,154 +16,94 @@ class EventBase(BaseModel, ABC): event: str """Name of the event used to identify what occurred.""" - @classmethod - def is_action_specific(cls) -> bool: - """Check if the event is specific to an action instance (i.e. the event has an "action" field).""" - return "action" in cls.model_fields - @classmethod - def is_device_specific(cls) -> bool: - """Check if the event is specific to a device instance (i.e. the event has a "device" field).""" - return "device" in cls.model_fields +class ContextualEventMixin: + action: str + """Unique identifier of the action""" + context: str + """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - @classmethod - def is_action_instance_specific(cls) -> bool: - """Check if the event is specific to an action instance (i.e. the event has a "context" field).""" - return "context" in cls.model_fields +class DeviceSpecificEventMixin: + device: str + """Unique identifier of the Stream Deck device that this event is associated with.""" -class ApplicationDidLaunchEvent(EventBase): +class ApplicationDidLaunch(EventBase): event: Literal["applicationDidLaunch"] payload: dict[Literal["application"], str] """Payload containing the name of the application that triggered the event.""" -class ApplicationDidTerminateEvent(EventBase): +class ApplicationDidTerminate(EventBase): event: Literal["applicationDidTerminate"] payload: dict[Literal["application"], str] """Payload containing the name of the application that triggered the event.""" -class DeviceDidConnectEvent(EventBase): +class DeviceDidConnect(EventBase, DeviceSpecificEventMixin): event: Literal["deviceDidConnect"] - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" deviceInfo: dict[str, Any] """Information about the newly connected device.""" -class DeviceDidDisconnectEvent(EventBase): +class DeviceDidDisconnect(EventBase, DeviceSpecificEventMixin): event: Literal["deviceDidDisconnect"] - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" -class DialDownEvent(EventBase): +class DialDown(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): event: Literal["dialDown"] - context: str - """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - action: str - """Unique identifier of the action""" payload: dict[str, Any] -class DialRotateEvent(EventBase): +class DialRotate(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): event: Literal["dialRotate"] - context: str - """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - action: str - """Unique identifier of the action""" payload: dict[str, Any] -class DialUpEvent(EventBase): +class DialUp(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): event: Literal["dialUp"] - context: str - """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - action: str - """Unique identifier of the action""" payload: dict[str, Any] -class DidReceiveDeepLinkEvent(EventBase): +class DidReceiveDeepLink(EventBase): event: Literal["didReceiveDeepLink"] payload: dict[Literal["url"], str] -class DidReceiveGlobalSettingsEvent(EventBase): +class DidReceiveGlobalSettings(EventBase): event: Literal["didReceiveGlobalSettings"] payload: dict[Literal["settings"], dict[str, Any]] -class DidReceivePropertyInspectorMessageEvent(EventBase): +class DidReceivePropertyInspectorMessage(EventBase, ContextualEventMixin): event: Literal["sendToPlugin"] - context: str - """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - action: str - """Unique identifier of the action""" payload: dict[str, Any] -class DidReceiveSettingsEvent(EventBase): +class DidReceiveSettings(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): event: Literal["didReceiveSettings"] - context: str - """UUID of the instance of an action that caused the event.""" - device: str - """UUID of the Stream Deck device that this event is associated with.""" - action: str - """UUID of the action.""" payload: dict[str, Any] -class KeyDownEvent(EventBase): +class KeyDown(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): event: Literal["keyDown"] - context: str - """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - action: str - """Unique identifier of the action""" payload: dict[str, Any] -class KeyUpEvent(EventBase): +class KeyUp(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): event: Literal["keyUp"] - context: str - """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - action: str - """Unique identifier of the action""" payload: dict[str, Any] -class PropertyInspectorDidAppearEvent(EventBase): +class PropertyInspectorDidAppear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): event: Literal["propertyInspectorDidAppear"] - context: str - """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - action: str - """Unique identifier of the action""" -class PropertyInspectorDidDisappearEvent(EventBase): +class PropertyInspectorDidDisappear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): event: Literal["propertyInspectorDidDisappear"] - context: str - """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - action: str - """Unique identifier of the action""" -class SystemDidWakeUpEvent(EventBase): +class SystemDidWakeUp(EventBase): event: Literal["systemDidWakeUp"] @@ -186,46 +126,25 @@ class TitleParametersDidChangePayload(TypedDict): titleParameters: TitleParametersDict -class TitleParametersDidChangeEvent(EventBase): +class TitleParametersDidChange(EventBase, DeviceSpecificEventMixin): event: Literal["titleParametersDidChange"] context: str """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - # payload: dict[str, Any] payload: TitleParametersDidChangePayload -class TouchTap(EventBase): +class TouchTap(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): event: Literal["touchTap"] - context: str - """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - action: str - """Unique identifier of the action""" payload: dict[str, Any] -class WillAppearEvent(EventBase): +class WillAppear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): event: Literal["willAppear"] - context: str - """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - action: str - """Unique identifier of the action""" payload: dict[str, Any] -class WillDisappearEvent(EventBase): +class WillDisappear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): event: Literal["willDisappear"] - context: str - """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - action: str - """Unique identifier of the action""" payload: dict[str, Any] @@ -234,26 +153,26 @@ class WillDisappearEvent(EventBase): event_adapter: TypeAdapter[EventBase] = TypeAdapter( Annotated[ Union[ # noqa: UP007 - ApplicationDidLaunchEvent, - ApplicationDidTerminateEvent, - DeviceDidConnectEvent, - DeviceDidDisconnectEvent, - DialDownEvent, - DialRotateEvent, - DialUpEvent, - DidReceiveDeepLinkEvent, - KeyUpEvent, - KeyDownEvent, - DidReceivePropertyInspectorMessageEvent, - PropertyInspectorDidAppearEvent, - PropertyInspectorDidDisappearEvent, - DidReceiveGlobalSettingsEvent, - DidReceiveSettingsEvent, - SystemDidWakeUpEvent, - TitleParametersDidChangeEvent, + ApplicationDidLaunch, + ApplicationDidTerminate, + DeviceDidConnect, + DeviceDidDisconnect, + DialDown, + DialRotate, + DialUp, + DidReceiveDeepLink, + KeyUp, + KeyDown, + DidReceivePropertyInspectorMessage, + PropertyInspectorDidAppear, + PropertyInspectorDidDisappear, + DidReceiveGlobalSettings, + DidReceiveSettings, + SystemDidWakeUp, + TitleParametersDidChange, TouchTap, - WillAppearEvent, - WillDisappearEvent, + WillAppear, + WillDisappear, ], Field(discriminator="event") ] diff --git a/tests/actions/test_action_registry.py b/tests/actions/test_action_registry.py index 95a8748..aac8797 100644 --- a/tests/actions/test_action_registry.py +++ b/tests/actions/test_action_registry.py @@ -6,13 +6,13 @@ from streamdeck.models import events -class DialUpEventFactory(ModelFactory[events.DialUpEvent]): +class DialUpEventFactory(ModelFactory[events.DialUp]): """Polyfactory factory for creating a fake dialUp event message based on our Pydantic model.""" -class DialDownEventFactory(ModelFactory[events.DialDownEvent]): +class DialDownEventFactory(ModelFactory[events.DialDown]): """Polyfactory factory for creating a fake dialDown event message based on our Pydantic model.""" -class KeyUpEventFactory(ModelFactory[events.KeyUpEvent]): +class KeyUpEventFactory(ModelFactory[events.KeyUp]): """Polyfactory factory for creating a fake keyUp event message based on our Pydantic model.""" @@ -37,7 +37,7 @@ def test_get_action_handlers_no_handlers(): registry.register(action) - fake_event_data: events.DialUpEvent = DialUpEventFactory.build() + fake_event_data: events.DialUp = DialUpEventFactory.build() handlers = list(registry.get_action_handlers(event_name=fake_event_data.event, event_action_uuid=fake_event_data.action)) assert len(handlers) == 0 @@ -53,7 +53,7 @@ def dial_down_handler(event: events.EventBase): registry.register(action) - fake_event_data: events.DialDownEvent = DialDownEventFactory.build(action=action.uuid) + fake_event_data: events.DialDown = DialDownEventFactory.build(action=action.uuid) handlers = list(registry.get_action_handlers(event_name=fake_event_data.event, event_action_uuid=fake_event_data.action)) # handlers = list(registry.get_action_handlers("dialDown")) assert len(handlers) == 1 @@ -78,7 +78,7 @@ def key_up_handler2(event): registry.register(action1) registry.register(action2) - fake_event_data: events.KeyUpEvent = KeyUpEventFactory.build(action=action1.uuid) + fake_event_data: events.KeyUp = KeyUpEventFactory.build(action=action1.uuid) # Notice no action uuid is passed in here, so we should get all handlers for the event. handlers = list(registry.get_action_handlers(event_name=fake_event_data.event)) diff --git a/tests/actions/test_event_handler_filtering.py b/tests/actions/test_event_handler_filtering.py index 8bf7c09..9bd887c 100644 --- a/tests/actions/test_event_handler_filtering.py +++ b/tests/actions/test_event_handler_filtering.py @@ -22,19 +22,19 @@ def dummy_handler(event: events.EventBase) -> None: return create_autospec(dummy_handler, spec_set=True) -class ApplicationDidLaunchEventFactory(ModelFactory[events.ApplicationDidLaunchEvent]): +class ApplicationDidLaunchEventFactory(ModelFactory[events.ApplicationDidLaunch]): """Polyfactory factory for creating fake applicationDidLaunch event message based on our Pydantic model. ApplicationDidLaunchEvent's hold no unique identifier properties, besides the almost irrelevant `event` name property. """ -class DeviceDidConnectFactory(ModelFactory[events.DeviceDidConnectEvent]): +class DeviceDidConnectFactory(ModelFactory[events.DeviceDidConnect]): """Polyfactory factory for creating fake deviceDidConnect event message based on our Pydantic model. DeviceDidConnectEvent's have `device` unique identifier property. """ -class KeyDownEventFactory(ModelFactory[events.KeyDownEvent]): +class KeyDownEventFactory(ModelFactory[events.KeyDown]): """Polyfactory factory for creating fake keyDown event message based on our Pydantic model. KeyDownEvent's have the unique identifier properties: @@ -81,7 +81,8 @@ def test_action_gets_triggered_by_event(mock_event_handler: Mock, event_name: st # Create a fake event model instance fake_event_data: events.EventBase = event_factory.build() # Extract the action UUID from the fake event data, or use a default value - action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else "my-fake-action-uuid" + # action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else "my-fake-action-uuid" + action_uuid: str = fake_event_data.action if isinstance(fake_event_data, events.ContextualEventMixin) else "my-fake-action-uuid" action = Action(uuid=action_uuid) @@ -108,7 +109,8 @@ def test_global_action_registry_get_action_handlers_filtering(mock_event_handler # Create a fake event model instance fake_event_data: events.EventBase = event_factory.build() # Extract the action UUID from the fake event data, or use a default value - action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else None + # action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else None + action_uuid: str | None = fake_event_data.action if isinstance(fake_event_data, events.ContextualEventMixin) else None registry = ActionRegistry() # Create an Action instance, without an action UUID as global actions aren't associated with a specific action @@ -139,7 +141,8 @@ def test_action_registry_get_action_handlers_filtering(mock_event_handler: Mock, # Create a fake event model instance fake_event_data: events.EventBase = event_factory.build() # Extract the action UUID from the fake event data, or use a default value - action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else None + # action_uuid: str = fake_event_data.action if fake_event_data.is_action_specific() else None + action_uuid: str | None = fake_event_data.action if isinstance(fake_event_data, events.ContextualEventMixin) else None registry = ActionRegistry() # Create an Action instance, using either the fake event's action UUID or a default value @@ -184,8 +187,8 @@ def _action_key_down_event_handler(event: events.EventBase): registry.register(action) # Create a fake event model instances - fake_app_did_launch_event_data: events.ApplicationDidLaunchEvent = ApplicationDidLaunchEventFactory.build() - fake_key_down_event_data: events.KeyDownEvent = KeyDownEventFactory.build(action=action.uuid) + fake_app_did_launch_event_data: events.ApplicationDidLaunch = ApplicationDidLaunchEventFactory.build() + fake_key_down_event_data: events.KeyDown = KeyDownEventFactory.build(action=action.uuid) for handler in registry.get_action_handlers(event_name=fake_app_did_launch_event_data.event): handler(fake_app_did_launch_event_data) diff --git a/tests/data/test_action1.py b/tests/data/test_action1.py index 6f9a29d..ccab56f 100644 --- a/tests/data/test_action1.py +++ b/tests/data/test_action1.py @@ -1,10 +1,10 @@ from streamdeck.actions import Action -from streamdeck.models.events import KeyDownEvent +from streamdeck.models.events import KeyDown test_action1 = Action("my-first-test-action") @test_action1.on("keyDown") -def handle_key_down_event(event_data: KeyDownEvent) -> None: +def handle_key_down_event(event_data: KeyDown) -> None: print("KeyDown event handled.") diff --git a/tests/data/test_action2.py b/tests/data/test_action2.py index 8328355..4ede7fd 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 ApplicationDidTerminateEvent +from streamdeck.models.events import ApplicationDidTerminate test_action2 = Action("my-first-test-action") @test_action2.on("applicationDidTerminate") -def handle_application_terminated_event(event_data: ApplicationDidTerminateEvent) -> None: +def handle_application_terminated_event(event_data: ApplicationDidTerminate) -> None: print("ApplicationDidTerminate event handled.") diff --git a/tests/test_plugin_manager.py b/tests/test_plugin_manager.py index d2bcaae..4d96209 100644 --- a/tests/test_plugin_manager.py +++ b/tests/test_plugin_manager.py @@ -7,11 +7,11 @@ from polyfactory.factories.pydantic_factory import ModelFactory from streamdeck.actions import Action from streamdeck.manager import PluginManager -from streamdeck.models.events import DialRotateEvent, EventBase, event_adapter +from streamdeck.models.events import DialRotate, EventBase, event_adapter from streamdeck.websocket import WebSocketClient -class DialRotateEventFactory(ModelFactory[DialRotateEvent]): +class DialRotateEventFactory(ModelFactory[DialRotate]): """Polyfactory factory for creating a fake event message based on our Pydantic model.""" @@ -51,7 +51,7 @@ def patch_websocket_client(monkeypatch: pytest.MonkeyPatch) -> tuple[MagicMock, mock_websocket_client.__enter__.return_value = mock_websocket_client # Create a fake event message, and convert it to a json string to be passed back by the client.listen_forever() method. - fake_event_message: DialRotateEvent = DialRotateEventFactory.build() + fake_event_message: DialRotate = DialRotateEventFactory.build() mock_websocket_client.listen_forever.return_value = [fake_event_message.model_dump_json()] monkeypatch.setattr("streamdeck.manager.WebSocketClient", lambda port: mock_websocket_client) From d1291b15fc1c75c6f5f458fcc9f26b9229e17118 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Wed, 12 Feb 2025 23:24:39 -0600 Subject: [PATCH 5/5] Update README.md with documentation for Global Actions --- README.md | 53 ++++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 42 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 2dbbb15..9b5512c 100644 --- a/README.md +++ b/README.md @@ -41,9 +41,13 @@ This guide will help you set up your first Stream Deck plugin using the library. - Stream Deck software installed - A valid `manifest.json` file for your plugin -### Creating an Action +### Creating Actions -An **Action** represents a specific functionality in your plugin. You can create multiple actions, each with its own set of event handlers. +The SDK provides two types of actions: `Action` and `GlobalAction`. Each represents functionality with different scopes in your plugin, determining how events are handled. + +#### Regular Actions + +An `Action` handles events that are specifically associated with it based on event metadata. When the Stream Deck sends an event, the action's handlers only run if the event metadata indicates it was triggered by or is intended for that specific action instance. ```python from streamdeck import Action @@ -52,18 +56,33 @@ from streamdeck import Action my_action = Action(uuid="com.example.myplugin.myaction") ``` +#### Global Actions + +A `GlobalAction` runs its event handlers for all events of a given type, regardless of which action the events were originally intended for. Unlike regular Actions which only process events specifically targeted at their UUID, GlobalActions handle events meant for any action in the plugin, making them useful for implementing plugin-wide behaviors or monitoring. + +```python +from streamdeck import GlobalAction + +# Create a global action +my_global_action = GlobalAction() +``` + +Choose `GlobalAction` when you want to handle events at the plugin-scope (i.e. globally) without filtering by action, and `Action` when you need to process events specific to particular actions. + +Note that an action with its UUID still needs to be defined in the manifest.json. Global Actions are an abstract component unique to this library — the global behavior is not how the Stream Deck software itself handles registering actions and publishing events. + ### Registering Event Handlers Use the `.on()` method to register event handlers for specific events. ```python @my_action.on("keyDown") -def handle_key_down(event): - print("Key Down event received:", event) +def handle_key_down(event_data): + print("Key Down event received:", event_data) @my_action.on("willAppear") -def handle_will_appear(event): - print("Will Appear event received:", event) +def handle_will_appear(event_data): + print("Will Appear event received:", event_data) ``` !!!INFO Handlers for action-specific events are dispatched only if the event is triggered by the associated action, ensuring isolation and predictability. For other types of events that are not associated with a specific action, handlers are dispatched without such restrictions. @@ -159,22 +178,34 @@ Below is an example of the pyproject.toml configuration and how to run the plugi ## Simple Example -Below is a complete example that creates a plugin with a single action. The action handles the `keyDown` event and simply prints a statement that the event occurred. +Below is a complete example that creates a plugin with a single action. The action handles the `keyDown` and `applicationDidLaunch` event and simply prints a statement that an event occurred. ```python # main.py import logging -from streamdeck import Action, PluginManager, events +from streamdeck import Action, GlobalAction, PluginManager, events logger = logging.getLogger("myaction") # Define your action my_action = Action(uuid="com.example.myplugin.myaction") -# Register event handlers +# Define your global action +my_global_action = GlobalAction() + +# Register event handlers for regular action +@my_action.on("applicationDidLaunch") +def handle_application_did_launch(event_data: events.ApplicationDidLaunch): + logger.debug("Application Did Launch event recieved:", event_data) + @my_action.on("keyDown") -def handle_key_down(event): - logger.debug("Key Down event received:", event) +def handle_key_down(event_data: events.KeyDown): + logger.debug("Key Down event received:", event_data) + +# Register event handlers for global action +@my_global_action.on("keyDown") +def handle_global_key_down(event_data: events.KeyDown): + logger.debug("Global Key Down event received:", event_data) ``` ```toml