diff --git a/pyproject.toml b/pyproject.toml index f5cafd1..fef294a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,8 +58,8 @@ "dependencies" = [ # "jmespath >= 1.0.1", # Upcoming pattern-filtering functionality "platformdirs >= 4.3.6", - "pydantic >= 2.9.2", - "pydantic_core >= 2.23.4", + "pydantic >= 2.11", + "pydantic_core >= 2.33", "tomli >= 2.0.2", "websockets >= 13.1", "typer >= 0.15.1", diff --git a/streamdeck/manager.py b/streamdeck/manager.py index e74ff63..7746a8d 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -9,7 +9,8 @@ from streamdeck.actions import Action, ActionBase, ActionRegistry from streamdeck.command_sender import StreamDeckCommandSender from streamdeck.event_listener import EventListener, EventListenerManager -from streamdeck.models.events import ContextualEventMixin, EventAdapter +from streamdeck.models.events.adapter import EventAdapter +from streamdeck.models.events.common import ContextualEventMixin from streamdeck.types import ( EventHandlerBasicFunc, EventHandlerFunc, diff --git a/streamdeck/models/events.py b/streamdeck/models/events.py deleted file mode 100644 index b13789d..0000000 --- a/streamdeck/models/events.py +++ /dev/null @@ -1,306 +0,0 @@ -from __future__ import annotations - -from abc import ABC -from typing import Annotated, Any, Final, Literal, Union, get_args, get_type_hints - -from pydantic import BaseModel, ConfigDict, Field, TypeAdapter -from typing_extensions import LiteralString, TypedDict, TypeIs # noqa: UP035 - - -# TODO: Create more explicitly-defined payload objects. - - -class EventBase(BaseModel, ABC): - """Base class for event models that represent Stream Deck Plugin SDK events.""" - # Configure to use the docstrings of the fields as the field descriptions. - model_config = ConfigDict(use_attribute_docstrings=True) - - event: str - """Name of the event used to identify what occurred. - - Subclass models must define this field as a Literal type with the event name string that the model represents. - """ - - def __init_subclass__(cls, **kwargs: Any) -> None: - """Validate that the event field is a Literal[str] type.""" - super().__init_subclass__(**kwargs) - - model_event_type = get_type_hints(cls)["event"] - - if not is_literal_str_type(model_event_type): - msg = f"The event field annotation must be a Literal[str] type. Given type: {model_event_type}" - raise TypeError(msg) - - @classmethod - def get_model_event_name(cls) -> tuple[str, ...]: - """Get the value of the subclass model's event field Literal annotation.""" - model_event_type = get_type_hints(cls)["event"] - - # Ensure that the event field annotation is a Literal type. - if not is_literal_str_type(model_event_type): - msg = "The `event` field annotation of an Event model must be a Literal[str] type." - raise TypeError(msg) - - return get_args(model_event_type) - - -def is_literal_str_type(value: object | None) -> TypeIs[LiteralString]: - """Check if a type is a Literal type.""" - if value is None: - return False - - event_field_base_type = getattr(value, "__origin__", None) - - if event_field_base_type is not Literal: - return False - - return all(isinstance(literal_value, str) for literal_value in get_args(value)) - - -## Mixin classes for common event model fields. - -class ContextualEventMixin: - """Mixin class for event models that have action and context fields.""" - 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.""" - -class DeviceSpecificEventMixin: - """Mixin class for event models that have a device field.""" - device: str - """Unique identifier of the Stream Deck device that this event is associated with.""" - - -## EventBase implementation models of the Stream Deck Plugin SDK events. - -class ApplicationDidLaunch(EventBase): - 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"] # 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"] # type: ignore[override] - deviceInfo: dict[str, Any] - """Information about the newly connected device.""" - - -class DeviceDidDisconnect(EventBase, DeviceSpecificEventMixin): - event: Literal["deviceDidDisconnect"] # type: ignore[override] - - -class DialDown(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["dialDown"] # type: ignore[override] - payload: dict[str, Any] - - -class DialRotate(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["dialRotate"] # type: ignore[override] - payload: dict[str, Any] - - -class DialUp(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["dialUp"] # type: ignore[override] - payload: dict[str, Any] - - -class DidReceiveDeepLink(EventBase): - event: Literal["didReceiveDeepLink"] # type: ignore[override] - payload: dict[Literal["url"], str] - - -class DidReceiveGlobalSettings(EventBase): - event: Literal["didReceiveGlobalSettings"] # type: ignore[override] - payload: dict[Literal["settings"], dict[str, Any]] - - -class DidReceivePropertyInspectorMessage(EventBase, ContextualEventMixin): - event: Literal["sendToPlugin"] # type: ignore[override] - payload: dict[str, Any] - - -class DidReceiveSettings(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["didReceiveSettings"] # type: ignore[override] - payload: dict[str, Any] - - -class KeyDown(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["keyDown"] # type: ignore[override] - payload: dict[str, Any] - - -class KeyUp(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["keyUp"] # type: ignore[override] - payload: dict[str, Any] - - -class PropertyInspectorDidAppear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["propertyInspectorDidAppear"] # type: ignore[override] - - -class PropertyInspectorDidDisappear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["propertyInspectorDidDisappear"] # type: ignore[override] - - -class SystemDidWakeUp(EventBase): - event: Literal["systemDidWakeUp"] # type: ignore[override] - - -class TitleParametersDict(TypedDict): - fontFamily: str - fontSize: int - fontStyle: str - fontUnderline: bool - showTitle: bool - titleAlignment: Literal["top", "middle", "bottom"] - titleColor: str - - -class TitleParametersDidChangePayload(TypedDict): - controller: Literal["Keypad", "Encoder"] - coordinates: dict[Literal["column", "row"], int] - settings: dict[str, Any] - state: int - title: str - titleParameters: TitleParametersDict - - -class TitleParametersDidChange(EventBase, DeviceSpecificEventMixin): - 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"] # type: ignore[override] - payload: dict[str, Any] - - -class WillAppear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["willAppear"] # type: ignore[override] - payload: dict[str, Any] - - -class WillDisappear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): - event: Literal["willDisappear"] # type: ignore[override] - payload: dict[str, Any] - - -## Default event models and names. - - -event_adapter: TypeAdapter[EventBase] = TypeAdapter( - Annotated[ - Union[ # noqa: UP007 - ApplicationDidLaunch, - ApplicationDidTerminate, - DeviceDidConnect, - DeviceDidDisconnect, - DialDown, - DialRotate, - DialUp, - DidReceiveDeepLink, - KeyUp, - KeyDown, - DidReceivePropertyInspectorMessage, - PropertyInspectorDidAppear, - PropertyInspectorDidDisappear, - DidReceiveGlobalSettings, - DidReceiveSettings, - SystemDidWakeUp, - TitleParametersDidChange, - TouchTap, - WillAppear, - WillDisappear, - ], - Field(discriminator="event") - ] -) - - -DEFAULT_EVENT_MODELS: Final[list[type[EventBase]]] = [ - ApplicationDidLaunch, - ApplicationDidTerminate, - DeviceDidConnect, - DeviceDidDisconnect, - DialDown, - DialRotate, - DialUp, - DidReceiveDeepLink, - KeyUp, - KeyDown, - DidReceivePropertyInspectorMessage, - PropertyInspectorDidAppear, - PropertyInspectorDidDisappear, - DidReceiveGlobalSettings, - DidReceiveSettings, - SystemDidWakeUp, - TitleParametersDidChange, - TouchTap, - WillAppear, - WillDisappear, -] - - -def _get_default_event_names() -> set[str]: - default_event_names: set[str] = set() - - for event_model in DEFAULT_EVENT_MODELS: - default_event_names.update(event_model.get_model_event_name()) - - return default_event_names - - -DEFAULT_EVENT_NAMES: Final[set[str]] = _get_default_event_names() - - -## EventAdapter class for handling and extending available event models. - -class EventAdapter: - """TypeAdapter-encompassing class for handling and extending available event models.""" - def __init__(self) -> None: - self._models: list[type[EventBase]] = [] - self._type_adapter: TypeAdapter[EventBase] | None = None - - self._event_names: set[str] = set() - """A set of all event names that have been registered with the adapter. - This set starts out containing the default event models defined by the library. - """ - - for model in DEFAULT_EVENT_MODELS: - self.add_model(model) - - def add_model(self, model: type[EventBase]) -> None: - """Add a model to the adapter, and add the event name of the model to the set of registered event names.""" - self._models.append(model) - self._event_names.update(model.get_model_event_name()) - - def event_name_exists(self, event_name: str) -> bool: - """Check if an event name has been registered with the adapter.""" - return event_name in self._event_names - - @property - def type_adapter(self) -> TypeAdapter[EventBase]: - """Get the TypeAdapter instance for the event models.""" - if self._type_adapter is None: - self._type_adapter = TypeAdapter( - Annotated[ - Union[tuple(self._models)], # noqa: UP007 - Field(discriminator="event") - ] - ) - - return self._type_adapter - - def validate_json(self, data: str | bytes) -> EventBase: - """Validate a JSON string or bytes object as an event model.""" - return self.type_adapter.validate_json(data) - diff --git a/streamdeck/models/events/__init__.py b/streamdeck/models/events/__init__.py new file mode 100644 index 0000000..75f86ef --- /dev/null +++ b/streamdeck/models/events/__init__.py @@ -0,0 +1,96 @@ +"""Stream Deck Event Models. + +These models are used to represent the events that can be received by the Stream Deck SDK plugin. +The "default" events are those passed from the Stream Deck software itself, but custom events can +be created by the plugin developer and listened for in the same way. + +The events are organized into the same categories as the Stream Deck SDK documentation defines them. +""" +from __future__ import annotations + +from typing import TYPE_CHECKING + +from streamdeck.models.events.application import ApplicationDidLaunch, ApplicationDidTerminate +from streamdeck.models.events.base import EventBase +from streamdeck.models.events.deep_link import DidReceiveDeepLink +from streamdeck.models.events.devices import DeviceDidConnect, DeviceDidDisconnect +from streamdeck.models.events.dials import DialDown, DialRotate, DialUp +from streamdeck.models.events.keys import KeyDown, KeyUp +from streamdeck.models.events.property_inspector import ( + DidReceivePropertyInspectorMessage, + PropertyInspectorDidAppear, + PropertyInspectorDidDisappear, +) +from streamdeck.models.events.settings import DidReceiveGlobalSettings, DidReceiveSettings +from streamdeck.models.events.system import SystemDidWakeUp +from streamdeck.models.events.title_parameters import TitleParametersDidChange +from streamdeck.models.events.touch_tap import TouchTap +from streamdeck.models.events.visibility import WillAppear, WillDisappear + + +if TYPE_CHECKING: + from typing import Final + + +DEFAULT_EVENT_MODELS: Final[list[type[EventBase]]] = [ + ApplicationDidLaunch, + ApplicationDidTerminate, + DeviceDidConnect, + DeviceDidDisconnect, + DialDown, + DialRotate, + DialUp, + DidReceiveDeepLink, + KeyUp, + KeyDown, + DidReceivePropertyInspectorMessage, + PropertyInspectorDidAppear, + PropertyInspectorDidDisappear, + DidReceiveGlobalSettings, + DidReceiveSettings, + SystemDidWakeUp, + TitleParametersDidChange, + TouchTap, + WillAppear, + WillDisappear, +] + + +def _get_default_event_names() -> set[str]: + default_event_names: set[str] = set() + + for event_model in DEFAULT_EVENT_MODELS: + default_event_names.update(event_model.get_model_event_name()) + + return default_event_names + + +DEFAULT_EVENT_NAMES: Final[set[str]] = _get_default_event_names() + + + +__all__ = [ + "DEFAULT_EVENT_MODELS", + "DEFAULT_EVENT_NAMES", + "ApplicationDidLaunch", + "ApplicationDidTerminate", + "DeviceDidConnect", + "DeviceDidDisconnect", + "DialDown", + "DialRotate", + "DialUp", + "DidReceiveDeepLink", + "DidReceiveGlobalSettings", + "DidReceivePropertyInspectorMessage", + "DidReceiveSettings", + "EventBase", + "KeyDown", + "KeyUp", + "PropertyInspectorDidAppear", + "PropertyInspectorDidDisappear", + "SystemDidWakeUp", + "TitleParametersDidChange", + "TouchTap", + "WillAppear", + "WillDisappear", +] diff --git a/streamdeck/models/events/adapter.py b/streamdeck/models/events/adapter.py new file mode 100644 index 0000000..347fa15 --- /dev/null +++ b/streamdeck/models/events/adapter.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Annotated, Union + +from pydantic import Field, TypeAdapter + +from streamdeck.models.events import DEFAULT_EVENT_MODELS + + +if TYPE_CHECKING: + from streamdeck.models.events.base import EventBase + + + +class EventAdapter: + """TypeAdapter-encompassing class for handling and extending available event models.""" + def __init__(self) -> None: + self._models: list[type[EventBase]] = [] + self._type_adapter: TypeAdapter[EventBase] | None = None + + self._event_names: set[str] = set() + """A set of all event names that have been registered with the adapter. + This set starts out containing the default event models defined by the library. + """ + + for model in DEFAULT_EVENT_MODELS: + self.add_model(model) + + def add_model(self, model: type[EventBase]) -> None: + """Add a model to the adapter, and add the event name of the model to the set of registered event names.""" + self._models.append(model) + self._event_names.update(model.get_model_event_name()) + + def event_name_exists(self, event_name: str) -> bool: + """Check if an event name has been registered with the adapter.""" + return event_name in self._event_names + + @property + def type_adapter(self) -> TypeAdapter[EventBase]: + """Get the TypeAdapter instance for the event models.""" + if self._type_adapter is None: + self._type_adapter = TypeAdapter( + Annotated[ + Union[tuple(self._models)], # noqa: UP007 + Field(discriminator="event") + ] + ) + + return self._type_adapter + + def validate_json(self, data: str | bytes) -> EventBase: + """Validate a JSON string or bytes object as an event model.""" + return self.type_adapter.validate_json(data) + diff --git a/streamdeck/models/events/application.py b/streamdeck/models/events/application.py new file mode 100644 index 0000000..6e4a788 --- /dev/null +++ b/streamdeck/models/events/application.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from typing import Literal + +from streamdeck.models.events.base import ConfiguredBaseModel, EventBase + + +class ApplicationPayload(ConfiguredBaseModel): + """Payload containing the name of the application that triggered the event.""" + application: str + """Name of the application that triggered the event.""" + + +class ApplicationDidLaunch(EventBase): + """Occurs when a monitored application is launched.""" + event: Literal["applicationDidLaunch"] # type: ignore[override] + payload: ApplicationPayload + """Payload containing the name of the application that triggered the event.""" + + +class ApplicationDidTerminate(EventBase): + """Occurs when a monitored application terminates.""" + event: Literal["applicationDidTerminate"] # type: ignore[override] + payload: ApplicationPayload + """Payload containing the name of the application that triggered the event.""" diff --git a/streamdeck/models/events/base.py b/streamdeck/models/events/base.py new file mode 100644 index 0000000..b003a76 --- /dev/null +++ b/streamdeck/models/events/base.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +from abc import ABC +from typing import Any, Literal, get_args, get_type_hints + +from pydantic import BaseModel, ConfigDict +from typing_extensions import ( # noqa: UP035 + LiteralString, + TypeIs, + override, +) + + +class ConfiguredBaseModel(BaseModel, ABC): + """Base class for models that share the same configuration.""" + # Configure to use the docstrings of the fields as the field descriptions, + # and to serialize the fields by their aliases. + model_config = ConfigDict(use_attribute_docstrings=True, serialize_by_alias=True) + + @override + def model_dump_json(self, **kwargs: Any) -> str: + """Dump the model to JSON, excluding default values by default. + + Fields with default values in this context are those that are not required by the model, + and are given a default value of None. Thus, for the serialized JSON to have parity with the + StreamDeck-provided JSON event messages, we need to exclude fields not found in the event message. + + Unfortunately, the `exclude_defaults` option is not available in the ConfigDict configuration, + nor in the Field parameters. To work around this, we wrap the `model_dump_json` method + to set `exclude_defaults` to `True` by default. + """ + kwargs.setdefault("exclude_defaults", True) + return super().model_dump_json(**kwargs) + + +def is_literal_str_type(value: object | None) -> TypeIs[LiteralString]: + """Check if a type is a Literal type with string values.""" + if value is None: + return False + + event_field_base_type = getattr(value, "__origin__", None) + + if event_field_base_type is not Literal: + return False + + return all(isinstance(literal_value, str) for literal_value in get_args(value)) + + +## EventBase implementation model of the Stream Deck Plugin SDK events. + +class EventBase(ConfiguredBaseModel, ABC): + """Base class for event models that represent Stream Deck Plugin SDK events.""" + # Configure to use the docstrings of the fields as the field descriptions. + model_config = ConfigDict(use_attribute_docstrings=True, serialize_by_alias=True) + + event: str + """Name of the event used to identify what occurred. + + Subclass models must define this field as a Literal type with the event name string that the model represents. + """ + + def __init_subclass__(cls, **kwargs: Any) -> None: + """Validate that the event field is a Literal[str] type.""" + super().__init_subclass__(**kwargs) + + model_event_type = cls.get_event_type_annotations() + + if not is_literal_str_type(model_event_type): + msg = f"The event field annotation must be a Literal[str] type. Given type: {model_event_type}" + raise TypeError(msg) + + @classmethod + def get_event_type_annotations(cls) -> type[object]: + """Get the type annotations of the subclass model's event field.""" + return get_type_hints(cls)["event"] + + @classmethod + def get_model_event_name(cls) -> tuple[str, ...]: + """Get the value of the subclass model's event field Literal annotation.""" + model_event_type = cls.get_event_type_annotations() + + # Ensure that the event field annotation is a Literal type. + if not is_literal_str_type(model_event_type): + msg = "The `event` field annotation of an Event model must be a Literal[str] type." + raise TypeError(msg) + + return get_args(model_event_type) diff --git a/streamdeck/models/events/common.py b/streamdeck/models/events/common.py new file mode 100644 index 0000000..8ecc281 --- /dev/null +++ b/streamdeck/models/events/common.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +from abc import ABC +from typing import Annotated, Any, Generic, Literal, NamedTuple, Optional, Union + +from pydantic import Field +from typing_extensions import TypedDict, TypeVar + +from streamdeck.models.events.base import ConfiguredBaseModel + + +## Mixin classes for common event model fields. + +class ContextualEventMixin: + """Mixin class for event models that have action and context fields.""" + 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.""" + + +class DeviceSpecificEventMixin: + """Mixin class for event models that have a device field.""" + device: str + """Unique identifier of the Stream Deck device that this event is associated with.""" + + +## Payload models and metadata used by multiple event models. + + +PluginDefinedData = dict[str, Any] +"""Data of arbitrary structure that is defined in and relevant to the plugin.""" + + +EncoderControllerType = Literal["Encoder"] +"""The 'Encoder' controller type refers to a dial or touchscreen on a 'Stream Deck +' device.""" +KeypadControllerType = Literal["Keypad"] +"""The 'Keypad' controller type refers to a standard action on a Stream Deck device, e.g. buttons or a pedal.""" +ControllerType = Literal[EncoderControllerType, KeypadControllerType] +"""Defines the controller type the action is applicable to.""" + +CT = TypeVar("CT", bound=ControllerType, default=ControllerType) + + +class BasePayload(ConfiguredBaseModel, Generic[CT], ABC): + """Base class for all complex payload models.""" + controller: CT + """Defines the controller type the action is applicable to. + + 'Keypad' refers to a standard action on a Stream Deck device, e.g. buttons or a pedal. + 'Encoder' refers to a dial / touchscreen on a 'Stream Deck +' device. + """ + settings: PluginDefinedData + """Settings associated with the action instance.""" + + +class CoordinatesDict(TypedDict): + """Coordinates that identify the location of an action.""" + + column: int + """Column the action instance is located in, indexed from 0.""" + row: int + """Row the action instance is located on, indexed from 0. + + When the device is DeviceType.StreamDeckPlus the row can be 0 for keys (Keypad), + and will always be 0 for dials (Encoder). + """ + + +class Coordinates(NamedTuple): + """Coordinates that identify the location of an action.""" + + column: int + """Column the action instance is located in, indexed from 0.""" + row: int + """Row the action instance is located on, indexed from 0. + + When the device is DeviceType.StreamDeckPlus the row can be 0 for keys (Keypad), + and will always be 0 for dials (Encoder). + """ + + +class CoordinatesPayloadMixin: + """Mixin class for event models that have a coordinates field.""" + coordinates_obj: Annotated[ + CoordinatesDict, Field(alias="coordinates", repr=False) + ] + """Coordinates dictionary that identify the location of the action instance on the device.""" + + @property + def coordinates(self) -> Coordinates: + """Coordinates that identify the location of the action instance on the device.""" + return Coordinates(**self.coordinates_obj) + + +class StatefulActionPayloadMixin: + """Mixin class for payload models that have an optional state field.""" + state: Optional[int] = None # noqa: UP007 + """Current state of the action. + + Only applicable to actions that have multiple states defined within the manifest.json file. + """ + + +class SingleActionPayloadMixin: + """Mixin class for event models that have a single action payload.""" + + is_in_multi_action: Annotated[Literal[False], Field(alias="isInMultiAction")] + """Indicates that this event is not part of a multi-action.""" + + +class MultiActionPayloadMixin: + """Mixin class for event models that have a multi-action payload.""" + + is_in_multi_action: Annotated[Literal[True], Field(alias="isInMultiAction")] + """Indicates that this event is part of a multi-action.""" + + +CardinalityDiscriminated = Annotated[ + Union[ # noqa: UP007 + TypeVar("S", bound=SingleActionPayloadMixin), + TypeVar("M", bound=MultiActionPayloadMixin), + ], + Field(discriminator="is_in_multi_action"), +] +"""Generic type for a payload that either subclasses SingleActionPayloadMixin or MultiActionPayloadMixin—meaning it can be either a single action or a multi-action.""" diff --git a/streamdeck/models/events/deep_link.py b/streamdeck/models/events/deep_link.py new file mode 100644 index 0000000..6331ce5 --- /dev/null +++ b/streamdeck/models/events/deep_link.py @@ -0,0 +1,21 @@ +from typing import Literal + +from streamdeck.models.events.base import ConfiguredBaseModel, EventBase + + +class DeepLinkPayload(ConfiguredBaseModel): + """Payload containing information about the URL that triggered the event.""" + url: str + """The deep-link URL, with the prefix omitted.""" + + +class DidReceiveDeepLink(EventBase): + """Occurs when Stream Deck receives a deep-link message intended for the plugin. + + The message is re-routed to the plugin, and provided as part of the payload. + One-way deep-link message can be routed to the plugin using the URL format: + streamdeck://plugins/message//{MESSAGE}. + """ + event: Literal["didReceiveDeepLink"] # type: ignore[override] + payload: DeepLinkPayload + """Payload containing information about the URL that triggered the event.""" diff --git a/streamdeck/models/events/devices.py b/streamdeck/models/events/devices.py new file mode 100644 index 0000000..50c1ac9 --- /dev/null +++ b/streamdeck/models/events/devices.py @@ -0,0 +1,90 @@ +from typing import Annotated, Final, Literal, NamedTuple + +from pydantic import Field +from typing_extensions import TypedDict + +from streamdeck.models.events.base import ConfiguredBaseModel, EventBase +from streamdeck.models.events.common import DeviceSpecificEventMixin + + +DeviceTypeId = Literal[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10] + + +DeviceType = Literal[ + "Stream Deck", + "Stream Deck Mini", + "Stream Deck XL", + "Stream Deck Mobile", + "Corsair GKeys", + "Stream Deck Pedal", + "Corsair Voyager", + "Stream Deck +", + "SCUF Controller", + "Stream Deck Neo", +] + + +DEVICE_TYPE_BY_ID: Final[dict[DeviceTypeId, DeviceType]] = { + 0: "Stream Deck", + 1: "Stream Deck Mini", + 2: "Stream Deck XL", + 3: "Stream Deck Mobile", + 4: "Corsair GKeys", + 5: "Stream Deck Pedal", + 6: "Corsair Voyager", + 7: "Stream Deck +", + 8: "SCUF Controller", + 9: "Stream Deck Neo", +} + + +class DeviceSizeDict(TypedDict): + """Number of action slots, excluding dials / touchscreens, available to the device.""" + columns: int + rows: int + + +class DeviceSize(NamedTuple): + """Number of action slots, excluding dials / touchscreens, available to the device.""" + columns: int + rows: int + + +class DeviceInfo(ConfiguredBaseModel): + """Information about the newly connected device.""" + name: str + """Name of the device, as specified by the user in the Stream Deck application.""" + type_id: Annotated[DeviceTypeId, Field(alias="type", repr=False)] + """The type (id) of the device that was connected.""" + size_obj: Annotated[DeviceSizeDict, Field(alias="size", repr=False)] + """Number of action slots, excluding dials / touchscreens, available to the device.""" + + @property + def type(self) -> DeviceType: + """The type (product name) of the device that was connected.""" + if self.type_id not in DEVICE_TYPE_BY_ID: + msg = f"Unknown device type id: {self.type_id}" + raise ValueError(msg) + + return DEVICE_TYPE_BY_ID[self.type_id] + + @property + def size(self) -> DeviceSize: + """Number of action slots, excluding dials / touchscreens, available to the device.""" + return DeviceSize(**self.size_obj) + + def __repr__(self) -> str: + """Return a string representation of the device info.""" + return f"DeviceInfo(name={self.name}, type={self.type}, size={self.size})" + + +class DeviceDidConnect(EventBase, DeviceSpecificEventMixin): + """Occurs when a Stream Deck device is connected.""" + event: Literal["deviceDidConnect"] # type: ignore[override] + device_info: Annotated[DeviceInfo, Field(alias="deviceInfo")] + """Information about the newly connected device.""" + + +class DeviceDidDisconnect(EventBase, DeviceSpecificEventMixin): + """Occurs when a Stream Deck device is disconnected.""" + event: Literal["deviceDidDisconnect"] # type: ignore[override] diff --git a/streamdeck/models/events/dials.py b/streamdeck/models/events/dials.py new file mode 100644 index 0000000..ba63ada --- /dev/null +++ b/streamdeck/models/events/dials.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from typing import Literal + +from streamdeck.models.events.base import EventBase +from streamdeck.models.events.common import ( + BasePayload, + ContextualEventMixin, + CoordinatesPayloadMixin, + DeviceSpecificEventMixin, + EncoderControllerType, +) + + +## Payload models used in the below DialDown, DialRotate, and DialUp events + +class EncoderPayload(BasePayload[EncoderControllerType], CoordinatesPayloadMixin): + """Contextualized information for a DialDown or DialUp event.""" + + +class DialRotatePayload(EncoderPayload): + """Contextualized information for a DialRotate event.""" + pressed: bool + """Determines whether the dial was pressed whilst the rotation occurred.""" + ticks: int + """Number of ticks the dial was rotated; this can be a positive (clockwise) or negative (counter-clockwise) number.""" + + +## Event models for DialDown, DialRotate, and DialUp events + +class DialDown(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): + """Occurs when the user presses a dial (Stream Deck +).""" + event: Literal["dialDown"] # type: ignore[override] + payload: EncoderPayload + """Contextualized information for this event.""" + + +class DialRotate(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): + """Occurs when the user rotates a dial (Stream Deck +).""" + event: Literal["dialRotate"] # type: ignore[override] + payload: DialRotatePayload + """Contextualized information for this event.""" + + +class DialUp(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): + """Occurs when the user releases a pressed dial (Stream Deck +).""" + event: Literal["dialUp"] # type: ignore[override] + payload: EncoderPayload + """Contextualized information for this event.""" diff --git a/streamdeck/models/events/keys.py b/streamdeck/models/events/keys.py new file mode 100644 index 0000000..6b74921 --- /dev/null +++ b/streamdeck/models/events/keys.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from typing import Annotated, Literal + +from pydantic import Field + +from streamdeck.models.events.base import EventBase +from streamdeck.models.events.common import ( + BasePayload, + CardinalityDiscriminated, + ContextualEventMixin, + CoordinatesPayloadMixin, + DeviceSpecificEventMixin, + KeypadControllerType, + MultiActionPayloadMixin, + SingleActionPayloadMixin, + StatefulActionPayloadMixin, +) + + +## Payload models used in the below KeyDown and KeyUp events + +# It seems like for keyDown and keyUp events, the "controller" field is probably always missing, despite being defined in the official documentation. +OptionalKeyControllerTypeField = Annotated[KeypadControllerType, Field(default=None)] + + +class SingleActionKeyGesturePayload( + BasePayload[OptionalKeyControllerTypeField], + SingleActionPayloadMixin, + StatefulActionPayloadMixin, + CoordinatesPayloadMixin, +): + """Contextualized information for a KeyDown or KeyUp event that is not part of a multi-action.""" + + +class MultiActionKeyGesturePayload( + BasePayload[OptionalKeyControllerTypeField], + MultiActionPayloadMixin, + StatefulActionPayloadMixin, +): + """Contextualized information for a KeyDown or KeyUp event that is part of a multi-action.""" + user_desired_state: Annotated[int, Field(alias="userDesiredState")] + """Desired state as specified by the user. + + Only applicable to actions that have multiple states defined within the manifest.json file, and when this action instance is part of a multi-action. + """ + + +## Event models for KeyDown and KeyUp events + +class KeyDown(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): + """Occurs when the user presses a action down.""" + event: Literal["keyDown"] # type: ignore[override] + payload: CardinalityDiscriminated[SingleActionKeyGesturePayload, MultiActionKeyGesturePayload] + """Contextualized information for this event.""" + + +class KeyUp(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): + """Occurs when the user releases a pressed action.""" + event: Literal["keyUp"] # type: ignore[override] + payload: CardinalityDiscriminated[SingleActionKeyGesturePayload, MultiActionKeyGesturePayload] + """Contextualized information for this event.""" diff --git a/streamdeck/models/events/property_inspector.py b/streamdeck/models/events/property_inspector.py new file mode 100644 index 0000000..8097fd4 --- /dev/null +++ b/streamdeck/models/events/property_inspector.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +from typing import Literal + +from streamdeck.models.events.base import EventBase +from streamdeck.models.events.common import ( + ContextualEventMixin, + DeviceSpecificEventMixin, + PluginDefinedData, +) + + +class DidReceivePropertyInspectorMessage(EventBase, ContextualEventMixin): + """Occurs when a message was received from the UI.""" + event: Literal["sendToPlugin"] # type: ignore[override] + payload: PluginDefinedData + """The data payload received from the UI.""" + + +class PropertyInspectorDidAppear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): + """Occurs when the property inspector associated with the action becomes visible. + + I.e. the user selected an action in the Stream Deck application. + """ + event: Literal["propertyInspectorDidAppear"] # type: ignore[override] + + +class PropertyInspectorDidDisappear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): + """Occurs when the property inspector associated with the action becomes invisible. + + I.e. the user unselected the action in the Stream Deck application. + """ + event: Literal["propertyInspectorDidDisappear"] # type: ignore[override] diff --git a/streamdeck/models/events/settings.py b/streamdeck/models/events/settings.py new file mode 100644 index 0000000..c9dab53 --- /dev/null +++ b/streamdeck/models/events/settings.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from typing import Literal + +from streamdeck.models.events.base import ConfiguredBaseModel, EventBase +from streamdeck.models.events.common import ( + BasePayload, + CardinalityDiscriminated, + ContextualEventMixin, + CoordinatesPayloadMixin, + DeviceSpecificEventMixin, + KeypadControllerType, + MultiActionPayloadMixin, + PluginDefinedData, + SingleActionPayloadMixin, + StatefulActionPayloadMixin, +) + + +## Models for didReceiveSettings event and its specific payloads. + + +class SingleActionSettingsPayload( + BasePayload, + SingleActionPayloadMixin, + StatefulActionPayloadMixin, + CoordinatesPayloadMixin, +): + """Contextualized information for a didReceiveSettings events that are not part of a multi-action.""" + + +class MultiActionSettingsPayload( + BasePayload[KeypadControllerType], + MultiActionPayloadMixin, + StatefulActionPayloadMixin, +): + """Contextualized information for a didReceiveSettings events that are part of a multi-action. + + NOTE: Action instances that are part of a multi-action are only applicable to the 'Keypad' controller type. + """ + + +class DidReceiveSettings(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): + """Occurs when the settings associated with an action instance are requested, or when the the settings were updated by the property inspector.""" + event: Literal["didReceiveSettings"] # type: ignore[override] + payload: CardinalityDiscriminated[SingleActionSettingsPayload, MultiActionSettingsPayload] + """Contextualized information for this event.""" + + +## Models for didReceiveGlobalSettings event and its specific payload. + +class GlobalSettingsPayload(ConfiguredBaseModel): + """Additional information about the didReceiveGlobalSettings event that occurred.""" + settings: PluginDefinedData + """The global settings received from the Stream Deck.""" + + +class DidReceiveGlobalSettings(EventBase): + """Occurs when the plugin receives the global settings from the Stream Deck.""" + event: Literal["didReceiveGlobalSettings"] # type: ignore[override] + payload: GlobalSettingsPayload + """Additional information about the event that occurred.""" diff --git a/streamdeck/models/events/system.py b/streamdeck/models/events/system.py new file mode 100644 index 0000000..2c6be51 --- /dev/null +++ b/streamdeck/models/events/system.py @@ -0,0 +1,10 @@ +from __future__ import annotations + +from typing import Literal + +from streamdeck.models.events.base import EventBase + + +class SystemDidWakeUp(EventBase): + """Occurs when the computer wakes up.""" + event: Literal["systemDidWakeUp"] # type: ignore[override] diff --git a/streamdeck/models/events/title_parameters.py b/streamdeck/models/events/title_parameters.py new file mode 100644 index 0000000..386d089 --- /dev/null +++ b/streamdeck/models/events/title_parameters.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +from typing import Annotated, Literal + +from pydantic import Field + +from streamdeck.models.events.base import ConfiguredBaseModel, EventBase +from streamdeck.models.events.common import ( + BasePayload, + CoordinatesPayloadMixin, + DeviceSpecificEventMixin, + StatefulActionPayloadMixin, +) + + +FontStyle = Literal["", "Bold Italic", "Bold", "Italic", "Regular"] +TitleAlignment = Literal["top", "middle", "bottom"] + + +class TitleParameters(ConfiguredBaseModel): + """Defines aesthetic properties that determine how the title should be rendered.""" + font_family: Annotated[str, Field(alias="fontFamily")] + """Font-family the title will be rendered with.""" + font_size: Annotated[int, Field(alias="fontSize")] + """Font-size the title will be rendered in.""" + font_style: Annotated[FontStyle, Field(alias="fontStyle")] + """Typography of the title.""" + font_underline: Annotated[bool, Field(alias="fontUnderline")] + """Whether the font should be underlined.""" + show_title: Annotated[bool, Field(alias="showTitle")] + """Whether the user has opted to show, or hide the title for this action instance.""" + title_alignment: Annotated[str, Field(alias="titleAlignment")] + """Alignment of the title.""" + title_color: Annotated[str, Field(alias="titleColor")] + """Color of the title, represented as a hexadecimal value.""" + + +class TitleParametersDidChangePayload( + BasePayload, + CoordinatesPayloadMixin, + StatefulActionPayloadMixin, +): + """Contextualized information for this event.""" + title: str + """Title of the action, as specified by the user or dynamically by the plugin.""" + title_parameters: Annotated[TitleParameters, Field(alias="titleParameters")] + """Defines aesthetic properties that determine how the title should be rendered.""" + + +class TitleParametersDidChange(EventBase, DeviceSpecificEventMixin): + """Occurs when the user updates an action's title settings in the Stream Deck application.""" + 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 diff --git a/streamdeck/models/events/touch_tap.py b/streamdeck/models/events/touch_tap.py new file mode 100644 index 0000000..4226a1c --- /dev/null +++ b/streamdeck/models/events/touch_tap.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +from typing import Annotated, Literal + +from pydantic import Field + +from streamdeck.models.events.base import EventBase +from streamdeck.models.events.common import ( + BasePayload, + ContextualEventMixin, + CoordinatesPayloadMixin, + DeviceSpecificEventMixin, + EncoderControllerType, +) + + +class TouchTapPayload(BasePayload[EncoderControllerType], CoordinatesPayloadMixin): + """Contextualized information for a TouchTap event.""" + hold: bool + """Determines whether the tap was considered 'held'.""" + tap_position: Annotated[tuple[int, int], Field(alias="tapPos")] + """Coordinates of where the touchscreen tap occurred, relative to the canvas of the action.""" + + +class TouchTap(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): + """Occurs when the user taps the touchscreen (Stream Deck +).""" + event: Literal["touchTap"] # type: ignore[override] + payload: TouchTapPayload + """Contextualized information for this event.""" diff --git a/streamdeck/models/events/visibility.py b/streamdeck/models/events/visibility.py new file mode 100644 index 0000000..cf35c34 --- /dev/null +++ b/streamdeck/models/events/visibility.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from typing import Literal + +from streamdeck.models.events.base import EventBase +from streamdeck.models.events.common import ( + BasePayload, + CardinalityDiscriminated, + ContextualEventMixin, + CoordinatesPayloadMixin, + DeviceSpecificEventMixin, + KeypadControllerType, + MultiActionPayloadMixin, + SingleActionPayloadMixin, + StatefulActionPayloadMixin, +) + + +class SingleActionVisibilityPayload( + BasePayload, + SingleActionPayloadMixin, + StatefulActionPayloadMixin, + CoordinatesPayloadMixin, +): + """Contextualized information for willAppear and willDisappear events that is not part of a multi-action.""" + + +class MultiActionVisibilityPayload( + BasePayload[KeypadControllerType], + MultiActionPayloadMixin, + StatefulActionPayloadMixin, +): + """Contextualized information for willAppear and willDisappear events that is part of a multi-action.""" + + +## Event models for WillAppear and WillDisappear events + +class WillAppear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): + """Occurs when an action appears on the Stream Deck due to the user navigating to another page, profile, folder, etc. + + This also occurs during startup if the action is on the "front page". + An action refers to all types of actions, e.g. keys, dials, touchscreens, pedals, etc. + """ + event: Literal["willAppear"] # type: ignore[override] + payload: CardinalityDiscriminated[SingleActionVisibilityPayload, MultiActionVisibilityPayload] + + +class WillDisappear(EventBase, ContextualEventMixin, DeviceSpecificEventMixin): + """Occurs when an action disappears from the Stream Deck due to the user navigating to another page, profile, folder, etc. + + An action refers to all types of actions, e.g. keys, dials, touchscreens, pedals, etc. + """ + event: Literal["willDisappear"] # type: ignore[override] + payload: CardinalityDiscriminated[SingleActionVisibilityPayload, MultiActionVisibilityPayload] diff --git a/streamdeck/websocket.py b/streamdeck/websocket.py index 8fa0504..113b9be 100644 --- a/streamdeck/websocket.py +++ b/streamdeck/websocket.py @@ -13,7 +13,7 @@ from websockets.sync.client import ClientConnection, connect from streamdeck.event_listener import EventListener, StopStreaming -from streamdeck.models import events +from streamdeck.models.events import DEFAULT_EVENT_MODELS, EventBase if TYPE_CHECKING: @@ -32,7 +32,7 @@ class WebSocketClient(EventListener): """A client for connecting to the Stream Deck device's WebSocket server and sending/receiving events.""" _client: ClientConnection | None - event_models: ClassVar[list[type[events.EventBase]]] = events.DEFAULT_EVENT_MODELS + event_models: ClassVar[list[type[EventBase]]] = DEFAULT_EVENT_MODELS def __init__(self, port: int): """Initialize a WebSocketClient instance. diff --git a/tests/actions/test_action_event_handler_filtering.py b/tests/actions/test_action_event_handler_filtering.py index 4c5e315..d55915d 100644 --- a/tests/actions/test_action_event_handler_filtering.py +++ b/tests/actions/test_action_event_handler_filtering.py @@ -4,9 +4,8 @@ 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 +from streamdeck.models.events.common import ContextualEventMixin from tests.test_utils.fake_event_factories import ( ApplicationDidLaunchEventFactory, @@ -18,6 +17,9 @@ if TYPE_CHECKING: from unittest.mock import Mock + from polyfactory.factories.pydantic_factory import ModelFactory + from streamdeck.models import events + @pytest.fixture @@ -34,7 +36,7 @@ def dummy_handler(event: events.EventBase) -> None: ApplicationDidLaunchEventFactory ]) def fake_event_data(request: pytest.FixtureRequest) -> events.EventBase: - event_factory = cast(ModelFactory[events.EventBase], request.param) + event_factory = cast("ModelFactory[events.EventBase]", request.param) return event_factory.build() @@ -67,7 +69,7 @@ def test_action_gets_triggered_by_event( Actions should only be triggered by events that have the same unique identifier properties as the action. """ # Extract the action UUID from the fake event data, or use a default value - action_uuid: str = fake_event_data.action if isinstance(fake_event_data, events.ContextualEventMixin) else "my-fake-action-uuid" + action_uuid: str = fake_event_data.action if isinstance(fake_event_data, ContextualEventMixin) else "my-fake-action-uuid" action = Action(uuid=action_uuid) @@ -90,7 +92,7 @@ def test_global_action_registry_get_action_handlers_filtering( fake_event_data: events.EventBase, ) -> None: # Extract the action UUID from the fake event data, or use a default value - action_uuid: str | None = fake_event_data.action if isinstance(fake_event_data, events.ContextualEventMixin) else None + action_uuid: str | None = fake_event_data.action if isinstance(fake_event_data, ContextualEventMixin) else None registry = ActionRegistry() # Create an Action instance, without an action UUID as global actions aren't associated with a specific action @@ -117,7 +119,7 @@ def test_action_registry_get_action_handlers_filtering( fake_event_data: events.EventBase, ) -> None: # Extract the action UUID from the fake event data, or use a default value - action_uuid: str | None = fake_event_data.action if isinstance(fake_event_data, events.ContextualEventMixin) else None + action_uuid: str | None = fake_event_data.action if isinstance(fake_event_data, ContextualEventMixin) else None registry = ActionRegistry() # Create an Action instance, using either the fake event's action UUID or a default value diff --git a/tests/actions/test_action_registry.py b/tests/actions/test_action_registry.py index 0909ef2..606d820 100644 --- a/tests/actions/test_action_registry.py +++ b/tests/actions/test_action_registry.py @@ -1,8 +1,9 @@ from __future__ import annotations +from typing import TYPE_CHECKING + import pytest from streamdeck.actions import Action, ActionRegistry -from streamdeck.models import events from tests.test_utils.fake_event_factories import ( DialDownEventFactory, @@ -11,6 +12,10 @@ ) +if TYPE_CHECKING: + from streamdeck.models import events + + def test_register_action() -> None: """Test that an action can be registered.""" registry = ActionRegistry() diff --git a/tests/actions/test_actions.py b/tests/actions/test_actions.py index 8decc51..85660a4 100644 --- a/tests/actions/test_actions.py +++ b/tests/actions/test_actions.py @@ -17,7 +17,7 @@ def action(request: pytest.FixtureRequest) -> ActionBase: We have to initialize the classes here to ensure fresh instances are used to avoid sharing data between tests. """ - action_class, init_args = cast(tuple[type[ActionBase], tuple[Any]], request.param) + action_class, init_args = cast("tuple[type[ActionBase], tuple[Any]]", request.param) return action_class(*init_args) diff --git a/tests/models/test_event_adapter.py b/tests/models/test_event_adapter.py index efcecc2..a644e07 100644 --- a/tests/models/test_event_adapter.py +++ b/tests/models/test_event_adapter.py @@ -3,12 +3,9 @@ import pytest from pydantic import ValidationError -from streamdeck.models.events import ( - DEFAULT_EVENT_MODELS, - DEFAULT_EVENT_NAMES, +from streamdeck.models.events import DEFAULT_EVENT_MODELS, DEFAULT_EVENT_NAMES, EventBase, KeyDown +from streamdeck.models.events.adapter import ( EventAdapter, - EventBase, - KeyDown, ) from tests.test_utils.fake_event_factories import KeyDownEventFactory