Merge pull request #10 from strohganoff/feature/typer-cli

Feature/typer cli

Author: strohganoff

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]

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()
+
+
+

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)

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]: