From f76fd22097d71ea61393a3486a345040f011a91f Mon Sep 17 00:00:00 2001 From: strohganoff Date: Mon, 21 Apr 2025 17:12:37 -0600 Subject: [PATCH 1/6] Replace generic subclass of EventBase with implementation; write tests --- streamdeck/models/events/base.py | 216 ++++++++++++++++--------------- tests/models/events/test_base.py | 174 +++++++++++++++++++++++++ 2 files changed, 287 insertions(+), 103 deletions(-) create mode 100644 tests/models/events/test_base.py diff --git a/streamdeck/models/events/base.py b/streamdeck/models/events/base.py index 78f291f..642056f 100644 --- a/streamdeck/models/events/base.py +++ b/streamdeck/models/events/base.py @@ -1,18 +1,10 @@ from __future__ import annotations from abc import ABC -from typing import Annotated, Any, Generic, Literal, get_args, get_origin +from typing import TYPE_CHECKING, Any, ClassVar, Literal -from pydantic import BaseModel, ConfigDict, GetPydanticSchema -from pydantic._internal._generics import get_origin as get_model_origin # type: ignore[import] -from pydantic_core import core_schema -from typing_extensions import ( # noqa: UP035 - LiteralString, - TypeAlias, - TypeGuard, - TypeVar, - override, -) +from pydantic import BaseModel, ConfigDict, create_model +from typing_extensions import LiteralString, override # noqa: UP035 class ConfiguredBaseModel(BaseModel, ABC): @@ -37,104 +29,122 @@ def model_dump_json(self, **kwargs: Any) -> str: return super().model_dump_json(**kwargs) -# We do this to get the typing module's _LiteralGenericAlias type, which is not formally exported. -_LiteralStrGenericAlias: TypeAlias = type(Literal["whatever"]) # type: ignore[valid-type] # noqa: UP040 -"""A generic alias for a Literal type used for internal mechanisms of this module. +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. -This is opposed to LiteralStrGenericAlias which is used for typing. -""" + from pydantic._internal._model_construction import ModelMetaclass # type: ignore[import] + class EventMeta(ModelMetaclass): + """Metaclass for EventBase stub to satisfy static type checkers.""" + @classmethod + def __getitem__(cls, event_names: LiteralString | tuple[LiteralString, ...]) -> type[EventBase]: ... -# Set this variable here to call the function just once. -_pydantic_str_schema = core_schema.str_schema() + class EventBase(BaseModel, metaclass=EventMeta): + """Base class for all event models.""" + event: LiteralString + """Name of the event used to identify what occurred. -GetPydanticStrSchema = GetPydanticSchema(lambda _ts, handler: handler(_pydantic_str_schema)) -"""A function that returns a Pydantic schema for a string type.""" - -PydanticLiteralStrGenericAlias: TypeAlias = Annotated[ # type: ignore[valid-type] # noqa: UP040 - _LiteralStrGenericAlias, - GetPydanticStrSchema, -] -"""A Pydantic-compatible generic alias for a Literal type. - -Pydantic will treat a field of this type as a string schema, while static type checkers -will still treat it as a _LiteralGenericAlias type. - -Even if a subclass of EventBase uses a Literal with multiple string values, -an event message will only ever have one of those values in the event field, -and so we don't need to handle this with a more complex Pydantic schema. -""" - - -# This type alias is used to handle static type checking accurately while still conveying that -# a value is expected to be a Literal with string type args. -LiteralStrGenericAlias: TypeAlias = Annotated[ # noqa: UP040 - LiteralString, - GetPydanticStrSchema, -] -"""Type alias for a generic literal string type that is compatible with Pydantic.""" - - -# covariant=True is used to allow subclasses of EventBase to be used in place of the base class. -LiteralEventName_co = TypeVar("LiteralEventName_co", bound=PydanticLiteralStrGenericAlias, default=PydanticLiteralStrGenericAlias, covariant=True) -"""Type variable for a Literal type with string args.""" - - -def is_literal_str_generic_alias_type(value: object | None) -> TypeGuard[LiteralStrGenericAlias]: - """Check if a type is a concrete Literal type with string args.""" - if value is None: - return False - - if isinstance(value, TypeVar): - return False - - event_field_base_type = get_origin(value) - - 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, Generic[LiteralEventName_co]): - """Base class for event models that represent Stream Deck Plugin SDK events.""" - - event: LiteralEventName_co - """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) - - # This is a GenericAlias (likely used in the subclass definition, i.e. `class ConcreteEvent(EventBase[Literal["event_name"]]):`) which is technically a subclass. - # We can safely ignore this case, as we only want to validate the concrete subclass itself (`ConscreteEvent`). - if get_model_origin(cls) is None: - return - - model_event_type = cls.__event_type__() + Subclass models must define this field as a Literal type with the event name string that the model represents. + """ - if not is_literal_str_generic_alias_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_names(cls) -> tuple[str, ...]: + """Return the event names for the event model.""" + ... - @classmethod - def __event_type__(cls) -> type[object]: - """Get the type annotations of the subclass model's event field.""" - return cls.model_fields["event"].annotation # type: ignore[index] +else: + class EventBase(ConfiguredBaseModel, ABC): + """Base class for all event models.""" + _subtypes: ClassVar[dict[str, type[EventBase]]] = {} + __args__: ClassVar[tuple[str, ...]] - @classmethod - def get_model_event_names(cls) -> tuple[str, ...]: - """Get the value of the subclass model's event field Literal annotation.""" - model_event_type = cls.__event_type__() + event: str + """Name of the event used to identify what occurred. - # Ensure that the event field annotation is a Literal type. - if not is_literal_str_generic_alias_type(model_event_type): - msg = f"The event field annotation of an Event model must be a Literal[str] type. Given type: {model_event_type}" - raise TypeError(msg) + Subclass models must define this field as a Literal type with the event name string that the model represents. + """ - return get_args(model_event_type) + 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 _subtypes 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._subtypes: + # Pass in the _is_base_subtype kwarg to __init_subclass__ to indicate that this is a base subtype of EventBase, and should be ignored. + cls._subtypes[new_name] = create_model( + new_name, + __base__=cls, + __args__=(tuple[str, ...], event_name_args), + event=(Literal[event_name_args], ...), + __cls_kwargs__={"_is_base_subtype": True}, + ) + + return cls._subtypes[new_name] + + @classmethod + def __init_subclass__(cls, _is_base_subtype: bool = False) -> None: + """Validate a child class of EventBase (not a subscripted base subclass) is subclassing from a subscripted EventBase.""" + if _is_base_subtype: + # 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." + 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 + + @classmethod + def get_model_event_names(cls) -> tuple[str, ...]: + """Return the event names for the event model.""" + return cls.__args__ diff --git a/tests/models/events/test_base.py b/tests/models/events/test_base.py new file mode 100644 index 0000000..a8e251f --- /dev/null +++ b/tests/models/events/test_base.py @@ -0,0 +1,174 @@ +from __future__ import annotations + +from contextlib import nullcontext +from typing import TYPE_CHECKING, Annotated, Any, Union + +import pytest +from pydantic import Discriminator, TypeAdapter +from streamdeck.models.events.base import EventBase + + +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") + + +def test_init_subscripted_subclass_raises_error() -> None: + """Test that initializing a subscripted subclass of the base class raises a TypeError.""" + with pytest.raises(TypeError): + EventBase["uhh"](event="uhh") + + + +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_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_base_subtype_issubclass(self) -> None: + concrete_base_subtype1 = EventBase["conc1"] + + class ConcreteEvent1(EventBase["conc1"]): ... + + assert issubclass(ConcreteEvent1, EventBase) + assert issubclass(ConcreteEvent1, concrete_base_subtype1) + + + def test_base_different_subtypes_not_issubclass(self) -> None: + concrete_base_subtype1 = EventBase["conc1"] + concrete_base_subtype2 = EventBase["conc2"] + + 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") + + 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) + + + def test_subclass_is_not_base_subtype(self) -> None: + concrete_base_subtype1 = EventBase["conc1"] + + class ConcreteEvent1(EventBase["conc1"]): ... + + # 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_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"]): ... + + # 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) + + test_event_instance = adapter.validate_json('{"event": "test"}') + assert isinstance(test_event_instance, TestEvent) + + another_event_instance = adapter.validate_json('{"event": "another"}') + assert isinstance(another_event_instance, AnotherEvent) + + yet_another_event_instance = adapter.validate_json('{"event": "yet_another"}') + assert isinstance(yet_another_event_instance, YetAnotherEvent) From 8efe19da54574bc328eb4fa22d47725ec8076c78 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Mon, 21 Apr 2025 17:34:19 -0600 Subject: [PATCH 2/6] Remove generic alias type alias from EventBase annotations --- streamdeck/actions.py | 7 ++++--- streamdeck/event_listener.py | 3 +-- streamdeck/manager.py | 5 ++--- streamdeck/models/events/__init__.py | 4 +--- streamdeck/models/events/adapter.py | 12 ++++++------ streamdeck/models/events/application.py | 6 ++---- streamdeck/models/events/deep_link.py | 4 +--- streamdeck/models/events/devices.py | 4 ++-- streamdeck/models/events/dials.py | 8 +++----- streamdeck/models/events/keys.py | 6 +++--- streamdeck/models/events/property_inspector.py | 12 ++++-------- streamdeck/models/events/settings.py | 6 ++---- streamdeck/models/events/system.py | 4 +--- streamdeck/models/events/title_parameters.py | 2 +- streamdeck/models/events/touch_tap.py | 4 ++-- streamdeck/models/events/visibility.py | 6 ++---- .../test_action_event_handler_filtering.py | 18 ++++++++---------- tests/actions/test_action_registry.py | 7 +++---- tests/actions/test_actions.py | 9 ++++----- tests/event_listener/test_event_listener.py | 7 +++---- tests/models/test_event_adapter.py | 6 ++---- .../test_command_sender_binding.py | 9 ++++----- tests/plugin_manager/test_plugin_manager.py | 3 +-- 23 files changed, 62 insertions(+), 90 deletions(-) diff --git a/streamdeck/actions.py b/streamdeck/actions.py index f69682c..8c67f36 100644 --- a/streamdeck/actions.py +++ b/streamdeck/actions.py @@ -49,7 +49,7 @@ def _wrapper(func: EventHandlerFunc[TEvent_contra]) -> EventHandlerFunc[TEvent_c return _wrapper - def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc[EventBase[LiteralStrGenericAlias]], None, None]: + def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc[EventBase], None, None]: """Get all event handlers for a specific event. Args: @@ -66,7 +66,7 @@ def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHand yield from self._events[event_name] - def get_registered_event_names(self) -> list[str]: + def get_registered_event_names(self) -> list[EventNameStr]: """Get all event names for which event handlers are registered. Returns: @@ -74,6 +74,7 @@ def get_registered_event_names(self) -> list[str]: """ return list(self._events.keys()) + class GlobalAction(ActionBase): """Represents an action that is performed at the plugin level, meaning it isn't associated with a specific device or action.""" @@ -111,7 +112,7 @@ def register(self, action: ActionBase) -> None: """ self._plugin_actions.append(action) - def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | None = None) -> Generator[EventHandlerFunc[EventBase[LiteralStrGenericAlias]], None, None]: + def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | None = None) -> Generator[EventHandlerFunc[EventBase], None, None]: """Get all event handlers for a specific event from all registered actions. Args: diff --git a/streamdeck/event_listener.py b/streamdeck/event_listener.py index 0aba535..2e1da2e 100644 --- a/streamdeck/event_listener.py +++ b/streamdeck/event_listener.py @@ -14,7 +14,6 @@ from typing_extensions import TypeIs from streamdeck.models.events import EventBase - from streamdeck.models.events.base import LiteralStrGenericAlias @@ -120,7 +119,7 @@ class EventListener(ABC): Event listeners are classes that listen for events and simply yield them as they come. The EventListenerManager will handle the threading and pushing the events yielded into a shared queue. """ - event_models: ClassVar[list[type[EventBase[LiteralStrGenericAlias]]]] + event_models: ClassVar[list[type[EventBase]]] """A list of event models that the listener can yield. Read in by the PluginManager to model the incoming event data off of. The plugin-developer must define this list in their subclass. diff --git a/streamdeck/manager.py b/streamdeck/manager.py index 4eae3f8..7746a8d 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -26,7 +26,6 @@ from typing import Any, Literal from streamdeck.models.events import EventBase - from streamdeck.models.events.base import LiteralStrGenericAlias # TODO: Fix this up to push to a log in the apropos directory and filename. @@ -121,7 +120,7 @@ def _inject_command_sender(self, handler: EventHandlerFunc[TEvent_contra], comma return handler - def _stream_event_data(self) -> Generator[EventBase[LiteralStrGenericAlias], None, None]: + def _stream_event_data(self) -> Generator[EventBase, None, None]: """Stream event data from the event listeners. Validate and model the incoming event data before yielding it. @@ -131,7 +130,7 @@ def _stream_event_data(self) -> Generator[EventBase[LiteralStrGenericAlias], Non """ for message in self._event_listener_manager.event_stream(): try: - data: EventBase[LiteralStrGenericAlias] = self._event_adapter.validate_json(message) + data: EventBase = self._event_adapter.validate_json(message) except ValidationError: logger.exception("Error modeling event data.") continue diff --git a/streamdeck/models/events/__init__.py b/streamdeck/models/events/__init__.py index d5a7fad..be53204 100644 --- a/streamdeck/models/events/__init__.py +++ b/streamdeck/models/events/__init__.py @@ -31,10 +31,8 @@ if TYPE_CHECKING: from typing import Final - from streamdeck.models.events.base import LiteralStrGenericAlias - -DEFAULT_EVENT_MODELS: Final[list[type[EventBase[LiteralStrGenericAlias]]]] = [ +DEFAULT_EVENT_MODELS: Final[list[type[EventBase]]] = [ ApplicationDidLaunch, ApplicationDidTerminate, DeviceDidConnect, diff --git a/streamdeck/models/events/adapter.py b/streamdeck/models/events/adapter.py index 6509f19..bcafb77 100644 --- a/streamdeck/models/events/adapter.py +++ b/streamdeck/models/events/adapter.py @@ -8,14 +8,14 @@ if TYPE_CHECKING: - from streamdeck.models.events.base import EventBase, LiteralStrGenericAlias + 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[LiteralStrGenericAlias]]] = [] - self._type_adapter: TypeAdapter[EventBase[LiteralStrGenericAlias]] | None = 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. @@ -25,7 +25,7 @@ def __init__(self) -> None: for model in DEFAULT_EVENT_MODELS: self.add_model(model) - def add_model(self, model: type[EventBase[LiteralStrGenericAlias]]) -> None: + 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) # Models can have multiple event names defined in the Literal args of the event field, @@ -37,7 +37,7 @@ def event_name_exists(self, event_name: str) -> bool: return event_name in self._event_names @property - def type_adapter(self) -> TypeAdapter[EventBase[LiteralStrGenericAlias]]: + def type_adapter(self) -> TypeAdapter[EventBase]: """Get the TypeAdapter instance for the event models.""" if self._type_adapter is None: self._type_adapter = TypeAdapter( @@ -49,7 +49,7 @@ def type_adapter(self) -> TypeAdapter[EventBase[LiteralStrGenericAlias]]: return self._type_adapter - def validate_json(self, data: str | bytes) -> EventBase[LiteralStrGenericAlias]: + 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 index c0d9dcf..c93cc79 100644 --- a/streamdeck/models/events/application.py +++ b/streamdeck/models/events/application.py @@ -1,7 +1,5 @@ from __future__ import annotations -from typing import Literal - from streamdeck.models.events.base import ConfiguredBaseModel, EventBase @@ -11,13 +9,13 @@ class ApplicationPayload(ConfiguredBaseModel): """Name of the application that triggered the event.""" -class ApplicationDidLaunch(EventBase[Literal["applicationDidLaunch"]]): +class ApplicationDidLaunch(EventBase["applicationDidLaunch"]): """Occurs when a monitored application is launched.""" payload: ApplicationPayload """Payload containing the name of the application that triggered the event.""" -class ApplicationDidTerminate(EventBase[Literal["applicationDidTerminate"]]): +class ApplicationDidTerminate(EventBase["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/deep_link.py b/streamdeck/models/events/deep_link.py index ae018ec..791e6b2 100644 --- a/streamdeck/models/events/deep_link.py +++ b/streamdeck/models/events/deep_link.py @@ -1,5 +1,3 @@ -from typing import Literal - from streamdeck.models.events.base import ConfiguredBaseModel, EventBase @@ -9,7 +7,7 @@ class DeepLinkPayload(ConfiguredBaseModel): """The deep-link URL, with the prefix omitted.""" -class DidReceiveDeepLink(EventBase[Literal["didReceiveDeepLink"]]): +class DidReceiveDeepLink(EventBase["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. diff --git a/streamdeck/models/events/devices.py b/streamdeck/models/events/devices.py index cbd3beb..82b53d5 100644 --- a/streamdeck/models/events/devices.py +++ b/streamdeck/models/events/devices.py @@ -78,11 +78,11 @@ def __repr__(self) -> str: return f"DeviceInfo(name={self.name}, type={self.type}, size={self.size})" -class DeviceDidConnect(EventBase[Literal["deviceDidConnect"]], DeviceSpecificEventMixin): +class DeviceDidConnect(EventBase["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[Literal["deviceDidDisconnect"]], DeviceSpecificEventMixin): +class DeviceDidDisconnect(EventBase["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 8cf5154..1e9ba7d 100644 --- a/streamdeck/models/events/dials.py +++ b/streamdeck/models/events/dials.py @@ -1,7 +1,5 @@ from __future__ import annotations -from typing import Literal - from streamdeck.models.events.base import EventBase from streamdeck.models.events.common import ( BasePayload, @@ -28,19 +26,19 @@ class DialRotatePayload(EncoderPayload): ## Event models for DialDown, DialRotate, and DialUp events -class DialDown(EventBase[Literal["dialDown"]], ContextualEventMixin, DeviceSpecificEventMixin): +class DialDown(EventBase["dialDown"], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user presses a dial (Stream Deck +).""" payload: EncoderPayload """Contextualized information for this event.""" -class DialRotate(EventBase[Literal["dialRotate"]], ContextualEventMixin, DeviceSpecificEventMixin): +class DialRotate(EventBase["dialRotate"], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user rotates a dial (Stream Deck +).""" payload: DialRotatePayload """Contextualized information for this event.""" -class DialUp(EventBase[Literal["dialUp"]], ContextualEventMixin, DeviceSpecificEventMixin): +class DialUp(EventBase["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 0d7f44b..cb368c1 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, Literal +from typing import Annotated from pydantic import Field @@ -48,13 +48,13 @@ class MultiActionKeyGesturePayload( ## Event models for KeyDown and KeyUp events -class KeyDown(EventBase[Literal["keyDown"]], ContextualEventMixin, DeviceSpecificEventMixin): +class KeyDown(EventBase["keyDown"], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user presses a action down.""" payload: CardinalityDiscriminated[SingleActionKeyGesturePayload, MultiActionKeyGesturePayload] """Contextualized information for this event.""" -class KeyUp(EventBase[Literal["keyUp"]], ContextualEventMixin, DeviceSpecificEventMixin): +class KeyUp(EventBase["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 5fa6de2..159cc4b 100644 --- a/streamdeck/models/events/property_inspector.py +++ b/streamdeck/models/events/property_inspector.py @@ -1,10 +1,6 @@ from __future__ import annotations -from typing import Literal - -from streamdeck.models.events.base import ( - EventBase, -) +from streamdeck.models.events.base import EventBase from streamdeck.models.events.common import ( ContextualEventMixin, DeviceSpecificEventMixin, @@ -12,20 +8,20 @@ ) -class DidReceivePropertyInspectorMessage(EventBase[Literal["sendToPlugin"]], ContextualEventMixin): +class DidReceivePropertyInspectorMessage(EventBase["sendToPlugin"], ContextualEventMixin): """Occurs when a message was received from the UI.""" payload: PluginDefinedData """The data payload received from the UI.""" -class PropertyInspectorDidAppear(EventBase[Literal["propertyInspectorDidAppear"]], ContextualEventMixin, DeviceSpecificEventMixin): +class PropertyInspectorDidAppear(EventBase["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[Literal["propertyInspectorDidDisappear"]], ContextualEventMixin, DeviceSpecificEventMixin): +class PropertyInspectorDidDisappear(EventBase["propertyInspectorDidDisappear"], 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. diff --git a/streamdeck/models/events/settings.py b/streamdeck/models/events/settings.py index ec183f8..f4f40cf 100644 --- a/streamdeck/models/events/settings.py +++ b/streamdeck/models/events/settings.py @@ -1,7 +1,5 @@ from __future__ import annotations -from typing import Literal - from streamdeck.models.events.base import ConfiguredBaseModel, EventBase from streamdeck.models.events.common import ( BasePayload, @@ -40,7 +38,7 @@ class MultiActionSettingsPayload( """ -class DidReceiveSettings(EventBase[Literal["didReceiveSettings"]], ContextualEventMixin, DeviceSpecificEventMixin): +class DidReceiveSettings(EventBase["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.""" @@ -54,7 +52,7 @@ class GlobalSettingsPayload(ConfiguredBaseModel): """The global settings received from the Stream Deck.""" -class DidReceiveGlobalSettings(EventBase[Literal["didReceiveGlobalSettings"]]): +class DidReceiveGlobalSettings(EventBase["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 040377d..1d8be50 100644 --- a/streamdeck/models/events/system.py +++ b/streamdeck/models/events/system.py @@ -1,9 +1,7 @@ from __future__ import annotations -from typing import Literal - from streamdeck.models.events.base import EventBase -class SystemDidWakeUp(EventBase[Literal["systemDidWakeUp"]]): +class SystemDidWakeUp(EventBase["systemDidWakeUp"]): """Occurs when the computer wakes up.""" diff --git a/streamdeck/models/events/title_parameters.py b/streamdeck/models/events/title_parameters.py index 97f9e0f..538fb8f 100644 --- a/streamdeck/models/events/title_parameters.py +++ b/streamdeck/models/events/title_parameters.py @@ -47,7 +47,7 @@ class TitleParametersDidChangePayload( """Defines aesthetic properties that determine how the title should be rendered.""" -class TitleParametersDidChange(EventBase[Literal["titleParametersDidChange"]], DeviceSpecificEventMixin): +class TitleParametersDidChange(EventBase["titleParametersDidChange"], DeviceSpecificEventMixin): """Occurs when the user updates an action's title settings in the Stream Deck application.""" context: str """Identifies the instance of an action that caused the event, i.e. the specific key or dial.""" diff --git a/streamdeck/models/events/touch_tap.py b/streamdeck/models/events/touch_tap.py index 1b1402d..9a14688 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, Literal +from typing import Annotated from pydantic import Field @@ -22,7 +22,7 @@ class TouchTapPayload(BasePayload[EncoderControllerType], CoordinatesPayloadMixi """Coordinates of where the touchscreen tap occurred, relative to the canvas of the action.""" -class TouchTap(EventBase[Literal["touchTap"]], ContextualEventMixin, DeviceSpecificEventMixin): +class TouchTap(EventBase["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 4f50533..43bd871 100644 --- a/streamdeck/models/events/visibility.py +++ b/streamdeck/models/events/visibility.py @@ -1,7 +1,5 @@ from __future__ import annotations -from typing import Literal - from streamdeck.models.events.base import EventBase from streamdeck.models.events.common import ( BasePayload, @@ -35,7 +33,7 @@ class MultiActionVisibilityPayload( ## Event models for WillAppear and WillDisappear events -class WillAppear(EventBase[Literal["willAppear"]], ContextualEventMixin, DeviceSpecificEventMixin): +class WillAppear(EventBase["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". @@ -44,7 +42,7 @@ class WillAppear(EventBase[Literal["willAppear"]], ContextualEventMixin, DeviceS payload: CardinalityDiscriminated[SingleActionVisibilityPayload, MultiActionVisibilityPayload] -class WillDisappear(EventBase[Literal["willDisappear"]], ContextualEventMixin, DeviceSpecificEventMixin): +class WillDisappear(EventBase["willDisappear"], 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. diff --git a/tests/actions/test_action_event_handler_filtering.py b/tests/actions/test_action_event_handler_filtering.py index 9941a9b..efd7d17 100644 --- a/tests/actions/test_action_event_handler_filtering.py +++ b/tests/actions/test_action_event_handler_filtering.py @@ -19,8 +19,6 @@ from polyfactory.factories.pydantic_factory import ModelFactory from streamdeck.models import events - from streamdeck.models.events.base import LiteralStrGenericAlias - @pytest.fixture @@ -36,14 +34,14 @@ def dummy_handler(event: events.EventBase) -> None: DeviceDidConnectFactory, ApplicationDidLaunchEventFactory ]) -def fake_event_data(request: pytest.FixtureRequest) -> events.EventBase[LiteralStrGenericAlias]: - event_factory = cast("ModelFactory[events.EventBase[LiteralStrGenericAlias]]", request.param) +def fake_event_data(request: pytest.FixtureRequest) -> events.EventBase: + event_factory = cast("ModelFactory[events.EventBase]", request.param) return event_factory.build() def test_global_action_gets_triggered_by_event( mock_event_handler: Mock, - fake_event_data: events.EventBase[LiteralStrGenericAlias], + fake_event_data: events.EventBase, ) -> None: """Test that a global action's event handlers are triggered by an event. @@ -63,7 +61,7 @@ def test_global_action_gets_triggered_by_event( def test_action_gets_triggered_by_event( mock_event_handler: Mock, - fake_event_data: events.EventBase[LiteralStrGenericAlias], + fake_event_data: events.EventBase, ) -> None: """Test that an action's event handlers are triggered by an event. @@ -90,7 +88,7 @@ def test_action_gets_triggered_by_event( def test_global_action_registry_get_action_handlers_filtering( mock_event_handler: Mock, - fake_event_data: events.EventBase[LiteralStrGenericAlias], + 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, ContextualEventMixin) else None @@ -117,7 +115,7 @@ def test_global_action_registry_get_action_handlers_filtering( def test_action_registry_get_action_handlers_filtering( mock_event_handler: Mock, - fake_event_data: events.EventBase[LiteralStrGenericAlias], + 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, ContextualEventMixin) else None @@ -151,12 +149,12 @@ def test_multiple_actions_filtering() -> None: action_event_handler_called = False @global_action.on("applicationDidLaunch") - def _global_app_did_launch_action_handler(event: events.EventBase[LiteralStrGenericAlias]) -> None: + def _global_app_did_launch_action_handler(event: events.EventBase) -> None: nonlocal global_action_event_handler_called global_action_event_handler_called = True @action.on("keyDown") - def _action_key_down_event_handler(event: events.EventBase[LiteralStrGenericAlias]) -> None: + def _action_key_down_event_handler(event: events.EventBase) -> None: nonlocal action_event_handler_called action_event_handler_called = True diff --git a/tests/actions/test_action_registry.py b/tests/actions/test_action_registry.py index abf9a3a..34cf396 100644 --- a/tests/actions/test_action_registry.py +++ b/tests/actions/test_action_registry.py @@ -14,7 +14,6 @@ if TYPE_CHECKING: from streamdeck.models import events - from streamdeck.models.events.base import LiteralStrGenericAlias def test_register_action() -> None: @@ -47,7 +46,7 @@ def test_get_action_handlers_with_handlers() -> None: action = Action("my-fake-action-uuid") @action.on("dialDown") - def dial_down_handler(event_data: events.EventBase[LiteralStrGenericAlias]) -> None: + def dial_down_handler(event_data: events.EventBase) -> None: pass registry.register(action) @@ -67,11 +66,11 @@ def test_get_action_handlers_multiple_actions() -> None: action2 = Action("fake-action-uuid-2") @action1.on("keyUp") - def key_up_handler1(event_data: events.EventBase[LiteralStrGenericAlias]) -> None: + def key_up_handler1(event_data: events.EventBase) -> None: pass @action2.on("keyUp") - def key_up_handler2(event_data: events.EventBase[LiteralStrGenericAlias]) -> None: + def key_up_handler2(event_data: events.EventBase) -> None: pass registry.register(action1) diff --git a/tests/actions/test_actions.py b/tests/actions/test_actions.py index 5ce8a20..12f77c8 100644 --- a/tests/actions/test_actions.py +++ b/tests/actions/test_actions.py @@ -9,7 +9,6 @@ if TYPE_CHECKING: from streamdeck.models.events import EventBase - from streamdeck.models.events.base import LiteralStrGenericAlias @pytest.fixture(params=[[Action, ("test.uuid.for.action",)], [GlobalAction, ()]]) @@ -26,7 +25,7 @@ def action(request: pytest.FixtureRequest) -> ActionBase: def test_action_register_event_handler(action: ActionBase, event_name: str) -> None: """Test that an event handler can be registered for each valid event name.""" @action.on(event_name) - def handler(event_data: EventBase[LiteralStrGenericAlias]) -> None: + def handler(event_data: EventBase) -> None: pass # Ensure the handler is registered for the correct event name @@ -41,7 +40,7 @@ def test_action_get_event_handlers(action: ActionBase) -> None: for _, event_name in enumerate(DEFAULT_EVENT_NAMES): # Register a handler for the given event name @action.on(event_name) - def handler(event_data: EventBase[LiteralStrGenericAlias]) -> None: + def handler(event_data: EventBase) -> None: pass # Retrieve the handlers using the generator @@ -74,11 +73,11 @@ def test_action_get_event_handlers_invalid_event_name(action: ActionBase) -> Non def test_action_register_multiple_handlers_for_event(action: ActionBase) -> None: """Test that multiple handlers can be registered for the same event on the same action.""" @action.on("keyDown") - def handler_one(event_data: EventBase[LiteralStrGenericAlias]) -> None: + def handler_one(event_data: EventBase) -> None: pass @action.on("keyDown") - def handler_two(event_data: EventBase[LiteralStrGenericAlias]) -> None: + def handler_two(event_data: EventBase) -> None: pass handlers = list(action.get_event_handlers("keyDown")) diff --git a/tests/event_listener/test_event_listener.py b/tests/event_listener/test_event_listener.py index d2bf1b9..ea49587 100644 --- a/tests/event_listener/test_event_listener.py +++ b/tests/event_listener/test_event_listener.py @@ -7,12 +7,11 @@ import pytest from streamdeck.event_listener import EventListener, EventListenerManager from streamdeck.models.events import ApplicationDidLaunch, EventBase -from streamdeck.models.events.base import LiteralStrGenericAlias class MockEventListener(EventListener): """Mock implementation of EventListener for testing.""" - event_models: ClassVar[list[type[EventBase[LiteralStrGenericAlias]]]] = [ApplicationDidLaunch] + event_models: ClassVar[list[type[EventBase]]] = [ApplicationDidLaunch] def __init__(self): self._running = True @@ -34,7 +33,7 @@ def stop(self) -> None: class SlowMockEventListener(EventListener): """Mock implementation of EventListener that yields events with a delay.""" - event_models: ClassVar[list[type[EventBase[LiteralStrGenericAlias]]]] = [ApplicationDidLaunch] + event_models: ClassVar[list[type[EventBase]]] = [ApplicationDidLaunch] def __init__(self, delay: float = 0.1): self._running = True @@ -54,7 +53,7 @@ def stop(self) -> None: class ExceptionEventListener(EventListener): """Mock implementation of EventListener that raises an exception.""" - event_models: ClassVar[list[type[EventBase[LiteralStrGenericAlias]]]] = [ApplicationDidLaunch] + event_models: ClassVar[list[type[EventBase]]] = [ApplicationDidLaunch] def listen(self) -> Generator[str, None, None]: self._running = True diff --git a/tests/models/test_event_adapter.py b/tests/models/test_event_adapter.py index a644e07..a4a51ae 100644 --- a/tests/models/test_event_adapter.py +++ b/tests/models/test_event_adapter.py @@ -30,8 +30,7 @@ def test_add_model() -> None: adapter = EventAdapter() # Create a fake event model - class DummyEvent(EventBase): - event: Literal["dummyEvent"] + class DummyEvent(EventBase["dummyEvent"]): payload: dict[str, str] # Add the custom model @@ -104,8 +103,7 @@ def test_adding_custom_event_allows_validation() -> None: adapter = EventAdapter() # Create a fake event model - class DummyEvent(EventBase): - event: Literal["dummyEvent"] + class DummyEvent(EventBase["dummyEvent"]): action: str context: str device: str diff --git a/tests/plugin_manager/test_command_sender_binding.py b/tests/plugin_manager/test_command_sender_binding.py index b137c63..46404a4 100644 --- a/tests/plugin_manager/test_command_sender_binding.py +++ b/tests/plugin_manager/test_command_sender_binding.py @@ -14,7 +14,6 @@ from streamdeck.command_sender import StreamDeckCommandSender from streamdeck.manager import PluginManager from streamdeck.models import events - from streamdeck.models.events.base import LiteralStrGenericAlias from streamdeck.types import ( EventHandlerBasicFunc, EventHandlerFunc, @@ -32,12 +31,12 @@ def create_event_handler(include_command_sender_param: bool = False) -> EventHan Callable[[events.EventBase], None] | Callable[[events.EventBase, StreamDeckCommandSender], None]: A dummy event handler function. """ if not include_command_sender_param: - def dummy_handler_without_cmd_sender(event_data: events.EventBase[LiteralStrGenericAlias]) -> None: + def dummy_handler_without_cmd_sender(event_data: events.EventBase) -> None: """Dummy event handler function that matches the EventHandlerFunc TypeAlias without `command_sender` param.""" return dummy_handler_without_cmd_sender - def dummy_handler_with_cmd_sender(event_data: events.EventBase[LiteralStrGenericAlias], command_sender: StreamDeckCommandSender) -> None: + def dummy_handler_with_cmd_sender(event_data: events.EventBase, command_sender: StreamDeckCommandSender) -> None: """Dummy event handler function that matches the EventHandlerFunc TypeAlias with `command_sender` param.""" return dummy_handler_with_cmd_sender @@ -46,7 +45,7 @@ def dummy_handler_with_cmd_sender(event_data: events.EventBase[LiteralStrGeneric @pytest.fixture(params=[True, False]) def mock_event_handler(request: pytest.FixtureRequest) -> Mock: include_command_sender_param: bool = request.param - dummy_handler: EventHandlerFunc[events.EventBase[LiteralStrGenericAlias]] = create_event_handler(include_command_sender_param) + dummy_handler: EventHandlerFunc[events.EventBase] = create_event_handler(include_command_sender_param) return create_autospec(dummy_handler, spec_set=True) @@ -58,7 +57,7 @@ def test_inject_command_sender_func( ) -> None: """Test that the command_sender is injected into the handler.""" mock_command_sender = Mock() - result_handler: EventHandlerBasicFunc[events.EventBase[LiteralStrGenericAlias]] = plugin_manager._inject_command_sender(mock_event_handler, mock_command_sender) + result_handler: EventHandlerFunc [events.EventBase]= plugin_manager._inject_command_sender(mock_event_handler, mock_command_sender) resulting_handler_params = inspect.signature(result_handler).parameters diff --git a/tests/plugin_manager/test_plugin_manager.py b/tests/plugin_manager/test_plugin_manager.py index d152fa0..b471539 100644 --- a/tests/plugin_manager/test_plugin_manager.py +++ b/tests/plugin_manager/test_plugin_manager.py @@ -10,7 +10,6 @@ DEFAULT_EVENT_NAMES, EventBase, ) -from streamdeck.models.events.base import LiteralStrGenericAlias @pytest.fixture @@ -116,7 +115,7 @@ def test_plugin_manager_adds_websocket_event_listener( @pytest.mark.integration @pytest.mark.usefixtures("patch_websocket_client") def test_plugin_manager_process_event( - mock_event_listener_manager_with_fake_events: tuple[Mock, list[EventBase[LiteralStrGenericAlias]]], + mock_event_listener_manager_with_fake_events: tuple[Mock, list[EventBase]], _spy_action_registry_get_action_handlers: None, # This fixture must come after mock_event_listener_manager_with_fake_events to ensure monkeypatching occurs. _spy_event_adapter_validate_json: None, # This fixture must come after mock_event_listener_manager_with_fake_events to ensure monkeypatching occurs. plugin_manager: PluginManager, # This fixture must come after patch_event_listener_manager and spy-fixtures to ensure things are mocked and spied correctly. From 35c0bdc133232d9c0e25d4ad2c356ee8a3826756 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Mon, 21 Apr 2025 17:38:18 -0600 Subject: [PATCH 3/6] Update EventHandlerFunc protocols. Delete streamdeck/types.py --- streamdeck/__init__.py | 2 - streamdeck/actions.py | 36 +++++++--- streamdeck/manager.py | 44 +++++++++++-- streamdeck/types.py | 66 ------------------- tests/data/dummy_event_listener.py | 3 +- tests/models/events/__init__.py | 0 .../test_command_sender_binding.py | 13 ++-- 7 files changed, 69 insertions(+), 95 deletions(-) delete mode 100644 streamdeck/types.py create mode 100644 tests/models/events/__init__.py diff --git a/streamdeck/__init__.py b/streamdeck/__init__.py index bac83ae..d0f3bf8 100644 --- a/streamdeck/__init__.py +++ b/streamdeck/__init__.py @@ -3,7 +3,6 @@ command_sender, manager, models, - types, utils, websocket, ) @@ -14,7 +13,6 @@ "command_sender", "manager", "models", - "types", "utils", "websocket", ] diff --git a/streamdeck/actions.py b/streamdeck/actions.py index 8c67f36..4b18d03 100644 --- a/streamdeck/actions.py +++ b/streamdeck/actions.py @@ -9,10 +9,27 @@ if TYPE_CHECKING: from collections.abc import Callable, Generator + from typing import Protocol + + from typing_extensions import ParamSpec, TypeAlias, TypeVar # noqa: UP035 from streamdeck.models.events import EventBase - from streamdeck.models.events.base import LiteralStrGenericAlias - from streamdeck.types import BaseEventHandlerFunc, EventHandlerFunc, EventNameStr, TEvent_contra + + + EventNameStr: TypeAlias = str # noqa: UP040 + """Type alias for the event name string. + + We don't define literal string values here, as the list of available event names can be added to dynamically. + """ + + + EventModel_contra = TypeVar("EventModel_contra", bound=EventBase, default=EventBase, contravariant=True) + InjectableParams = ParamSpec("InjectableParams", default=...) + + class EventHandlerFunc(Protocol[EventModel_contra, InjectableParams]): + """Protocol for an event handler function that takes an event (of subtype of EventBase) and other parameters that are injectable.""" + def __call__(self, event_data: EventModel_contra, *args: InjectableParams.args, **kwargs: InjectableParams.kwargs) -> None: ... + logger = getLogger("streamdeck.actions") @@ -27,9 +44,9 @@ def __init__(self) -> None: Args: uuid (str): The unique identifier for the action. """ - self._events: dict[EventNameStr, set[BaseEventHandlerFunc]] = defaultdict(set) + self._events: dict[EventNameStr, set[EventHandlerFunc]] = defaultdict(set) - def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[TEvent_contra] | BaseEventHandlerFunc], EventHandlerFunc[TEvent_contra] | BaseEventHandlerFunc]: + def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[EventModel_contra, InjectableParams]], EventHandlerFunc[EventModel_contra, InjectableParams]]: """Register an event handler for a specific event. Args: @@ -41,15 +58,14 @@ def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[TEvent_c Raises: KeyError: If the provided event name is not available. """ - def _wrapper(func: EventHandlerFunc[TEvent_contra]) -> EventHandlerFunc[TEvent_contra]: - # Cast to BaseEventHandlerFunc so that the storage type is consistent. - self._events[event_name].add(cast("BaseEventHandlerFunc", func)) - + def _wrapper(func: EventHandlerFunc[EventModel_contra, InjectableParams]) -> EventHandlerFunc[EventModel_contra, InjectableParams]: + # Cast to EventHandlerFunc with default type arguments so that the storage type is consistent. + self._events[event_name].add(cast("EventHandlerFunc", func)) return func return _wrapper - def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc[EventBase], None, None]: + def get_event_handlers(self, event_name: EventNameStr, /) -> Generator[EventHandlerFunc, None, None]: """Get all event handlers for a specific event. Args: @@ -112,7 +128,7 @@ def register(self, action: ActionBase) -> None: """ self._plugin_actions.append(action) - def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | None = None) -> Generator[EventHandlerFunc[EventBase], None, None]: + def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | None = None) -> Generator[EventHandlerFunc, None, None]: """Get all event handlers for a specific event from all registered actions. Args: diff --git a/streamdeck/manager.py b/streamdeck/manager.py index 7746a8d..a99b606 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -1,22 +1,18 @@ from __future__ import annotations import functools +import inspect from logging import getLogger from typing import TYPE_CHECKING from pydantic import ValidationError +from typing_extensions import TypeGuard # noqa: UP035 from streamdeck.actions import Action, ActionBase, ActionRegistry from streamdeck.command_sender import StreamDeckCommandSender from streamdeck.event_listener import EventListener, EventListenerManager from streamdeck.models.events.adapter import EventAdapter from streamdeck.models.events.common import ContextualEventMixin -from streamdeck.types import ( - EventHandlerBasicFunc, - EventHandlerFunc, - TEvent_contra, - is_bindable_handler, -) from streamdeck.utils.logging import configure_streamdeck_logger from streamdeck.websocket import WebSocketClient @@ -25,13 +21,42 @@ from collections.abc import Generator from typing import Any, Literal + from streamdeck.actions import ( + EventHandlerFunc, + EventModel_contra, + InjectableParams, + ) from streamdeck.models.events import EventBase + BindableEventHandlerFunc = EventHandlerFunc[EventModel_contra, [StreamDeckCommandSender]] + """Type alias for a bindable event handler function that takes an event (of subtype of EventBase) and a command_sender parameter that is to be injected.""" + BoundEventHandlerFunc = EventHandlerFunc[EventModel_contra, []] + """Type alias for a bound event handler function that takes an event (of subtype of EventBase) and no other parameters. + + Typically used for event handlers that have already had parameters injected. + """ + + + # TODO: Fix this up to push to a log in the apropos directory and filename. logger = getLogger("streamdeck.manager") +def is_bindable_handler(handler: EventHandlerFunc[EventModel_contra, InjectableParams]) -> TypeGuard[BindableEventHandlerFunc[EventModel_contra]]: + """Check if the handler is prebound with the `command_sender` parameter.""" + # Check dynamically if the `command_sender`'s name is in the handler's arguments. + return "command_sender" in inspect.signature(handler).parameters + + +def is_not_bindable_handler(handler: EventHandlerFunc[EventModel_contra, InjectableParams]) -> TypeGuard[BoundEventHandlerFunc[EventModel_contra]]: + """Check if the handler only accepts the event_data parameter. + + If this function returns False after the is_bindable_handler check is True, then the function has invalid parameters, and will subsequently need to be handled in the calling code. + """ + handler_params = inspect.signature(handler).parameters + return len(handler_params) == 1 and "event_data" in handler_params + class PluginManager: """Manages plugin actions and communicates with a WebSocket server to handle events.""" @@ -105,7 +130,7 @@ def register_event_listener(self, listener: EventListener) -> None: for event_model in listener.event_models: self._event_adapter.add_model(event_model) - def _inject_command_sender(self, handler: EventHandlerFunc[TEvent_contra], command_sender: StreamDeckCommandSender) -> EventHandlerBasicFunc[TEvent_contra]: + def _inject_command_sender(self, handler: EventHandlerFunc[EventModel_contra, InjectableParams], command_sender: StreamDeckCommandSender) -> BoundEventHandlerFunc[EventModel_contra]: """Inject command_sender into handler if it accepts it as a parameter. Args: @@ -116,8 +141,13 @@ def _inject_command_sender(self, handler: EventHandlerFunc[TEvent_contra], comma The handler with command_sender injected if needed """ if is_bindable_handler(handler): + # If the handler accepts command_sender, inject it and return the handler. return functools.partial(handler, command_sender=command_sender) + if not is_not_bindable_handler(handler): + # If the handler is neither bindable nor not bindable, raise an error. + raise TypeError(f"Invalid event handler function signature: {handler}") # noqa: TRY003, EM102 + return handler def _stream_event_data(self) -> Generator[EventBase, None, None]: diff --git a/streamdeck/types.py b/streamdeck/types.py deleted file mode 100644 index d8e37dc..0000000 --- a/streamdeck/types.py +++ /dev/null @@ -1,66 +0,0 @@ -from __future__ import annotations - -import inspect -from typing import TYPE_CHECKING, Protocol, TypeVar, Union - -from streamdeck.models.events import EventBase - - -if TYPE_CHECKING: - from typing_extensions import TypeAlias, TypeIs # noqa: UP035 - - from streamdeck.command_sender import StreamDeckCommandSender - - - -EventNameStr: TypeAlias = str # noqa: UP040 -"""Type alias for the event name string. - -We don't define literal string values here, as the list of available event names can be added to dynamically. -""" - - -### Event Handler Type Definitions ### - -## Protocols for event handler functions that act on subtypes of EventBase instances in a Generic way. - -TEvent_contra = TypeVar("TEvent_contra", bound=EventBase, contravariant=True) -"""Type variable for a subtype of EventBase.""" - - -class EventHandlerBasicFunc(Protocol[TEvent_contra]): - """Protocol for a basic event handler function that takes just an event (of subtype of EventBase).""" - def __call__(self, event_data: TEvent_contra) -> None: ... - - -class EventHandlerBindableFunc(Protocol[TEvent_contra]): - """Protocol for an event handler function that takes an event (of subtype of EventBase) and a command sender.""" - def __call__(self, event_data: TEvent_contra, command_sender: StreamDeckCommandSender) -> None: ... - - -EventHandlerFunc = Union[EventHandlerBasicFunc[TEvent_contra], EventHandlerBindableFunc[TEvent_contra]] # noqa: UP007 -"""Type alias for an event handler function that takes an event (of subtype of EventBase), and optionally a command sender.""" - - -## Protocols for event handler functions that act on EventBase instances. - -class BaseEventHandlerBasicFunc(EventHandlerBasicFunc[EventBase]): - """Protocol for a basic event handler function that takes just an EventBase.""" - -class BaseEventHandlerBindableFunc(EventHandlerBindableFunc[EventBase]): - """Protocol for an event handler function that takes an event (of subtype of EventBase) and a command sender.""" - - -BaseEventHandlerFunc = Union[BaseEventHandlerBasicFunc, BaseEventHandlerBindableFunc] # noqa: UP007 -"""Type alias for a base event handler function that takes an actual EventBase instance argument, and optionally a command sender. - -This is used for type hinting internal storage of event handlers. -""" - - - -# def is_bindable_handler(handler: EventHandlerFunc[TEvent_contra] | BaseEventHandlerFunc) -> TypeIs[EventHandlerBindableFunc[TEvent_contra] | BaseEventHandlerBindableFunc]: -def is_bindable_handler(handler: EventHandlerBasicFunc[TEvent_contra] | EventHandlerBindableFunc[TEvent_contra]) -> TypeIs[EventHandlerBindableFunc[TEvent_contra]]: - """Check if the handler is prebound with the `command_sender` parameter.""" - # Check dynamically if the `command_sender`'s name is in the handler's arguments. - return "command_sender" in inspect.signature(handler).parameters diff --git a/tests/data/dummy_event_listener.py b/tests/data/dummy_event_listener.py index 7bdc3da..092cead 100644 --- a/tests/data/dummy_event_listener.py +++ b/tests/data/dummy_event_listener.py @@ -12,9 +12,8 @@ from typing import Any, ClassVar -class DummyEvent(EventBase): +class DummyEvent(EventBase["dummy"]): """A dummy event for testing purposes.""" - event: Literal["dummy"] # type: ignore[assignment] something: int diff --git a/tests/models/events/__init__.py b/tests/models/events/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/plugin_manager/test_command_sender_binding.py b/tests/plugin_manager/test_command_sender_binding.py index 46404a4..98a25f2 100644 --- a/tests/plugin_manager/test_command_sender_binding.py +++ b/tests/plugin_manager/test_command_sender_binding.py @@ -11,17 +11,14 @@ if TYPE_CHECKING: from functools import partial + from streamdeck.actions import EventHandlerFunc from streamdeck.command_sender import StreamDeckCommandSender from streamdeck.manager import PluginManager from streamdeck.models import events - from streamdeck.types import ( - EventHandlerBasicFunc, - EventHandlerFunc, - ) -def create_event_handler(include_command_sender_param: bool = False) -> EventHandlerFunc[events.EventBase]: +def create_event_handler(include_command_sender_param: bool = False) -> EventHandlerFunc: """Create a dummy event handler function that matches the EventHandlerFunc TypeAlias. Args: @@ -45,7 +42,7 @@ def dummy_handler_with_cmd_sender(event_data: events.EventBase, command_sender: @pytest.fixture(params=[True, False]) def mock_event_handler(request: pytest.FixtureRequest) -> Mock: include_command_sender_param: bool = request.param - dummy_handler: EventHandlerFunc[events.EventBase] = create_event_handler(include_command_sender_param) + dummy_handler: EventHandlerFunc = create_event_handler(include_command_sender_param) return create_autospec(dummy_handler, spec_set=True) @@ -57,7 +54,7 @@ def test_inject_command_sender_func( ) -> None: """Test that the command_sender is injected into the handler.""" mock_command_sender = Mock() - result_handler: EventHandlerFunc [events.EventBase]= plugin_manager._inject_command_sender(mock_event_handler, mock_command_sender) + result_handler: EventHandlerFunc = plugin_manager._inject_command_sender(mock_event_handler, mock_command_sender) resulting_handler_params = inspect.signature(result_handler).parameters @@ -78,7 +75,7 @@ def test_inject_command_sender_func( @pytest.mark.usefixtures("patch_websocket_client") def test_run_manager_events_handled_with_correct_params( - mock_event_listener_manager_with_fake_events: tuple[Mock, list[events.EventBase[LiteralStrGenericAlias]]], + mock_event_listener_manager_with_fake_events: tuple[Mock, list[events.EventBase]], plugin_manager: PluginManager, mock_command_sender: Mock, ) -> None: From 3a69a87092f5d1b71b0e0a91024f405c1da0d809 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Tue, 22 Apr 2025 21:38:10 -0600 Subject: [PATCH 4/6] Clean up EventBase internals --- streamdeck/models/events/base.py | 96 ++++++++++++++++++++++++++------ 1 file changed, 79 insertions(+), 17 deletions(-) diff --git a/streamdeck/models/events/base.py b/streamdeck/models/events/base.py index 642056f..195f3b6 100644 --- a/streamdeck/models/events/base.py +++ b/streamdeck/models/events/base.py @@ -1,10 +1,16 @@ from __future__ import annotations from abc import ABC -from typing import TYPE_CHECKING, Any, ClassVar, Literal +from typing import TYPE_CHECKING, Literal +from weakref import WeakValueDictionary from pydantic import BaseModel, ConfigDict, create_model -from typing_extensions import LiteralString, override # noqa: UP035 +from typing_extensions import LiteralString, TypedDict, override # noqa: UP035 + + +if TYPE_CHECKING: + from typing import Any, ClassVar + class ConfiguredBaseModel(BaseModel, ABC): @@ -29,11 +35,24 @@ def model_dump_json(self, **kwargs: Any) -> str: return super().model_dump_json(**kwargs) +class EventMetadataDict(TypedDict): + """Metadata for specialized EventBase submodels. + + Similar to the __pydantic_generic_metadata__ attribute, but for use in the EventBase class——which isn't actually generic. + """ + 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 # type: ignore[import] class EventMeta(ModelMetaclass): @@ -42,12 +61,32 @@ class EventMeta(ModelMetaclass): def __getitem__(cls, event_names: LiteralString | tuple[LiteralString, ...]) -> type[EventBase]: ... class EventBase(BaseModel, metaclass=EventMeta): - """Base class for all event models.""" + """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"]` + ... + + ``` + """ event: LiteralString """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[str, ...]: @@ -56,9 +95,27 @@ def get_model_event_names(cls) -> tuple[str, ...]: else: class EventBase(ConfiguredBaseModel, ABC): - """Base class for all event models.""" - _subtypes: ClassVar[dict[str, type[EventBase]]] = {} - __args__: ClassVar[tuple[str, ...]] + """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. @@ -106,25 +163,30 @@ def __class_getitem__(cls: type[EventBase], event_names: LiteralString | tuple[L 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 _subtypes dictionary, otherwise return the existing subclass. + 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._subtypes: - # Pass in the _is_base_subtype kwarg to __init_subclass__ to indicate that this is a base subtype of EventBase, and should be ignored. - cls._subtypes[new_name] = create_model( + 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, - __args__=(tuple[str, ...], event_name_args), + __event_metadata__=(EventMetadataDict, {"origin": cls, "args": event_name_args}), event=(Literal[event_name_args], ...), - __cls_kwargs__={"_is_base_subtype": True}, + __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._subtypes[new_name] + return cls._cached_specialized_submodels[new_name] @classmethod - def __init_subclass__(cls, _is_base_subtype: bool = False) -> None: - """Validate a child class of EventBase (not a subscripted base subclass) is subclassing from a subscripted EventBase.""" - if _is_base_subtype: + 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 @@ -147,4 +209,4 @@ def __event_type__(cls) -> type[object] | None: @classmethod def get_model_event_names(cls) -> tuple[str, ...]: """Return the event names for the event model.""" - return cls.__args__ + return cls.__event_metadata__["args"] From 9bd84a0162e50e7482cf3d595efd6ac619f57f65 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Fri, 25 Apr 2025 12:38:55 -0600 Subject: [PATCH 5/6] Update README.md documentation to reflect change in how Event models are defined --- README.md | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index a3d3b8e..20ac607 100644 --- a/README.md +++ b/README.md @@ -292,7 +292,7 @@ The SDK allows you to create custom event listeners and events by extending the To create a custom event listener: -1. Create new event model that inherits from `EventBase`. +1. Create new event model that inherits from `EventBase["eventName"]`. 2. Create a new class that inherits from `EventListener`. a. Implement the required `listen` and `stop` methods. The `listen` method should yield results as a json string that matches the new event model. b. List the new event classes in the `event_models` class variable of the new `EventListener` class. @@ -307,11 +307,15 @@ from streamdeck.event_listener import EventListener from streamdeck.models.events import EventBase -class MyCustomEvent(EventBase): - event: Literal["somethingHappened"] - ... # Define additional data attributes here +class MyCustomEvent(EventBase["somethingHappened"]): + # The 'event' field's type annotation is internally set as Literal["somethingHappened"] + # Define additional data attributes here + result: str + class MyCustomEventListener(EventListener): + event_models = [MyCustomEvent] + def listen(self) -> Generator[str | bytes, None, None]: ... # Listen/poll for something here in a loop, and yield the result. @@ -320,7 +324,7 @@ class MyCustomEventListener(EventListener): # while self._running is True: # result = module.check_status() # if result is not None: - # yield json.dumps({"event": "somethingHappend", "result": result}) + # yield json.dumps({"event": "somethingHappened", "result": result}) # time.sleep(1) def stop(self) -> None: @@ -344,7 +348,7 @@ To use your custom event listener, add it to your `pyproject.toml` file: ] ``` -The `event_listeners` list should contain strings in module format for each module you want to use. +The `event_listener_modules` list should contain strings in module format for each module you want to use. ## Creating and Packaging Plugins From 1d98b3b628859bb00496d1015a9158c0f9d99c47 Mon Sep 17 00:00:00 2001 From: strohganoff Date: Fri, 25 Apr 2025 13:12:00 -0600 Subject: [PATCH 6/6] Consolidate string aliases to streamdeck.types --- streamdeck/actions.py | 19 ++--- streamdeck/command_sender.py | 73 ++++++++++++------- streamdeck/models/events/__init__.py | 9 ++- streamdeck/models/events/adapter.py | 7 +- streamdeck/models/events/base.py | 6 +- streamdeck/models/events/common.py | 18 +++-- .../models/events/property_inspector.py | 2 +- streamdeck/models/events/settings.py | 2 +- streamdeck/models/events/title_parameters.py | 5 +- streamdeck/types.py | 33 +++++++++ tests/plugin_manager/conftest.py | 3 +- 11 files changed, 114 insertions(+), 63 deletions(-) create mode 100644 streamdeck/types.py diff --git a/streamdeck/actions.py b/streamdeck/actions.py index 4b18d03..757261d 100644 --- a/streamdeck/actions.py +++ b/streamdeck/actions.py @@ -11,17 +11,12 @@ from collections.abc import Callable, Generator from typing import Protocol - from typing_extensions import ParamSpec, TypeAlias, TypeVar # noqa: UP035 + from typing_extensions import ParamSpec, TypeVar from streamdeck.models.events import EventBase + from streamdeck.types import ActionUUIDStr, EventNameStr - EventNameStr: TypeAlias = str # noqa: UP040 - """Type alias for the event name string. - - We don't define literal string values here, as the list of available event names can be added to dynamically. - """ - EventModel_contra = TypeVar("EventModel_contra", bound=EventBase, default=EventBase, contravariant=True) InjectableParams = ParamSpec("InjectableParams", default=...) @@ -39,11 +34,7 @@ class ActionBase(ABC): """Base class for all actions.""" def __init__(self) -> None: - """Initialize an Action instance. - - Args: - uuid (str): The unique identifier for the action. - """ + """Initialize an Action instance.""" self._events: dict[EventNameStr, set[EventHandlerFunc]] = defaultdict(set) def on(self, event_name: EventNameStr, /) -> Callable[[EventHandlerFunc[EventModel_contra, InjectableParams]], EventHandlerFunc[EventModel_contra, InjectableParams]]: @@ -98,7 +89,7 @@ class GlobalAction(ActionBase): class Action(ActionBase): """Represents an action that can be performed for a specific action, with event handlers for specific event types.""" - def __init__(self, uuid: str) -> None: + def __init__(self, uuid: ActionUUIDStr) -> None: """Initialize an Action instance. Args: @@ -128,7 +119,7 @@ def register(self, action: ActionBase) -> None: """ self._plugin_actions.append(action) - def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: str | None = None) -> Generator[EventHandlerFunc, None, None]: + def get_action_handlers(self, event_name: EventNameStr, event_action_uuid: ActionUUIDStr | None = None) -> Generator[EventHandlerFunc, None, None]: """Get all event handlers for a specific event from all registered actions. Args: diff --git a/streamdeck/command_sender.py b/streamdeck/command_sender.py index 9afa230..ed1b531 100644 --- a/streamdeck/command_sender.py +++ b/streamdeck/command_sender.py @@ -7,6 +7,13 @@ if TYPE_CHECKING: from typing import Any, Literal + from streamdeck.types import ( + ActionInstanceUUIDStr, + ActionUUIDStr, + DeviceUUIDStr, + EventNameStr, + PluginDefinedData, + ) from streamdeck.websocket import WebSocketClient @@ -16,30 +23,31 @@ class StreamDeckCommandSender: """Class for sending command event messages to the Stream Deck software through a WebSocket client.""" + def __init__(self, client: WebSocketClient, plugin_registration_uuid: str): self._client = client self._plugin_registration_uuid = plugin_registration_uuid - def _send_event(self, event: str, **kwargs: Any) -> None: + def _send_event(self, event: EventNameStr, **kwargs: Any) -> None: self._client.send_event({ "event": event, **kwargs, }) - def set_settings(self, context: str, payload: dict[str, Any]) -> None: + def set_settings(self, context: ActionInstanceUUIDStr, payload: PluginDefinedData) -> None: self._send_event( event="setSettings", context=context, payload=payload, ) - def get_settings(self, context: str) -> None: + def get_settings(self, context: ActionInstanceUUIDStr) -> None: self._send_event( event="getSettings", context=context, ) - def set_global_settings(self, payload: dict[str, Any]) -> None: + def set_global_settings(self, payload: PluginDefinedData) -> None: self._send_event( event="setGlobalSettings", context=self._plugin_registration_uuid, @@ -52,14 +60,14 @@ def get_global_settings(self) -> None: context=self._plugin_registration_uuid, ) - def open_url(self, context: str, url: str) -> None: + def open_url(self, context: ActionInstanceUUIDStr, url: str) -> None: self._send_event( event="openUrl", context=context, payload={"url": url}, ) - def log_message(self, context: str, message: str) -> None: + def log_message(self, context: ActionInstanceUUIDStr, message: str) -> None: self._send_event( event="logMessage", context=context, @@ -68,10 +76,10 @@ def log_message(self, context: str, message: str) -> None: def set_title( self, - context: str, + context: ActionInstanceUUIDStr, state: int | None = None, - target: str | None = None, - title: str | None = None + target: Literal["hardware", "software", "both"] | None = None, + title: str | None = None, ) -> None: payload = {} @@ -90,10 +98,10 @@ def set_title( def set_image( self, - context: str, - image: str, # base64 encoded image, - target: Literal["hardware", "software", "both"], # software, hardware, or both, - state: int, # 0-based integer + context: ActionInstanceUUIDStr, + image: str, # base64 encoded image, + target: Literal["hardware", "software", "both"], + state: int, ) -> None: """... @@ -117,14 +125,26 @@ def set_image( }, ) - def set_feedback(self, context: str, payload: dict[str, Any]) -> None: + def set_feedback(self, context: ActionInstanceUUIDStr, payload: PluginDefinedData) -> None: + """Set's the feedback of an existing layout associated with an action instance. + + Args: + context (str): Defines the context of the command, e.g. which action instance the command is intended for. + payload (PluginDefinedData): Additional information supplied as part of the command. + """ self._send_event( event="setFeedback", context=context, payload=payload, ) - def set_feedback_layout(self, context: str, layout: str) -> None: + def set_feedback_layout(self, context: ActionInstanceUUIDStr, layout: str) -> None: + """Sets the layout associated with an action instance. + + Args: + context (str): Defines the context of the command, e.g. which action instance the command is intended for. + layout (str): Name of a pre-defined layout, or relative path to a custom one. + """ self._send_event( event="setFeedbackLayout", context=context, @@ -133,7 +153,7 @@ def set_feedback_layout(self, context: str, layout: str) -> None: def set_trigger_description( self, - context: str, + context: ActionInstanceUUIDStr, rotate: str | None = None, push: str | None = None, touch: str | None = None, @@ -170,21 +190,21 @@ def set_trigger_description( }, ) - def show_alert(self, context: str) -> None: + def show_alert(self, context: ActionInstanceUUIDStr) -> None: """Temporarily show an alert icon on the image displayed by an instance of an action.""" self._send_event( event="showAlert", context=context, ) - def show_ok(self, context: str) -> None: + def show_ok(self, context: ActionInstanceUUIDStr) -> None: """Temporarily show an OK checkmark icon on the image displayed by an instance of an action.""" self._send_event( event="showOk", context=context, ) - def set_state(self, context: str, state: int) -> None: + def set_state(self, context: ActionInstanceUUIDStr, state: int) -> None: self._send_event( event="setState", context=context, @@ -193,8 +213,8 @@ def set_state(self, context: str, state: int) -> None: def switch_to_profile( self, - context: str, - device: str, + context: ActionInstanceUUIDStr, + device: DeviceUUIDStr, profile: str | None = None, page: int = 0, ) -> None: @@ -211,7 +231,7 @@ def switch_to_profile( page (int): Page to show when switching to the profile; indexed from 0. """ # TODO: Should validation happen that ensures the specified profile is declared in manifest.yaml? - payload = {} + payload: dict[str, str | int | None] = {} if profile is not None: payload = { @@ -226,7 +246,9 @@ def switch_to_profile( payload=payload, ) - def send_to_property_inspector(self, context: str, payload: dict[str, Any]) -> None: + def send_to_property_inspector( + self, context: ActionInstanceUUIDStr, payload: PluginDefinedData + ) -> None: self._send_event( event="sendToPropertyInspector", context=context, @@ -234,10 +256,7 @@ def send_to_property_inspector(self, context: str, payload: dict[str, Any]) -> N ) def send_to_plugin( - self, - context: str, - action: str, - payload: dict[str, Any] + self, context: ActionInstanceUUIDStr, action: ActionUUIDStr, payload: PluginDefinedData ) -> None: """Send a payload to another plugin. diff --git a/streamdeck/models/events/__init__.py b/streamdeck/models/events/__init__.py index be53204..6f072e0 100644 --- a/streamdeck/models/events/__init__.py +++ b/streamdeck/models/events/__init__.py @@ -31,6 +31,9 @@ if TYPE_CHECKING: from typing import Final + from streamdeck.types import EventNameStr + + DEFAULT_EVENT_MODELS: Final[list[type[EventBase]]] = [ ApplicationDidLaunch, @@ -56,8 +59,8 @@ ] -def _get_default_event_names() -> set[str]: - default_event_names: set[str] = set() +def _get_default_event_names() -> set[EventNameStr]: + default_event_names: set[EventNameStr] = set() for event_model in DEFAULT_EVENT_MODELS: default_event_names.update(event_model.get_model_event_names()) @@ -65,7 +68,7 @@ def _get_default_event_names() -> set[str]: return default_event_names -DEFAULT_EVENT_NAMES: Final[set[str]] = _get_default_event_names() +DEFAULT_EVENT_NAMES: Final[set[EventNameStr]] = _get_default_event_names() diff --git a/streamdeck/models/events/adapter.py b/streamdeck/models/events/adapter.py index bcafb77..4c4e1d8 100644 --- a/streamdeck/models/events/adapter.py +++ b/streamdeck/models/events/adapter.py @@ -8,7 +8,10 @@ if TYPE_CHECKING: + from typing_extensions import TypeIs + from streamdeck.models.events.base import EventBase + from streamdeck.types import EventNameStr class EventAdapter: @@ -17,7 +20,7 @@ def __init__(self) -> None: self._models: list[type[EventBase]] = [] self._type_adapter: TypeAdapter[EventBase] | None = None - self._event_names: set[str] = set() + self._event_names: set[EventNameStr] = 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. """ @@ -32,7 +35,7 @@ def add_model(self, model: type[EventBase]) -> None: # so `get_model_event_names()` returns a tuple of all event names, even if there is only one. self._event_names.update(model.get_model_event_names()) - def event_name_exists(self, event_name: str) -> bool: + def event_name_exists(self, event_name: str) -> TypeIs[EventNameStr]: """Check if an event name has been registered with the adapter.""" return event_name in self._event_names diff --git a/streamdeck/models/events/base.py b/streamdeck/models/events/base.py index 195f3b6..9a7dce6 100644 --- a/streamdeck/models/events/base.py +++ b/streamdeck/models/events/base.py @@ -55,6 +55,8 @@ class EventMetadataDict(TypedDict): from pydantic._internal._model_construction import ModelMetaclass # type: ignore[import] + from streamdeck.types import EventNameStr + class EventMeta(ModelMetaclass): """Metaclass for EventBase stub to satisfy static type checkers.""" @classmethod @@ -78,7 +80,7 @@ class TestEvent(EventBase["test", "testing"]): ``` """ - event: LiteralString + 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. @@ -89,7 +91,7 @@ class TestEvent(EventBase["test", "testing"]): """Return the event type for the event model.""" @classmethod - def get_model_event_names(cls) -> tuple[str, ...]: + def get_model_event_names(cls) -> tuple[EventNameStr, ...]: """Return the event names for the event model.""" ... diff --git a/streamdeck/models/events/common.py b/streamdeck/models/events/common.py index 14b10cc..7a9ad12 100644 --- a/streamdeck/models/events/common.py +++ b/streamdeck/models/events/common.py @@ -3,36 +3,38 @@ from abc import ABC from typing import Annotated, Generic, Literal, NamedTuple, Optional, Union -from pydantic import Field, JsonValue +from pydantic import Field from typing_extensions import TypedDict, TypeVar from streamdeck.models.events.base import ConfiguredBaseModel +from streamdeck.types import ( # noqa: TC001 + ActionInstanceUUIDStr, + DeviceUUIDStr, + EventNameStr, + PluginDefinedData, +) ## Mixin classes for common event model fields. class ContextualEventMixin: """Mixin class for event models that have action and context fields.""" - action: str + action: EventNameStr """Unique identifier of the action""" - context: str + context: ActionInstanceUUIDStr """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 + device: DeviceUUIDStr """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, JsonValue] -"""Data of arbitrary structure that is defined in and relevant to the plugin. -The root of the data structure will always be a dict of string keys, while the values can be any JSON-compatible type. -""" EncoderControllerType = Literal["Encoder"] diff --git a/streamdeck/models/events/property_inspector.py b/streamdeck/models/events/property_inspector.py index 159cc4b..4d5fe33 100644 --- a/streamdeck/models/events/property_inspector.py +++ b/streamdeck/models/events/property_inspector.py @@ -4,8 +4,8 @@ from streamdeck.models.events.common import ( ContextualEventMixin, DeviceSpecificEventMixin, - PluginDefinedData, ) +from streamdeck.types import PluginDefinedData # noqa: TC001 class DidReceivePropertyInspectorMessage(EventBase["sendToPlugin"], ContextualEventMixin): diff --git a/streamdeck/models/events/settings.py b/streamdeck/models/events/settings.py index f4f40cf..61a86fb 100644 --- a/streamdeck/models/events/settings.py +++ b/streamdeck/models/events/settings.py @@ -9,10 +9,10 @@ DeviceSpecificEventMixin, KeypadControllerType, MultiActionPayloadMixin, - PluginDefinedData, SingleActionPayloadMixin, StatefulActionPayloadMixin, ) +from streamdeck.types import PluginDefinedData # noqa: TC001 ## Models for didReceiveSettings event and its specific payloads. diff --git a/streamdeck/models/events/title_parameters.py b/streamdeck/models/events/title_parameters.py index 538fb8f..8446900 100644 --- a/streamdeck/models/events/title_parameters.py +++ b/streamdeck/models/events/title_parameters.py @@ -7,6 +7,7 @@ from streamdeck.models.events.base import ConfiguredBaseModel, EventBase from streamdeck.models.events.common import ( BasePayload, + ContextualEventMixin, CoordinatesPayloadMixin, DeviceSpecificEventMixin, StatefulActionPayloadMixin, @@ -47,8 +48,6 @@ class TitleParametersDidChangePayload( """Defines aesthetic properties that determine how the title should be rendered.""" -class TitleParametersDidChange(EventBase["titleParametersDidChange"], DeviceSpecificEventMixin): +class TitleParametersDidChange(EventBase["titleParametersDidChange"], ContextualEventMixin, DeviceSpecificEventMixin): """Occurs when the user updates an action's title settings in the Stream Deck application.""" - 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/types.py b/streamdeck/types.py new file mode 100644 index 0000000..a449b50 --- /dev/null +++ b/streamdeck/types.py @@ -0,0 +1,33 @@ +# ruff: noqa: UP040 +from __future__ import annotations + +from typing import TYPE_CHECKING + +from pydantic import JsonValue + + +if TYPE_CHECKING: + from typing_extensions import TypeAlias # noqa: UP035 + + + +EventNameStr: TypeAlias = str +"""Type alias for the event name string. + +We don't define literal string values here, as the list of available event names can be added to dynamically. +""" + +DeviceUUIDStr: TypeAlias = str +"""Unique identifier string of a Stream Deck device that an event is associated with.""" + +ActionUUIDStr: TypeAlias = str +"""Unique identifier string of a Stream Deck action that an event is associated with.""" + +ActionInstanceUUIDStr: TypeAlias = str +"""Unique identifier string of a specific instance of an action that an event is associated with (e.g. a specific key or dial that the action is assigned to).""" + +PluginDefinedData: TypeAlias = dict[str, JsonValue] +"""Key-value pair data of arbitrary structure that is defined in and relevant to the plugin. + +The root of the data structure will always be a dict of string keys, while the values can be any JSON-compatible type. +""" diff --git a/tests/plugin_manager/conftest.py b/tests/plugin_manager/conftest.py index 87c02aa..02a4194 100644 --- a/tests/plugin_manager/conftest.py +++ b/tests/plugin_manager/conftest.py @@ -14,7 +14,6 @@ if TYPE_CHECKING: import pytest_mock from streamdeck.models import events - from streamdeck.models.events.base import LiteralStrGenericAlias @pytest.fixture @@ -104,7 +103,7 @@ def mock_event_listener_manager_with_fake_events(patch_event_listener_manager: M """ print("MOCK EVENT LISTENER MANAGER") # Create a list of fake event messages, and convert them to json strings to be passed back by the client.listen() method. - fake_event_messages: list[events.EventBase[LiteralStrGenericAlias]] = [ + fake_event_messages: list[events.EventBase] = [ KeyDownEventFactory.build(action="my-fake-action-uuid"), ]