diff --git a/README.md b/README.md index 5f125d5..5224a13 100644 --- a/README.md +++ b/README.md @@ -121,7 +121,24 @@ def resolve_api_client(event_data): return api_client ``` -The resolver can return a module-level instance or compute an event-specific value. `scope` defaults to `Plugin` when omitted. +The resolver can return a module-level instance or compute an event-specific value. Its `event_data` argument is the event being handled. `scope` declares what identity the event must provide for the parameter to be injected; it defaults to `Plugin` when omitted: + +- `Plugin`: no device or action-instance identity is needed, so the parameter is available for every event. +- `Device`: the event must identify a device. Device events and action-instance events that include a device qualify. +- `ActionInstance`: the event must identify an action instance; a device is optional. +- `ActionInstanceWithDevice`: the event must identify both an action instance and a device. + +`Device` and `ActionInstance` are separate identity capabilities. `ActionInstanceWithDevice` provides both, so it can satisfy either requirement. When handlers are registered, the SDK checks their event types against the requested scope and raises `TypeError` if an event cannot supply the required identity. + +The SDK also provides built-in context parameters, which do not need a resolver entry in `pyproject.toml`. Request one by adding its name to an action handler's signature: + +```python +def handle_will_appear(event_data, action_instance_context): + settings = action_instance_context.settings + current_state = action_instance_context.state +``` + +The SDK seeds `plugin_context` with launch information and updates the contexts from relevant Stream Deck events before user handlers run. `plugin_context` includes the plugin UUID, registration UUID, launch info, and global settings. `device_context` includes a device ID, connection status, and the latest device information. `action_context` identifies an action by UUID. `action_instance_context` identifies one action instance and holds its latest known settings, state, controller, coordinates, and associated action and device. Until an event supplies a value, context fields retain their empty or `None` defaults. The SDK rejects handlers that request a context unavailable for one of their event types. diff --git a/streamdeck/__init__.py b/streamdeck/__init__.py index 906a1b5..4bd52c8 100644 --- a/streamdeck/__init__.py +++ b/streamdeck/__init__.py @@ -1,5 +1,6 @@ from . import ( command_sender, + contexts, manager, models, utils, @@ -11,6 +12,7 @@ __all__ = [ "actions", "command_sender", + "contexts", "manager", "models", "utils", diff --git a/streamdeck/__main__.py b/streamdeck/__main__.py index d9f455a..0975344 100644 --- a/streamdeck/__main__.py +++ b/streamdeck/__main__.py @@ -4,11 +4,15 @@ import logging import sys -from typing import TYPE_CHECKING, TypeVar from pydantic import ValidationError -from pydantic_settings import SettingsError +from pydantic_settings import CliApp, SettingsError +from streamdeck.contexts import ( + ContextStore, + register_context_handlers, + register_context_injectables, +) from streamdeck.manager import PluginManager from streamdeck.models.configs import StreamDeckToolConfig from streamdeck.models.launch import LaunchArguments @@ -16,24 +20,13 @@ logger = logging.getLogger("streamdeck") -SettingsT = TypeVar("SettingsT") - -if TYPE_CHECKING: - from collections.abc import Callable - - -def _parse_settings_from_cli( - settings_cls: Callable[..., SettingsT], args: list[str] -) -> SettingsT: - """Instantiate Pydantic Settings with its private CLI argument hook.""" - return settings_cls(_cli_parse_args=args) def main(argv: list[str] | None = None) -> None: """Console entry point for the Pydantic Settings CLI.""" args = sys.argv[1:] if argv is None else list(argv) try: - launch = _parse_settings_from_cli(LaunchArguments, args) + launch = CliApp.run(LaunchArguments, cli_args=args) except (ValidationError, SettingsError) as exc: logger.error("Invalid launch arguments: %s", exc) raise SystemExit(2) from exc @@ -50,7 +43,7 @@ def main(argv: list[str] | None = None) -> None: if launch.debug_port: setup_debug_mode(launch.debug_port) - config = _parse_settings_from_cli(StreamDeckToolConfig, args) + config = CliApp.run(StreamDeckToolConfig, cli_args=args) run_plugin(launch, config) except Exception: logger.exception("Plugin startup or execution failed") @@ -76,6 +69,12 @@ def run_plugin(launch: LaunchArguments, config: StreamDeckToolConfig) -> None: register_event=launch.register_event, ) + context_store = ContextStore( + plugin_registration_uuid=launch.plugin_registration_uuid, + info=launch.info, + ) + manager.register_action(register_context_handlers(context_store)) + for name, param_config in config.injectable_handler_params.items(): manager.register_injectable_handler_param( name, @@ -83,6 +82,8 @@ def run_plugin(launch: LaunchArguments, config: StreamDeckToolConfig) -> None: scope=param_config.scope, ) + register_context_injectables(context_store, manager.register_injectable_handler_param) + # Register custom event types before validating action handlers. for listener_type in config.event_listeners: manager.register_event_listener(listener_type()) diff --git a/streamdeck/contexts.py b/streamdeck/contexts.py new file mode 100644 index 0000000..785684a --- /dev/null +++ b/streamdeck/contexts.py @@ -0,0 +1,381 @@ +"""Context objects built from Stream Deck startup arguments and events.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from functools import singledispatchmethod +from typing import TYPE_CHECKING, Protocol + +from streamdeck.event_handlers.actions import GlobalAction +from streamdeck.models.events import DEFAULT_EVENT_MODELS +from streamdeck.models.events.common import ( + BasePayload, + ContextualEventMixin, + ContextualPayloadEventMixin, + ControllerType, + Coordinates, + CoordinatesPayloadMixin, + DeviceSpecificEventMixin, + SettingsPayloadMixin, + StatefulActionPayloadMixin, +) +from streamdeck.models.events.devices import DeviceDidConnect, DeviceDidDisconnect, DeviceInfo +from streamdeck.models.events.property_inspector import DidReceivePropertyInspectorMessage +from streamdeck.models.events.settings import DidReceiveGlobalSettings + + +if TYPE_CHECKING: + from streamdeck.event_handlers.injectables import InjectableResolver + from streamdeck.event_handlers.scopes import ScopeName + from streamdeck.models.events.base import EventBase + from streamdeck.models.launch import RegistrationInfo + from streamdeck.types import PluginDefinedData + + +class _InjectableHandlerParamRegistrar(Protocol): + """Registration callable accepted by the context injectable setup helper.""" + + def __call__( + self, + name: str, + resolver: InjectableResolver[object], + *, + scope: ScopeName | None = None, + ) -> None: + """Register a resolver under a handler parameter name and scope.""" + ... + + +@dataclass(slots=True) +class PluginContext: + """Plugin identity, startup metadata, and global settings.""" + + plugin_uuid: str + """UUID configured for the plugin in its manifest.""" + plugin_registration_uuid: str + """Per-launch UUID supplied by Stream Deck for plugin registration.""" + info: PluginDefinedData + """Plugin information passed in the launch arguments.""" + global_settings: PluginDefinedData = field(default_factory=dict) + """Most recently received global plugin settings.""" + + +@dataclass(slots=True) +class DeviceContext: + """Information and connection status for one Stream Deck device.""" + + device_id: str + """Device identifier supplied by Stream Deck events.""" + info: DeviceInfo | None = None + """Device details from the most recent deviceDidConnect event.""" + connected: bool | None = None + """Whether Stream Deck has reported this device as connected.""" + + +@dataclass(slots=True) +class ActionContext: + """Identity shared by all instances of one registered action.""" + + action_uuid: str + """Action UUID supplied by Stream Deck events.""" + + +@dataclass(slots=True) +class ActionInstanceContext: + """Latest known settings and placement data for one action instance.""" + + context_id: str + """Context identifier supplied by Stream Deck events.""" + action: ActionContext + """Action that owns this instance.""" + device: DeviceContext | None = None + """Associated device when an event identifies one.""" + settings: PluginDefinedData = field(default_factory=dict) + """Most recently received settings for this action instance.""" + state: int | None = None + """Most recently reported state, when the action has multiple states.""" + controller: ControllerType | None = None + """Most recently reported controller type.""" + coordinates: Coordinates | None = None + """Most recently reported key or dial coordinates.""" + + +class ContextStore: + """Cache plugin, device, action, and action-instance contexts.""" + + def __init__( + self, + plugin_registration_uuid: str, + info: RegistrationInfo, + ) -> None: + """Initialize the context cache with plugin startup metadata. + + Args: + plugin_registration_uuid: Per-launch registration UUID supplied by Stream Deck. + info: Validated registration information from the startup arguments. + """ + self.plugin = PluginContext( + plugin_uuid=info.plugin_uuid, + plugin_registration_uuid=plugin_registration_uuid, + info=info.model_dump(), + ) + self.devices: dict[str, DeviceContext] = {} + self.actions: dict[str, ActionContext] = {} + self.action_instances: dict[str, ActionInstanceContext] = {} + + @singledispatchmethod + def update_from_event(self, event_data: EventBase) -> None: # noqa: ARG002 + """Ignore event models that do not update a context. + + Registered implementations update global settings, device data, or action-instance + data according to the concrete event model. + + Args: + event_data: Validated event received from Stream Deck. + """ + return + + @update_from_event.register + def _update_global_settings(self, event_data: DidReceiveGlobalSettings) -> None: + """Replace cached plugin settings with settings received from Stream Deck. + + Args: + event_data: Event containing the latest global plugin settings. + """ + self.plugin.global_settings = event_data.payload.settings + + @update_from_event.register + def _update_device_connected(self, event_data: DeviceDidConnect) -> None: + """Record device information and mark the device as connected. + + Args: + event_data: Event identifying the connected device and its information. + """ + device = self.get_device_context(event_data.device) + device.info = event_data.device_info + device.connected = True + + @update_from_event.register + def _update_device_disconnected(self, event_data: DeviceDidDisconnect) -> None: + """Mark the identified device as disconnected. + + Args: + event_data: Event identifying the disconnected device. + """ + self.get_device_context(event_data.device).connected = False + + @update_from_event.register + def _update_contextual_event(self, event_data: ContextualEventMixin) -> None: + """Update action-instance identity for a contextual event. + + Args: + event_data: Event identifying an action and one of its instances. + """ + self._get_or_create_action_instance_context(event_data) + + @update_from_event.register + def _update_contextual_payload_event( + self, + event_data: ContextualPayloadEventMixin, + ) -> None: + """Update action-instance context values from a contextual event payload. + + Args: + event_data: Contextual event with payload data that may update the context. + """ + instance = self._get_or_create_action_instance_context(event_data) + payload = event_data.payload + + # SettingsPayloadMixin is used by didReceiveSettings, keyDown/keyUp, willAppear/ + # willDisappear, dialDown/dialRotate/dialUp, titleParametersDidChange, and touchTap. + if isinstance(payload, SettingsPayloadMixin): + instance.settings = payload.settings + + # BasePayload subclasses on those events also carry controller information. + if isinstance(payload, BasePayload) and payload.controller is not None: + instance.controller = payload.controller + + # Stateful payloads are used by keyDown/keyUp, didReceiveSettings, willAppear/ + # willDisappear, and titleParametersDidChange. + if isinstance(payload, StatefulActionPayloadMixin) and payload.state is not None: + instance.state = payload.state + + # Coordinates occur on single-action didReceiveSettings, keyDown/keyUp, and + # willAppear/willDisappear payloads, plus dialDown/dialRotate/dialUp, + # titleParametersDidChange, and touchTap payloads. + if isinstance(payload, CoordinatesPayloadMixin): + instance.coordinates = payload.coordinates + + def get_device_context(self, device_id: str) -> DeviceContext: + """Get a device context, creating a placeholder if it is not cached. + + Args: + device_id: Stream Deck identifier for the device. + + Returns: + The cached context for the requested device. + """ + if device_id not in self.devices: + self.devices[device_id] = DeviceContext(device_id=device_id) + return self.devices[device_id] + + def get_action_context(self, action_uuid: str) -> ActionContext: + """Get an action context, creating it if it is not cached. + + Args: + action_uuid: UUID identifying the action. + + Returns: + The cached context for the requested action. + """ + if action_uuid not in self.actions: + self.actions[action_uuid] = ActionContext(action_uuid=action_uuid) + return self.actions[action_uuid] + + def _get_or_create_action_instance_context( + self, + event_data: ContextualEventMixin, + ) -> ActionInstanceContext: + """Get or create an action-instance context using event identity fields. + + Args: + event_data: Event identifying the action instance and, when present, its device. + + Returns: + The cached context for the identified action instance. + """ + action = self.get_action_context(event_data.action) + device = ( + self.get_device_context(event_data.device) + if isinstance(event_data, DeviceSpecificEventMixin) + else None + ) + instance = self.action_instances.get(event_data.context) + if instance is None: + instance = ActionInstanceContext( + context_id=event_data.context, + action=action, + device=device, + ) + self.action_instances[event_data.context] = instance + else: + instance.action = action + if device is not None: + instance.device = device + return instance + + def resolve_device_context(self, event_data: EventBase) -> DeviceContext: + """Resolve the device context identified by a device-specific event. + + Args: + event_data: Event expected to identify a device. + + Returns: + The cached context for the event's device. + + Raises: + TypeError: If the event does not identify a device. + """ + if not isinstance(event_data, DeviceSpecificEventMixin): + msg = f"Event {event_data.event!r} does not identify a device" + raise TypeError(msg) + return self.get_device_context(event_data.device) + + def resolve_action_context(self, event_data: EventBase) -> ActionContext: + """Resolve the action context identified by a contextual event. + + Args: + event_data: Event expected to identify an action. + + Returns: + The cached context for the event's action. + + Raises: + TypeError: If the event does not identify an action. + """ + if not isinstance(event_data, ContextualEventMixin): + msg = f"Event {event_data.event!r} does not identify an action" + raise TypeError(msg) + return self.get_action_context(event_data.action) + + def resolve_action_instance_context(self, event_data: EventBase) -> ActionInstanceContext: + """Resolve the action-instance context identified by a contextual event. + + Args: + event_data: Event expected to identify an action instance. + + Returns: + The cached context for the event's action instance. + + Raises: + TypeError: If the event does not identify an action instance. + """ + if not isinstance(event_data, ContextualEventMixin): + msg = f"Event {event_data.event!r} does not identify an action instance" + raise TypeError(msg) + + return self._get_or_create_action_instance_context(event_data) + + +def register_context_handlers(context_store: ContextStore) -> GlobalAction: + """Register context updates for relevant default Stream Deck event models. + + Args: + context_store: Store updated by the internal event handlers. + + Returns: + A global action containing the internal context update handlers. + """ + context_actions = GlobalAction() + + def update_context(event_data: EventBase) -> None: + """Forward event data through the store's singledispatch method.""" + context_store.update_from_event(event_data) + + for event_model in DEFAULT_EVENT_MODELS: + # Skip sendToPlugin: its arbitrary payload is the only contextual event payload not used + # to update cached context values. + if event_model is DidReceivePropertyInspectorMessage: + continue + + # Only register events that identify an action instance, a device, or global settings. + if not issubclass( + event_model, + (ContextualEventMixin, DeviceSpecificEventMixin, DidReceiveGlobalSettings), + ): + continue + + for event_name in event_model.get_model_event_names(): + context_actions.on(event_name)(update_context) + + return context_actions + + +def register_context_injectables( + context_store: ContextStore, + register_injectable: _InjectableHandlerParamRegistrar, +) -> None: + """Register built-in context resolvers with their event identity scopes. + + Args: + context_store: Store that resolves the context objects. + register_injectable: Registration callable for the built-in injectable resolvers. + """ + register_injectable( + "plugin_context", + lambda _event_data: context_store.plugin, + scope="Plugin", + ) + register_injectable( + "device_context", context_store.resolve_device_context, scope="Device" + ) + register_injectable( + "action_context", + context_store.resolve_action_context, + scope="ActionInstance", + ) + register_injectable( + "action_instance_context", + context_store.resolve_action_instance_context, + scope="ActionInstance", + ) diff --git a/tests/test_launch_settings.py b/tests/test_launch_settings.py index e20c9b3..3bcd299 100644 --- a/tests/test_launch_settings.py +++ b/tests/test_launch_settings.py @@ -1,7 +1,7 @@ import json import subprocess import sys -from unittest.mock import Mock +from unittest.mock import Mock, call import pytest from pydantic import ValidationError @@ -242,6 +242,11 @@ def test_launcher_imports_from_plugin_directory_and_registers_listeners_first( try: entry.main(args) if explicit_args else entry.main() assert [call[0] for call in manager.mock_calls] == [ + "register_action", + "register_injectable_handler_param", + "register_injectable_handler_param", + "register_injectable_handler_param", + "register_injectable_handler_param", "register_event_listener", "register_action", "run", @@ -261,11 +266,11 @@ def test_main_logs_startup_and_runtime_failures(monkeypatch, caplog, phase): error = RuntimeError(f"{phase} failed") listener_type = Mock() config = Mock(event_listeners=[listener_type], actions=[object()], injectable_handler_params={}) - settings_factory = Mock(return_value=config) + launch = LaunchArguments(_cli_parse_args=host_args()) + settings_result = error if phase == "settings" else config + cli_app_run = Mock(side_effect=[launch, settings_result]) manager = Mock() - if phase == "settings": - settings_factory.side_effect = error - elif phase == "listener": + if phase == "listener": listener_type.side_effect = error elif phase == "action": manager.register_action.side_effect = error @@ -274,11 +279,15 @@ def test_main_logs_startup_and_runtime_failures(monkeypatch, caplog, phase): monkeypatch.setattr(sys, "path", list(sys.path)) monkeypatch.setattr(entry, "configure_streamdeck_logger", Mock()) - monkeypatch.setattr(entry, "StreamDeckToolConfig", settings_factory) + monkeypatch.setattr(entry.CliApp, "run", cli_app_run) monkeypatch.setattr(entry, "PluginManager", Mock(return_value=manager)) with pytest.raises(RuntimeError) as exc: entry.main(host_args()) assert exc.value is error + assert cli_app_run.call_args_list == [ + call(LaunchArguments, cli_args=host_args()), + call(StreamDeckToolConfig, cli_args=host_args()), + ] assert "Plugin startup or execution failed" in caplog.text @@ -300,6 +309,11 @@ def test_run_plugin_uses_supplied_settings_without_reading_process_args(tmp_path monkeypatch.setattr(entry, "PluginManager", Mock(return_value=manager)) entry.run_plugin(launch, config) assert [call[0] for call in manager.mock_calls] == [ + "register_action", + "register_injectable_handler_param", + "register_injectable_handler_param", + "register_injectable_handler_param", + "register_injectable_handler_param", "register_event_listener", "register_action", "run",