diff --git a/README.md b/README.md index fbf853a..6a34ac9 100644 --- a/README.md +++ b/README.md @@ -323,7 +323,7 @@ The SDK allows you to create custom event listeners and events by extending the To create a custom event listener: -1. Create new event model that inherits from `EventBase["eventName"]`. +1. Create new event model that inherits from `EventBase[Literal["eventName"]]`. 2. Create a new class that inherits from `EventListener`. a. Implement the required `listen` and `stop` methods. The `listen` method should yield results as a json string that matches the new event model. b. List the new event classes in the `event_models` class variable of the new `EventListener` class. @@ -338,7 +338,7 @@ from streamdeck.event_listener import EventListener from streamdeck.models.events import EventBase -class MyCustomEvent(EventBase["somethingHappened"]): +class MyCustomEvent(EventBase[Literal["somethingHappened"]]): # The 'event' field's type annotation is internally set as Literal["somethingHappened"] # Define additional data attributes here result: str diff --git a/streamdeck/models/configs.py b/streamdeck/models/configs.py index da835e9..d3d7db0 100644 --- a/streamdeck/models/configs.py +++ b/streamdeck/models/configs.py @@ -112,5 +112,10 @@ def event_models(self) -> list[type[EventBase]]: return [ obj for obj in self._event_listener_objects - if isinstance(obj, type) and obj is not EventBase and issubclass(obj, EventBase) + if ( + isinstance(obj, type) + and obj is not EventBase # Exclude the base class itself. + and issubclass(obj, EventBase) + and getattr(obj, "__pydantic_generic_metadata__", {}).get("origin") is not EventBase + ) ] diff --git a/streamdeck/models/events/adapter.py b/streamdeck/models/events/adapter.py index e13e888..ee406cb 100644 --- a/streamdeck/models/events/adapter.py +++ b/streamdeck/models/events/adapter.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Annotated, Union +from typing import TYPE_CHECKING, Annotated, Union, cast from pydantic import Field, TypeAdapter @@ -8,6 +8,8 @@ if TYPE_CHECKING: + from typing_extensions import TypeForm + from streamdeck.models.events.base import EventBase from streamdeck.types import EventNameStr @@ -41,11 +43,15 @@ def event_name_exists(self, event_name: str) -> bool: def type_adapter(self) -> TypeAdapter[EventBase]: """Get the TypeAdapter instance for the event models.""" if self._type_adapter is None: + # Mypy cannot infer the type form of this union because its models are collected at runtime. self._type_adapter = TypeAdapter( - Annotated[ - Union[tuple(self._models)], # noqa: UP007 - Field(discriminator="event") - ] + cast( + "TypeForm[EventBase]", + Annotated[ + Union[tuple(self._models)], # noqa: UP007 + Field(discriminator="event") + ], + ) ) return self._type_adapter diff --git a/streamdeck/models/events/application.py b/streamdeck/models/events/application.py index c93cc79..2b24e46 100644 --- a/streamdeck/models/events/application.py +++ b/streamdeck/models/events/application.py @@ -1,21 +1,26 @@ 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["applicationDidLaunch"]): +class ApplicationDidLaunch(EventBase[Literal["applicationDidLaunch"]]): """Occurs when a monitored application is launched.""" + payload: ApplicationPayload """Payload containing the name of the application that triggered the event.""" -class ApplicationDidTerminate(EventBase["applicationDidTerminate"]): +class ApplicationDidTerminate(EventBase[Literal["applicationDidTerminate"]]): """Occurs when a monitored application terminates.""" + 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 index 0de0538..a99c840 100644 --- a/streamdeck/models/events/base.py +++ b/streamdeck/models/events/base.py @@ -1,20 +1,30 @@ +"""Shared Pydantic base classes for incoming event models.""" + from __future__ import annotations from abc import ABC -from typing import TYPE_CHECKING, Literal -from weakref import WeakValueDictionary +from typing import TYPE_CHECKING, Annotated, Any, Generic, Literal, get_args, get_origin -from pydantic import BaseModel, ConfigDict, create_model -from typing_extensions import LiteralString, TypedDict, override # noqa: UP035 +from pydantic import BaseModel, ConfigDict, GetPydanticSchema +from typing_extensions import LiteralString, TypeVar, override if TYPE_CHECKING: - from typing import Any, ClassVar + from typing import ClassVar, TypeAlias + + +StringLiteralType: TypeAlias = Annotated[ + LiteralString, + GetPydanticSchema(lambda _source_type, handler: handler(str)), +] +"""A string-valued Literal type, with a Pydantic schema fallback to str.""" +EventNameT = TypeVar("EventNameT", bound=StringLiteralType, default=Any) 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) @@ -35,180 +45,74 @@ def model_dump_json(self, **kwargs: Any) -> str: return super().model_dump_json(**kwargs) -class EventMetadataDict(TypedDict): - """Metadata for specialized EventBase submodels. +class EventBase(ConfiguredBaseModel, Generic[EventNameT]): + """Base class for all event models. - Similar to the __pydantic_generic_metadata__ attribute, but for use in the EventBase class——which isn't actually generic. + EventBase itself and its generic specializations should not be instantiated. + Define concrete event models by specializing with a string-valued Literal, + for example `class KeyDown(EventBase[Literal["keyDown"]]): ...`. """ - origin: type[EventBase] - """Origin class of the specialized EventBase submodel.""" - args: tuple[str, ...] - """Event names for the specialized EventBase submodel.""" - - -if TYPE_CHECKING: - # Because we can't override a BaseModel's metaclass __getitem__ method without angering Pydantic during runtime, - # we define this stub here to satisfy static type checkers that introspect the metaclass method annotations - # to determine expected types in the class subscriptions. - - from collections.abc import Callable - - from pydantic._internal._model_construction import ModelMetaclass - - from streamdeck.types import EventNameStr - - class EventMeta(ModelMetaclass): - """Metaclass for EventBase stub to satisfy static type checkers.""" - @classmethod - def __getitem__(cls, event_names: LiteralString | tuple[LiteralString, ...]) -> type[EventBase]: ... - class EventBase(BaseModel, metaclass=EventMeta): - """Base class for all event models. + event: EventNameT + """Name of the event used to identify what occurred.""" - EventBase itself should not be instantiated, nor should it be subclassed directly. - Instead, use a subscripted subclass of EventBase, e.g. `EventBase["eventName"]`, to subclass from. + __event_names__: ClassVar[tuple[str, ...]] = () + """The event names declared by the generic EventBase specialization.""" - Examples: - ``` - class KeyDown(EventBase["keyDown"]): - # 'event' field's type annotation is internally set here as `Literal["keyDown"]` - ... + @classmethod + def __pydantic_init_subclass__(cls, **kwargs: Any) -> None: # noqa: ANN401 + """Validate EventBase generic arguments and require a specialization on subclasses.""" + super().__pydantic_init_subclass__(**kwargs) - class TestEvent(EventBase["test", "testing"]): - # 'event' field's type annotation is internally set here as `Literal["test", "testing"]` - ... + if cls is EventBase: + return - ``` - """ - event: EventNameStr - """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. - """ - __event_metadata__: ClassVar[EventMetadataDict] - """Metadata for specialized EventBase submodels.""" - __event_type__: ClassVar[Callable[[], type[object] | None]] - """Return the event type for the event model.""" - - @classmethod - def get_model_event_names(cls) -> tuple[EventNameStr, ...]: - """Return the event names for the event model.""" - ... - -else: - class EventBase(ConfiguredBaseModel, ABC): - """Base class for all event models. - - EventBase itself should not be instantiated, nor should it be subclassed directly. - Instead, use a subscripted subclass of EventBase, e.g. `EventBase["eventName"]`, to subclass from. - - Examples: - ``` - class KeyDown(EventBase["keyDown"]): - # 'event' field's type annotation is internally set here as `Literal["keyDown"]` - ... - - class TestEvent(EventBase["test", "testing"]): - # 'event' field's type annotation is internally set here as `Literal["test", "testing"]` - ... - - ``` - """ - # A weak reference dictionary to store subscripted subclasses of EventBase. Weak references are used to minimize memory usage. - _cached_specialized_submodels: ClassVar[WeakValueDictionary[str, type[EventBase]]] = WeakValueDictionary() - __event_metadata__: ClassVar[EventMetadataDict] - """Metadata for specialized EventBase submodels.""" - - 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 __new__(cls, *args: Any, **kwargs: Any) -> EventBase: - """Create a new instance of the event model. - - This method is called when the model is instantiated. - It validates that the event name is a string and not empty. - """ - if cls is EventBase: - raise TypeError("Can't instantiate abstract class EventBase directly.") - if EventBase in cls.__bases__: - raise TypeError(f"Can't instantiate abstract subscripted EventBase class {cls.__name__} directly.") - return super().__new__(cls) - - @override - @classmethod - def __class_getitem__(cls: type[EventBase], event_names: LiteralString | tuple[LiteralString, ...]) -> type[EventBase]: # type: ignore[override] - """Create a subclass alias of EventBase with the given event names. - - This method is called when the class is subscripted. - If the event names are not already registered, a new subclass is created. - Validations are performed to ensure that the event names are strings and not empty, and that the subscripted class is not a subclass of EventBase. - """ - if cls is not EventBase: # type: ignore[misc] - raise TypeError(f"Subclasses of EventBase are not subscriptable — '{cls.__name__}'") # noqa: TRY003, EM102 - - # Whether event_names is a string or tuple of strings, validate that it/they is/are a string - if not isinstance(event_names, tuple): - event_names = (event_names,) - - # Validate that all event names are strings - if any(not isinstance(name, str) for name in event_names): # type: ignore[misc] - raise TypeError(f"Event names must be strings, not {type(event_names).__name__}; args: {event_names}") # noqa: TRY003, EM102 - - subtype_name = f"{cls.__name__}[{event_names}]" - - return cls.__new_subscripted_base__(subtype_name, event_names) - - @classmethod - def __new_subscripted_base__(cls: type[EventBase], new_name: str, event_name_args: tuple[str, ...]) -> type[EventBase]: - """Dynamically create a new Singleton subclass of EventBase with the given event names for the event field. - - Only create a new subscripted subclass if it doesn't already exist in the _cached_specialized_submodels dictionary, otherwise return the existing subclass. - The new subclasses created here will be ignored in the __init_subclass__ method. - """ - if new_name not in cls._cached_specialized_submodels: - # Make sure not to pass in a value `_cached_specialized_submodels` in the create_model() call, in order to avoid shadowing the class variable. - cls._cached_specialized_submodels[new_name] = create_model( - new_name, - __base__=cls, - __event_metadata__=(EventMetadataDict, {"origin": cls, "args": event_name_args}), - event=(Literal[event_name_args], ...), - __cls_kwargs__={"_is_specialized_base": True}, # This gets passed to __init_subclass__ as a kwarg to indicate that this is a specialized (subscripted) subclass of EventBase. - ) - - return cls._cached_specialized_submodels[new_name] - - @classmethod - def __init_subclass__(cls, _is_specialized_base: bool = False) -> None: - """Validate a child class of EventBase (not a subscripted base subclass) is subclassing from a subscripted EventBase. - - Args: - _is_specialized_base: Whether this is a specialized submodel of EventBase (i.e., a subscripted subclass). - This should only be True for the subscripted subclasses created in __class_getitem__. - """ - if _is_specialized_base: - # This is a subscripted subclass of EventBase, so we don't need to do anything. - return - - if EventBase in cls.__bases__: - # Normally, only subscripted subclasses of EventBase (filtered out above) will have (un-subscripted) EventBase in their __bases__. - # If we get here, it means that this is a subclass of EventBase that is not subscripted, which is not allowed. - msg = f"Child classes of EventBase cannot subclass from non-subscripted EventBase. '{cls.__name__}' must subclass from a subscripted EventBase." + metadata = cls.__pydantic_generic_metadata__ + if metadata.get("origin") is EventBase: + args = metadata["args"] + if len(args) != 1 or get_origin(args[0]) is not Literal: + msg = "EventBase must be specialized with Literal[str, ...]" raise TypeError(msg) - def __str__(self) -> str: - """Return a string representation of the event model in the format 'EventName(event=event_name, field_name=field_value, etc.)'.""" - field_key_value_strs: list[str] = [f"{field_name}={getattr(self, field_name)}" for field_name in type(self).model_fields] - return f"{self.__class__.__name__}({', '.join(field_key_value_strs)})" - - @classmethod - def __event_type__(cls) -> type[object] | None: - """Return the event type for the event model.""" - return cls.model_fields["event"].annotation + names = get_args(args[0]) + if not names or any(type(name) is not str for name in names): + msg = "EventBase Literal arguments must be strings" + raise TypeError(msg) - @classmethod - def get_model_event_names(cls) -> tuple[str, ...]: - """Return the event names for the event model.""" - return cls.__event_metadata__["args"] + cls.__event_names__ = names + return + + if not any( + getattr(base, "__pydantic_generic_metadata__", {}).get("origin") is EventBase + for base in cls.__mro__[1:] + ): + msg = ( + "Child classes of EventBase must subclass from a specialized EventBase. " + f"'{cls.__name__}' must use EventBase[Literal[...]]." + ) + raise TypeError(msg) + + def __new__(cls, *args: Any, **kwargs: Any) -> EventBase[Any]: # noqa: ANN401, ARG004 + """Prevent instantiation of EventBase and its generic specializations.""" + metadata = cls.__pydantic_generic_metadata__ + if cls is EventBase or metadata.get("origin") is EventBase: + msg = f"Can't instantiate abstract event model {cls.__name__} directly." + raise TypeError(msg) + return super().__new__(cls) + + def __str__(self) -> str: + """Return a readable representation of the event model and its fields.""" + field_key_value_strs = [ + f"{field_name}={getattr(self, field_name)}" for field_name in type(self).model_fields + ] + return f"{self.__class__.__name__}({', '.join(field_key_value_strs)})" + + @classmethod + def __event_type__(cls) -> Any: # noqa: ANN401 + """Return the Pydantic field annotation for this event model.""" + return cls.model_fields["event"].annotation + + @classmethod + def get_model_event_names(cls) -> tuple[str, ...]: + """Return the event names declared by this model's EventBase specialization.""" + return cls.__event_names__ diff --git a/streamdeck/models/events/deep_link.py b/streamdeck/models/events/deep_link.py index 791e6b2..33dfbe3 100644 --- a/streamdeck/models/events/deep_link.py +++ b/streamdeck/models/events/deep_link.py @@ -1,18 +1,22 @@ +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["didReceiveDeepLink"]): +class DidReceiveDeepLink(EventBase[Literal["didReceiveDeepLink"]]): """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}. """ + 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 index 82b53d5..ec1640e 100644 --- a/streamdeck/models/events/devices.py +++ b/streamdeck/models/events/devices.py @@ -40,18 +40,21 @@ 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)] @@ -78,11 +81,12 @@ def __repr__(self) -> str: return f"DeviceInfo(name={self.name}, type={self.type}, size={self.size})" -class DeviceDidConnect(EventBase["deviceDidConnect"], DeviceSpecificEventMixin): +class DeviceDidConnect(EventBase[Literal["deviceDidConnect"]], DeviceSpecificEventMixin): """Occurs when a Stream Deck device is connected.""" + device_info: Annotated[DeviceInfo, Field(alias="deviceInfo")] """Information about the newly connected device.""" -class DeviceDidDisconnect(EventBase["deviceDidDisconnect"], DeviceSpecificEventMixin): +class DeviceDidDisconnect(EventBase[Literal["deviceDidDisconnect"]], DeviceSpecificEventMixin): """Occurs when a Stream Deck device is disconnected.""" diff --git a/streamdeck/models/events/dials.py b/streamdeck/models/events/dials.py index 1e9ba7d..bc7bcff 100644 --- a/streamdeck/models/events/dials.py +++ b/streamdeck/models/events/dials.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import Literal + from streamdeck.models.events.base import EventBase from streamdeck.models.events.common import ( BasePayload, @@ -12,12 +14,14 @@ ## 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 @@ -26,19 +30,23 @@ class DialRotatePayload(EncoderPayload): ## Event models for DialDown, DialRotate, and DialUp events -class DialDown(EventBase["dialDown"], ContextualEventMixin, DeviceSpecificEventMixin): + +class DialDown(EventBase[Literal["dialDown"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user presses a dial (Stream Deck +).""" + payload: EncoderPayload """Contextualized information for this event.""" -class DialRotate(EventBase["dialRotate"], ContextualEventMixin, DeviceSpecificEventMixin): +class DialRotate(EventBase[Literal["dialRotate"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user rotates a dial (Stream Deck +).""" + payload: DialRotatePayload """Contextualized information for this event.""" -class DialUp(EventBase["dialUp"], ContextualEventMixin, DeviceSpecificEventMixin): +class DialUp(EventBase[Literal["dialUp"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user releases a pressed dial (Stream Deck +).""" + payload: EncoderPayload """Contextualized information for this event.""" diff --git a/streamdeck/models/events/keys.py b/streamdeck/models/events/keys.py index cb368c1..c874d95 100644 --- a/streamdeck/models/events/keys.py +++ b/streamdeck/models/events/keys.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Annotated +from typing import Annotated, Literal from pydantic import Field @@ -39,6 +39,7 @@ class MultiActionKeyGesturePayload( 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. @@ -48,13 +49,16 @@ class MultiActionKeyGesturePayload( ## Event models for KeyDown and KeyUp events -class KeyDown(EventBase["keyDown"], ContextualEventMixin, DeviceSpecificEventMixin): + +class KeyDown(EventBase[Literal["keyDown"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user presses a action down.""" + payload: CardinalityDiscriminated[SingleActionKeyGesturePayload, MultiActionKeyGesturePayload] """Contextualized information for this event.""" -class KeyUp(EventBase["keyUp"], ContextualEventMixin, DeviceSpecificEventMixin): +class KeyUp(EventBase[Literal["keyUp"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user releases a pressed action.""" + 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 index 4d5fe33..d879845 100644 --- a/streamdeck/models/events/property_inspector.py +++ b/streamdeck/models/events/property_inspector.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import Literal + from streamdeck.models.events.base import EventBase from streamdeck.models.events.common import ( ContextualEventMixin, @@ -8,20 +10,25 @@ from streamdeck.types import PluginDefinedData # noqa: TC001 -class DidReceivePropertyInspectorMessage(EventBase["sendToPlugin"], ContextualEventMixin): +class DidReceivePropertyInspectorMessage(EventBase[Literal["sendToPlugin"]], ContextualEventMixin): """Occurs when a message was received from the UI.""" + payload: PluginDefinedData """The data payload received from the UI.""" -class PropertyInspectorDidAppear(EventBase["propertyInspectorDidAppear"], ContextualEventMixin, DeviceSpecificEventMixin): +class PropertyInspectorDidAppear( + EventBase[Literal["propertyInspectorDidAppear"]], + 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. """ -class PropertyInspectorDidDisappear(EventBase["propertyInspectorDidDisappear"], ContextualEventMixin, DeviceSpecificEventMixin): +class PropertyInspectorDidDisappear(EventBase[Literal["propertyInspectorDidDisappear"]], ContextualEventMixin, DeviceSpecificEventMixin): # fmt: skip """Occurs when the property inspector associated with the action becomes invisible. I.e. the user unselected the action in the Stream Deck application. diff --git a/streamdeck/models/events/settings.py b/streamdeck/models/events/settings.py index 61a86fb..f066a4d 100644 --- a/streamdeck/models/events/settings.py +++ b/streamdeck/models/events/settings.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import Literal + from streamdeck.models.events.base import ConfiguredBaseModel, EventBase from streamdeck.models.events.common import ( BasePayload, @@ -38,21 +40,27 @@ class MultiActionSettingsPayload( """ -class DidReceiveSettings(EventBase["didReceiveSettings"], ContextualEventMixin, DeviceSpecificEventMixin): +class DidReceiveSettings( + EventBase[Literal["didReceiveSettings"]], ContextualEventMixin, DeviceSpecificEventMixin +): """Occurs when the settings associated with an action instance are requested, or when the the settings were updated by the property inspector.""" + 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["didReceiveGlobalSettings"]): +class DidReceiveGlobalSettings(EventBase[Literal["didReceiveGlobalSettings"]]): """Occurs when the plugin receives the global settings from the Stream Deck.""" + payload: GlobalSettingsPayload """Additional information about the event that occurred.""" diff --git a/streamdeck/models/events/system.py b/streamdeck/models/events/system.py index 1d8be50..040377d 100644 --- a/streamdeck/models/events/system.py +++ b/streamdeck/models/events/system.py @@ -1,7 +1,9 @@ from __future__ import annotations +from typing import Literal + from streamdeck.models.events.base import EventBase -class SystemDidWakeUp(EventBase["systemDidWakeUp"]): +class SystemDidWakeUp(EventBase[Literal["systemDidWakeUp"]]): """Occurs when the computer wakes up.""" diff --git a/streamdeck/models/events/title_parameters.py b/streamdeck/models/events/title_parameters.py index 8446900..b7c9c7f 100644 --- a/streamdeck/models/events/title_parameters.py +++ b/streamdeck/models/events/title_parameters.py @@ -20,6 +20,7 @@ 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")] @@ -42,12 +43,14 @@ class TitleParametersDidChangePayload( 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["titleParametersDidChange"], ContextualEventMixin, DeviceSpecificEventMixin): +class TitleParametersDidChange(EventBase[Literal["titleParametersDidChange"]], ContextualEventMixin, DeviceSpecificEventMixin): # fmt: skip """Occurs when the user updates an action's title settings in the Stream Deck application.""" + payload: TitleParametersDidChangePayload diff --git a/streamdeck/models/events/touch_tap.py b/streamdeck/models/events/touch_tap.py index 9a14688..b3717fa 100644 --- a/streamdeck/models/events/touch_tap.py +++ b/streamdeck/models/events/touch_tap.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Annotated +from typing import Annotated, Literal from pydantic import Field @@ -16,13 +16,15 @@ 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["touchTap"], ContextualEventMixin, DeviceSpecificEventMixin): +class TouchTap(EventBase[Literal["touchTap"]], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user taps the touchscreen (Stream Deck +).""" + payload: TouchTapPayload """Contextualized information for this event.""" diff --git a/streamdeck/models/events/visibility.py b/streamdeck/models/events/visibility.py index 43bd871..d0396f7 100644 --- a/streamdeck/models/events/visibility.py +++ b/streamdeck/models/events/visibility.py @@ -1,5 +1,7 @@ from __future__ import annotations +from typing import Literal + from streamdeck.models.events.base import EventBase from streamdeck.models.events.common import ( BasePayload, @@ -33,18 +35,21 @@ class MultiActionVisibilityPayload( ## Event models for WillAppear and WillDisappear events -class WillAppear(EventBase["willAppear"], ContextualEventMixin, DeviceSpecificEventMixin): + +class WillAppear(EventBase[Literal["willAppear"]], 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. """ + payload: CardinalityDiscriminated[SingleActionVisibilityPayload, MultiActionVisibilityPayload] -class WillDisappear(EventBase["willDisappear"], ContextualEventMixin, DeviceSpecificEventMixin): +class WillDisappear(EventBase[Literal["willDisappear"]], ContextualEventMixin, DeviceSpecificEventMixin): # fmt: skip """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. """ + payload: CardinalityDiscriminated[SingleActionVisibilityPayload, MultiActionVisibilityPayload] diff --git a/streamdeck/types.py b/streamdeck/types.py index a449b50..f533baf 100644 --- a/streamdeck/types.py +++ b/streamdeck/types.py @@ -1,4 +1,3 @@ -# ruff: noqa: UP040 from __future__ import annotations from typing import TYPE_CHECKING @@ -7,8 +6,7 @@ if TYPE_CHECKING: - from typing_extensions import TypeAlias # noqa: UP035 - + from typing import TypeAlias EventNameStr: TypeAlias = str diff --git a/tests/data/dummy_event_listener.py b/tests/data/dummy_event_listener.py index c0a3546..0486fdd 100644 --- a/tests/data/dummy_event_listener.py +++ b/tests/data/dummy_event_listener.py @@ -2,9 +2,10 @@ from typing import TYPE_CHECKING, Literal +from typing_extensions import override # noqa: UP035 + from streamdeck.event_listener import EventListener from streamdeck.models.events import EventBase -from typing_extensions import override # noqa: UP035 if TYPE_CHECKING: @@ -12,13 +13,15 @@ from typing import ClassVar -class DummyEvent(EventBase["dummy"]): +class DummyEvent(EventBase[Literal["dummy"]]): """A dummy event for testing purposes.""" + something: int class DummyEventListener(EventListener): """A dummy event listener for testing purposes.""" + event_models: ClassVar[list[type[EventBase]]] = [DummyEvent] @override diff --git a/tests/models/events/test_base.py b/tests/models/events/test_base.py index a8e251f..ea4f851 100644 --- a/tests/models/events/test_base.py +++ b/tests/models/events/test_base.py @@ -1,174 +1,88 @@ +"""Tests for the generic incoming event base model.""" + from __future__ import annotations -from contextlib import nullcontext -from typing import TYPE_CHECKING, Annotated, Any, Union +from typing import Any, Literal import pytest -from pydantic import Discriminator, TypeAdapter -from streamdeck.models.events.base import EventBase +from pydantic import TypeAdapter, ValidationError - -if TYPE_CHECKING: - from _pytest.python_api import RaisesContext # type: ignore[import] - - -class TestBaseSubscripting: - """Test expected behavior of subscripting the base class.""" - common_test_parameters: pytest.MarkDecorator = pytest.mark.parametrize( - ("subscript_value", "expected_context"), - [ - ("ok", nullcontext()), - (404, pytest.raises(TypeError)), - (False, pytest.raises(TypeError)), - ], - ) - - @common_test_parameters - def test_base_type_subscripting( - self, - subscript_value: Any, - expected_context: RaisesContext[TypeError] | nullcontext[None], - ) -> None: - """Test that only string types are allowed for base class subscripting.""" - with expected_context: - EventBase[subscript_value] - - @common_test_parameters - def test_base_type_subscription_subclassing( - self, - subscript_value: Any, - expected_context: RaisesContext[TypeError] | nullcontext[None], - ) -> None: - """Test that only string types are allowed for base class subscripting--even in subclasses.""" - with expected_context: - class MyEvent(EventBase[subscript_value]): # type: ignore[valid-type] - """Event model subclass with invalid event field type.""" - - def test_subscripting_subclass_raises_type_error(self) -> None: - """Test that subscripting the subclass raises a TypeError.""" - class MyEvent(EventBase["ok"]): ... - with pytest.raises(TypeError): - MyEvent["not_ok"] - - with pytest.raises(TypeError): - class BadMyEventSubclass(MyEvent["also_not_ok"]): ... # type: ignore[valid-type] - - def test_no_base_subscript_in_subclass_raises_type_error(self) -> None: - """Test that subscripting the base class without a value raises a TypeError.""" - with pytest.raises(TypeError): - class BadNoSubscriptEvent(EventBase): # type: ignore[valid-type] - ... - - # def test_overridden_event_name_raises_type_error(self) -> None: - # """Test that overriding the event name in a subclass raises a TypeError.""" - # with pytest.raises(TypeError): - # class MyEvent(EventBase["ok"]): - - - -def test_init_base_class_raises_error() -> None: - """Test that initializing the base class directly raises a TypeError.""" - with pytest.raises(TypeError): - EventBase(event="uhh") +from streamdeck.models.events.base import EventBase -def test_init_subscripted_subclass_raises_error() -> None: - """Test that initializing a subscripted subclass of the base class raises a TypeError.""" +@pytest.mark.parametrize( + "event_name_type", + [str, Literal[404], Literal[False], Literal["ok", 404]], +) +def test_event_name_type_must_be_literal_of_strings(event_name_type: Any) -> None: # noqa: ANN401 + """Only Literal types containing strings can specialize EventBase.""" with pytest.raises(TypeError): - EventBase["uhh"](event="uhh") - - + EventBase[event_name_type] -class TestBaseComparisons: - """Test expected behavior of comparisons between EventBase and its subscripted subclasses.""" - def test_base_subtype_singleton_equality(self) -> None: - concrete_base_subtype1 = EventBase["conc1"] - assert concrete_base_subtype1 == EventBase["conc1"] - assert concrete_base_subtype1 is EventBase["conc1"] +def test_event_base_must_be_specialized_when_subclassed() -> None: + """Concrete event models must specialize EventBase with a Literal.""" + with pytest.raises(TypeError, match="specialized EventBase"): + class BadEvent(EventBase): ... - def test_base_different_subtypes_inequality(self) -> None: - concrete_base_subtype1 = EventBase["conc1"] - concrete_base_subtype2 = EventBase["conc2"] - assert concrete_base_subtype1 != concrete_base_subtype2 +def test_string_literal_specialization_records_event_names() -> None: + """A concrete event subclass inherits the names from its Literal specialization.""" + class MyEvent(EventBase[Literal["ok"]]): ... - def test_base_subtype_issubclass(self) -> None: - concrete_base_subtype1 = EventBase["conc1"] + assert MyEvent.get_model_event_names() == ("ok",) + assert MyEvent.__event_type__() == Literal["ok"] - class ConcreteEvent1(EventBase["conc1"]): ... - assert issubclass(ConcreteEvent1, EventBase) - assert issubclass(ConcreteEvent1, concrete_base_subtype1) +def test_multiple_string_literal_values_are_supported() -> None: + """A single event model may declare more than one accepted event name.""" + class MyEvent(EventBase[Literal["ok", "also_ok"]]): ... - def test_base_different_subtypes_not_issubclass(self) -> None: - concrete_base_subtype1 = EventBase["conc1"] - concrete_base_subtype2 = EventBase["conc2"] + assert MyEvent.get_model_event_names() == ("ok", "also_ok") + assert TypeAdapter(MyEvent).validate_python({"event": "ok"}) == MyEvent(event="ok") + assert TypeAdapter(MyEvent).validate_python({"event": "also_ok"}) == MyEvent(event="also_ok") + with pytest.raises(ValidationError): + TypeAdapter(MyEvent).validate_python({"event": "bad"}) - class ConcreteEvent1(EventBase["conc1"]): ... - class ConcreteEvent2(EventBase["conc2"]): ... - assert not issubclass(ConcreteEvent1, concrete_base_subtype2) - assert not issubclass(ConcreteEvent2, concrete_base_subtype1) - - - def test_subclass_isinstance_of_base_subtype(self) -> None: - class ConcreteEvent1(EventBase["conc1"]): ... - concrete_event_instance1 = ConcreteEvent1(event="conc1") - - assert isinstance(concrete_event_instance1, EventBase) - assert isinstance(concrete_event_instance1, EventBase["conc1"]) - - - def test_subclass_not_isinstance_of_different_subtypes(self) -> None: - concrete_base_subtype1 = EventBase["conc1"] - concrete_base_subtype2 = EventBase["conc2"] - - class ConcreteEvent1(EventBase["conc1"]): ... - class ConcreteEvent2(EventBase["conc2"]): ... - - concrete_event_instance1 = ConcreteEvent1(event="conc1") - concrete_event_instance2 = ConcreteEvent2(event="conc2") +def test_event_base_and_generic_specialization_cannot_be_instantiated() -> None: + """Only concrete event model subclasses can be instantiated.""" + with pytest.raises(TypeError, match="directly"): + EventBase(event="uhh") - assert not isinstance(concrete_event_instance1, ConcreteEvent2) - assert not isinstance(concrete_event_instance1, concrete_base_subtype2) - assert not isinstance(concrete_event_instance2, concrete_base_subtype1) + with pytest.raises(TypeError, match="directly"): + EventBase[Literal["uhh"]](event="uhh") - def test_subclass_is_not_base_subtype(self) -> None: - concrete_base_subtype1 = EventBase["conc1"] +def test_generic_specializations_are_cached() -> None: + """Pydantic returns the same specialized class for repeated type arguments.""" + specialized_base = EventBase[Literal["conc1"]] - class ConcreteEvent1(EventBase["conc1"]): ... + assert specialized_base is EventBase[Literal["conc1"]] + assert specialized_base != EventBase[Literal["conc2"]] - # ConcreteEvent1 is an instance of concrete_event_instance1, but not equal to nor "is" concrete_event_instance1 itself. - assert ConcreteEvent1 != concrete_base_subtype1 - assert ConcreteEvent1 is not concrete_base_subtype1 +def test_concrete_subclass_is_a_subclass_of_its_specialized_base() -> None: + """Concrete event models inherit the specialized Pydantic model class.""" + specialized_base = EventBase[Literal["conc1"]] + class ConcreteEvent(specialized_base): ... -def test_type_adapter() -> None: - """Test that Event subclasses can be discriminated by the event field.""" - class ConcreteEvent(EventBase["concrete"]): ... - class TestEvent(EventBase["test"]): ... - class AnotherEvent(EventBase["another"]): ... - class YetAnotherEvent(EventBase["yet_another"]): ... + event = ConcreteEvent(event="conc1") + assert issubclass(ConcreteEvent, EventBase) + assert issubclass(ConcreteEvent, specialized_base) + assert isinstance(event, EventBase) + assert isinstance(event, specialized_base) - # Create a TypeAdapter for the base class - adapter: TypeAdapter[EventBase] = TypeAdapter( - Annotated[Union[ConcreteEvent, TestEvent, AnotherEvent, YetAnotherEvent], Discriminator("event")] # noqa: UP007 - ) - concrete_event_instance = adapter.validate_json('{"event": "concrete"}') - assert isinstance(concrete_event_instance, ConcreteEvent) +def test_subclasses_of_event_models_keep_event_metadata() -> None: + """Further subclasses retain the event name metadata of their base model.""" - test_event_instance = adapter.validate_json('{"event": "test"}') - assert isinstance(test_event_instance, TestEvent) + class ConcreteEvent(EventBase[Literal["conc1"]]): ... - another_event_instance = adapter.validate_json('{"event": "another"}') - assert isinstance(another_event_instance, AnotherEvent) + class ChildEvent(ConcreteEvent): ... - yet_another_event_instance = adapter.validate_json('{"event": "yet_another"}') - assert isinstance(yet_another_event_instance, YetAnotherEvent) + assert ChildEvent.get_model_event_names() == ("conc1",) diff --git a/tests/models/test_event_adapter.py b/tests/models/test_event_adapter.py index a4a51ae..80e35c7 100644 --- a/tests/models/test_event_adapter.py +++ b/tests/models/test_event_adapter.py @@ -3,11 +3,11 @@ import pytest from pydantic import ValidationError + from streamdeck.models.events import DEFAULT_EVENT_MODELS, DEFAULT_EVENT_NAMES, EventBase, KeyDown from streamdeck.models.events.adapter import ( EventAdapter, ) - from tests.test_utils.fake_event_factories import KeyDownEventFactory @@ -25,12 +25,13 @@ def test_init_with_default_models() -> None: for model in DEFAULT_EVENT_MODELS: assert model in adapter._models + def test_add_model() -> None: """Test that models can be added to the adapter.""" adapter = EventAdapter() # Create a fake event model - class DummyEvent(EventBase["dummyEvent"]): + class DummyEvent(EventBase[Literal["dummyEvent"]]): payload: dict[str, str] # Add the custom model @@ -44,6 +45,7 @@ class DummyEvent(EventBase["dummyEvent"]): assert DummyEvent in adapter._models assert len(adapter._models) == len(DEFAULT_EVENT_MODELS) + 1 + def test_event_name_exists() -> None: """Test that event_name_exists correctly identifies registered events.""" adapter = EventAdapter() @@ -55,6 +57,7 @@ def test_event_name_exists() -> None: # Should return False for non-registered events assert not adapter.event_name_exists("nonExistentEvent") + def test_type_adapter_creation() -> None: """Test that the type_adapter property creates a TypeAdapter.""" adapter = EventAdapter() @@ -66,6 +69,7 @@ def test_type_adapter_creation() -> None: # Second access should return the same instance assert adapter.type_adapter is type_adapter + def test_validate_json_with_valid_data() -> None: """Test that validate_json correctly parses valid JSON into the correct event model instance.""" adapter = EventAdapter() @@ -81,10 +85,20 @@ def test_validate_json_with_valid_data() -> None: assert isinstance(event, KeyDown) assert event == fake_key_down_event -@pytest.mark.parametrize("invalid_event_json", [ - json.dumps({"event": "keyDown"}), # Missing required fields: action, context, device - json.dumps({"event": "unknownEvent", "action": "com.example.plugin.action", "context": "context123"}) # Unknown event -]) + +@pytest.mark.parametrize( + "invalid_event_json", + [ + json.dumps({"event": "keyDown"}), # Missing required fields: action, context, device + json.dumps( + { + "event": "unknownEvent", + "action": "com.example.plugin.action", + "context": "context123", + } + ), # Unknown event + ], +) def test_validate_json_with_invalid_data(invalid_event_json: str) -> None: """Test that validate_json raises an error for invalid JSON. @@ -103,7 +117,7 @@ def test_adding_custom_event_allows_validation() -> None: adapter = EventAdapter() # Create a fake event model - class DummyEvent(EventBase["dummyEvent"]): + class DummyEvent(EventBase[Literal["dummyEvent"]]): action: str context: str device: str @@ -118,7 +132,7 @@ class DummyEvent(EventBase["dummyEvent"]): action="com.example.plugin.action", context="context123", device="device123", - payload={"key": "value"} + payload={"key": "value"}, ) fake_event_json = fake_event.model_dump_json() diff --git a/tests/plugin_manager/conftest.py b/tests/plugin_manager/conftest.py index 02a4194..b77f924 100644 --- a/tests/plugin_manager/conftest.py +++ b/tests/plugin_manager/conftest.py @@ -36,6 +36,12 @@ def plugin_manager(port_number: int, plugin_registration_uuid: str) -> PluginMan ) +@pytest.fixture(autouse=True) +def mock_streamdeck_logger(monkeypatch: pytest.MonkeyPatch) -> None: + """Keep plugin manager tests independent of the user's log directory.""" + monkeypatch.setattr("streamdeck.manager.configure_streamdeck_logger", Mock()) + + @pytest.fixture def patch_websocket_client(monkeypatch: pytest.MonkeyPatch) -> Mock: """Fixture that uses pytest's MonkeyPatch to mock WebSocketClient for the PluginManager run method.