diff --git a/pyproject.toml b/pyproject.toml index ae8e30b..53e6b07 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -62,6 +62,7 @@ "pydantic_core >= 2.23.4", "tomli >= 2.0.2", "websockets >= 13.1", + "typer >= 0.15.1", ] [project.optional-dependencies] diff --git a/streamdeck/__main__.py b/streamdeck/__main__.py index b6d44bf..8f7ed56 100644 --- a/streamdeck/__main__.py +++ b/streamdeck/__main__.py @@ -1,11 +1,10 @@ -from __future__ import annotations - import json import logging import sys -from argparse import ArgumentParser from pathlib import Path -from typing import Protocol, cast +from typing import Annotated, Union + +import typer from streamdeck.manager import PluginManager from streamdeck.models.configs import PyProjectConfigs @@ -16,99 +15,45 @@ -class DirectoryNotFoundError(FileNotFoundError): - """Custom exception to indicate that a specified directory was not found.""" - def __init__(self, *args: object, directory: Path): - super().__init__(*args) - self.directory = directory - - -class CliArgsNamespace(Protocol): - """Represents the command-line arguments namespace.""" - plugin_dir: Path | None - action_scripts: list[str] | None - - # Args always passed in by StreamDeck software - port: int - pluginUUID: str # noqa: N815 - registerEvent: str # noqa: N815 - info: str # Actually a string representation of json object +plugin = typer.Typer() -def setup_cli() -> ArgumentParser: - """Set up the command-line interface for the script. +@plugin.command() +def main( + port: Annotated[int, typer.Option("-p", "-port")], + plugin_registration_uuid: Annotated[str, typer.Option("-pluginUUID")], + register_event: Annotated[str, typer.Option("-registerEvent")], + info: Annotated[str, typer.Option("-info")], + plugin_dir: Annotated[Path, typer.Option(file_okay=False, exists=True, readable=True)] = Path.cwd(), # noqa: B008 + action_scripts: Union[list[str], None] = None, # noqa: UP007 +) -> None: + """Start the Stream Deck plugin with the given configuration. - Returns: - argparse.ArgumentParser: The argument parser for the CLI. + NOTE: Single flag long-name options are extected & passed in by the Stream Deck software. + Double flag long-name options are used during development and testing. """ - parser = ArgumentParser(description="CLI to load Actions from action scripts.") - group = parser.add_mutually_exclusive_group(required=False) - group.add_argument( - "plugin_dir", - type=Path, - nargs="?", - help="The directory containing plugin files to load Actions from.", - ) - group.add_argument( - "--action-scripts", - type=str, - nargs="+", - help="A list of action script file paths to load Actions from or a single value to be processed.", - ) - - # Options that will always be passed in by the StreamDeck software when running this plugin. - parser.add_argument("-port", dest="port", type=int, help="Port", required=True) - parser.add_argument( - "-pluginUUID", dest="pluginUUID", type=str, help="pluginUUID", required=True - ) - parser.add_argument( - "-registerEvent", dest="registerEvent", type=str, help="registerEvent", required=True - ) - parser.add_argument("-info", dest="info", type=str, help="info", required=True) - - return parser - - -def main() -> None: - """Main function to parse arguments, load actions, and execute them.""" - parser = setup_cli() - args = cast(CliArgsNamespace, parser.parse_args()) - - # If `plugin_dir` was not passed in as a cli option, then fall back to using the CWD. - if args.plugin_dir is None: - plugin_dir = Path.cwd() - # Also validate the plugin_dir argument. - elif not args.plugin_dir.is_dir(): - msg = f"The provided plugin directory '{args.plugin_dir}' is not a directory." - raise NotADirectoryError(msg) - elif not args.plugin_dir.exists(): - msg = f"The provided plugin directory '{args.plugin_dir}' does not exist." - raise DirectoryNotFoundError(msg, directory=args.plugin_dir) - else: - plugin_dir = args.plugin_dir - # Ensure plugin_dir is in `sys.path`, so that import statements in the plugin module will work as expected. if str(plugin_dir) not in sys.path: sys.path.insert(0, str(plugin_dir)) - info = json.loads(args.info) - plugin_uuid = info["plugin"]["uuid"] + info_data = json.loads(info) + plugin_uuid = info_data["plugin"]["uuid"] # After configuring once here, we can grab the logger in any other module with `logging.getLogger("streamdeck")`, or # a child logger with `logging.getLogger("streamdeck.mycomponent")`, all with the same handler/formatter configuration. configure_streamdeck_logger(name="streamdeck", plugin_uuid=plugin_uuid) - pyproject = PyProjectConfigs.validate_from_toml_file(plugin_dir / "pyproject.toml") - actions = list(pyproject.streamdeck_plugin_actions) + pyproject = PyProjectConfigs.validate_from_toml_file(plugin_dir / "pyproject.toml", action_scripts=action_scripts) + actions = pyproject.streamdeck_plugin_actions manager = PluginManager( - port=args.port, + port=port, plugin_uuid=plugin_uuid, # NOT the configured plugin UUID in the manifest.json, # which can be pulled out of `info["plugin"]["uuid"]` - plugin_registration_uuid=args.pluginUUID, - register_event=args.registerEvent, - info=info, + plugin_registration_uuid=plugin_registration_uuid, + register_event=register_event, + info=info_data, ) for action in actions: @@ -117,5 +62,9 @@ def main() -> None: manager.run() -if __name__ == "__main__": - main() +# Also run the plugin if this script is ran as a console script. +if __name__ in ("__main__", "streamdeck.__main__"): + plugin() + + + diff --git a/streamdeck/manager.py b/streamdeck/manager.py index 0690078..1cd76d6 100644 --- a/streamdeck/manager.py +++ b/streamdeck/manager.py @@ -4,7 +4,7 @@ from logging import getLogger from typing import TYPE_CHECKING -from streamdeck.actions import ActionRegistry +from streamdeck.actions import Action, ActionBase, ActionRegistry from streamdeck.command_sender import StreamDeckCommandSender from streamdeck.models.events import ContextualEventMixin, event_adapter from streamdeck.types import ( @@ -21,7 +21,6 @@ if TYPE_CHECKING: from typing import Any, Literal - from streamdeck.actions import Action from streamdeck.models.events import EventBase @@ -62,14 +61,14 @@ def __init__( self._registry = ActionRegistry() - def register_action(self, action: Action) -> None: + def register_action(self, action: ActionBase) -> None: """Register an action with the PluginManager, and configure its logger. Args: action (Action): The action to register. """ # First, configure a logger for the action, giving it the last part of its uuid as name (if it has one). - action_component_name = action.uuid.split(".")[-1] if hasattr(action, "uuid") else "global" + action_component_name = action.uuid.split(".")[-1] if isinstance(action, Action) else "global" configure_streamdeck_logger(name=action_component_name, plugin_uuid=self.uuid) self._registry.register(action) diff --git a/streamdeck/models/configs.py b/streamdeck/models/configs.py index 610b29b..94d576d 100644 --- a/streamdeck/models/configs.py +++ b/streamdeck/models/configs.py @@ -10,6 +10,7 @@ ImportString, ValidationInfo, field_validator, + model_validator, ) from streamdeck.actions import ActionBase @@ -33,14 +34,32 @@ def validate_from_toml_file(cls, filepath: Path, action_scripts: list[str] | Non with filepath.open("rb") as f: pyproject_configs = toml.load(f) - # Pass the action scripts to the context dictionary if they are provided, so they can be used in the before-validater for the nested StreamDeckToolConfig model. + # Pass the action scripts to the context dictionary if they are provided, + # so they can be used in the before-validater for the nested StreamDeckToolConfig model. ctx = {"action_scripts": action_scripts} if action_scripts else None # Return the loaded PyProjectConfigs model instance. return cls.model_validate(pyproject_configs, context=ctx) + @model_validator(mode="before") + @classmethod + def overwrite_action_scripts(cls, data: object, info: ValidationInfo) -> object: + """If action scripts were provided as a context variable, overwrite the action_scripts field in the PyProjectConfigs model.""" + context = info.context + + # If no action scripts were provided, return the data as-is. + if context is None or "action_scripts" not in context: + return data + + # If data isn't a dict as expected, let Pydantic's validation handle them as usual in its next validations step. + if isinstance(data, dict): + # We also need to ensure the "tool" and "streamdeck" sections exist in the data dictionary in case they were not defined in the PyProject.toml file. + data.setdefault("tool", {}).setdefault("streamdeck", {})["action_scripts"] = context["action_scripts"] + + return data + @property - def streamdeck_plugin_actions(self) -> Generator[type[ActionBase], Any, None]: + def streamdeck_plugin_actions(self) -> Generator[ActionBase, Any, None]: """Reach into the [tool.streamdeck] section of the PyProject.toml file and yield the plugin's actions configured by the developer.""" for loaded_action_script in self.tool.streamdeck.action_script_modules: for object_name in dir(loaded_action_script): @@ -72,19 +91,6 @@ class StreamDeckToolConfig(BaseModel, arbitrary_types_allowed=True): This field is filtered to only include objects that are subclasses of ActionBase (as well as the built-in magic methods and attributes typically found in a module). """ - @field_validator("action_script_modules", mode="before") - @classmethod - def overwrite_action_scripts_with_user_provided_data(cls, value: list[str], info: ValidationInfo) -> list[str]: - """Overwrite the list of action script modules with the user-provided data. - - NOTE: This is a before-validator that runs before the next field_validator method on the same field. - """ - # If the user provided action_scripts to load, use that instead of the value from the PyProject.toml file. - if info.context is not None and "action_scripts" in info.context: - return info.context["action_scripts"] - - return value - @field_validator("action_script_modules", mode="after") @classmethod def filter_module_objects(cls, value: list[ModuleType]) -> list[ModuleType]: