diff --git a/streamdeck/__main__.py b/streamdeck/__main__.py index 6832762..b6d44bf 100644 --- a/streamdeck/__main__.py +++ b/streamdeck/__main__.py @@ -1,37 +1,38 @@ from __future__ import annotations -import importlib.util import json import logging import sys from argparse import ArgumentParser from pathlib import Path -from typing import TYPE_CHECKING, cast - -import tomli as toml - -from streamdeck.actions import ActionBase -from streamdeck.cli.errors import ( - DirectoryNotFoundError, - NotAFileError, -) -from streamdeck.cli.models import ( - CliArgsNamespace, - PyProjectConfigDict, - StreamDeckConfigDict, -) +from typing import Protocol, cast + from streamdeck.manager import PluginManager +from streamdeck.models.configs import PyProjectConfigs from streamdeck.utils.logging import configure_streamdeck_logger -if TYPE_CHECKING: - from collections.abc import Generator # noqa: I001 - from importlib.machinery import ModuleSpec - from types import ModuleType - from typing_extensions import Self # noqa: UP035 +logger = logging.getLogger("streamdeck") + -logger = logging.getLogger("streamdeck") +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 def setup_cli() -> ArgumentParser: @@ -68,145 +69,27 @@ def setup_cli() -> ArgumentParser: return parser -def determine_action_scripts( - plugin_dir: Path, - action_scripts: list[str] | None, -) -> list[str]: - """Determine the action scripts to be loaded based on provided arguments. - - plugin_dir and action_scripts cannot both have values -> either only one of them isn't None, or they are both None. - - Args: - plugin_dir (Path | None): The directory containing plugin files to load Actions from. - action_scripts (list[str] | None): A list of action script file paths. - - Returns: - list[str]: A list of action script file paths. - - Raises: - KeyError: If the 'action_scripts' setting is missing from the streamdeck config. - """ - # If `action_scripts` arg was provided, then we can ignore plugin_dir (because we can assume plugin_dir is None). - if action_scripts is not None: - return action_scripts - - # If `action_scripts` is None, then either plugin_dir has a value or it is the default CWD. - # Thus either use the value given to plugin_value if it was given one, or fallback to using the current working directory. - streamdeck_config = read_streamdeck_config_from_pyproject(plugin_dir=plugin_dir) - try: - return streamdeck_config["action_scripts"] - - except KeyError as e: - msg = f"'action_plugin' setting missing from streamdeck config in pyproject.toml in '{plugin_dir}'." - raise KeyError(msg) from e - - -def read_streamdeck_config_from_pyproject(plugin_dir: Path) -> StreamDeckConfigDict: - """Get the streamdeck section from a plugin directory by reading pyproject.toml. - - Plugin devs add a section to their pyproject.toml for "streamdeck" to configure setup for their plugin. - - Args: - plugin_dir (Path): The directory containing the pyproject.toml and plugin files. - - Returns: - List[Path]: A list of file paths found in the specified scripts. - - Raises: - DirectoryNotFoundError: If the specified plugin_dir does not exist. - NotADirectoryError: If the specified plugin_dir is not a directory. - FileNotFoundError: If the pyproject.toml file does not exist in the plugin_dir. - """ - if not plugin_dir.exists(): - msg = f"The directory '{plugin_dir}' does not exist." - raise DirectoryNotFoundError(msg, directory=plugin_dir) - - pyproject_path = plugin_dir / "pyproject.toml" - with pyproject_path.open("rb") as f: - try: - pyproject_config: PyProjectConfigDict = toml.load(f) - - except FileNotFoundError as e: - msg = f"There is no 'pyproject.toml' in the given directory '{plugin_dir}" - raise FileNotFoundError(msg) from e - - except NotADirectoryError as e: - msg = f"The provided directory exists but is not a directory: '{plugin_dir}'." - raise NotADirectoryError(msg) from e - - try: - streamdeck_config = pyproject_config["tool"]["streamdeck"] - - except KeyError as e: - msg = f"Section 'tool.streamdeck' is missing from '{pyproject_path}'." - raise KeyError(msg) from e - - return streamdeck_config - - -class ActionLoader: - @classmethod - def load_actions(cls: type[Self], plugin_dir: Path, files: list[str]) -> Generator[ActionBase, None, None]: - # Ensure the parent directory of the plugin modules 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)) - - for action_script in files: - module = cls._load_module_from_file(filepath=Path(action_script)) - yield from cls._get_actions_from_loaded_module(module=module) - - @staticmethod - def _load_module_from_file(filepath: Path) -> ModuleType: - """Load module from a given Python file. - - Args: - filepath (str): The path to the Python file. - - Returns: - ModuleType: A loaded module located at the specified filepath. - - Raises: - FileNotFoundError: If the specified file does not exist. - NotAFileError: If the specified file exists, but is not a file. - """ - # First validate the filepath arg here. - if not filepath.exists(): - msg = f"The file '{filepath}' does not exist." - raise FileNotFoundError(msg) - if not filepath.is_file(): - msg = f"The provided filepath '{filepath}' is not a file." - raise NotAFileError(msg) - - # Create a module specification for a module located at the given filepath. - # A "specification" is an object that contains information about how to load the module, such as its location and loader. - # "module.name" is an arbitrary name used to identify the module internally. - spec: ModuleSpec = importlib.util.spec_from_file_location("module.name", str(filepath)) # type: ignore - # Create a new module object from the given specification. - # At this point, the module is created but not yet loaded (i.e. its code hasn't been executed). - module: ModuleType = importlib.util.module_from_spec(spec) - # Load the module by executing its code, making available its functions, classes, and variables. - spec.loader.exec_module(module) # type: ignore - - return module - - @staticmethod - def _get_actions_from_loaded_module(module: ModuleType) -> Generator[ActionBase, None, None]: - # Iterate over all attributes in the module to find Action subclasses - for attribute_name in dir(module): - attribute = getattr(module, attribute_name) - # Check if the attribute is an instance of the Action class or GlobalAction class. - if issubclass(type(attribute), ActionBase): - yield attribute - - -def main(): +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. - plugin_dir = args.plugin_dir or Path.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"] @@ -215,12 +98,8 @@ def main(): # a child logger with `logging.getLogger("streamdeck.mycomponent")`, all with the same handler/formatter configuration. configure_streamdeck_logger(name="streamdeck", plugin_uuid=plugin_uuid) - action_scripts = determine_action_scripts( - plugin_dir=plugin_dir, - action_scripts=args.action_scripts, - ) - - actions = list(ActionLoader.load_actions(plugin_dir=plugin_dir, files=action_scripts)) + pyproject = PyProjectConfigs.validate_from_toml_file(plugin_dir / "pyproject.toml") + actions = list(pyproject.streamdeck_plugin_actions) manager = PluginManager( port=args.port, diff --git a/streamdeck/cli/__init__.py b/streamdeck/cli/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/streamdeck/cli/errors.py b/streamdeck/cli/errors.py deleted file mode 100644 index eacaae0..0000000 --- a/streamdeck/cli/errors.py +++ /dev/null @@ -1,20 +0,0 @@ -"""Custom exceptions used during validations of cli & config parsing.""" -from __future__ import annotations - -from typing import TYPE_CHECKING - - -if TYPE_CHECKING: - from pathlib import Path - from typing import Any - - -class DirectoryNotFoundError(FileNotFoundError): - """Custom exception to indicate that a specified directory was not found.""" - def __init__(self, *args: Any, directory: Path): - super().__init__(*args) - self.directory = directory - - -class NotAFileError(NotADirectoryError): - """Custom exception to indicate that a provided path is not a file.""" diff --git a/streamdeck/cli/models.py b/streamdeck/cli/models.py deleted file mode 100644 index f9234b6..0000000 --- a/streamdeck/cli/models.py +++ /dev/null @@ -1,52 +0,0 @@ -"""Models for CLI Arguments and pyproject.toml config to be parsed and used by the entry-point.""" -from __future__ import annotations - -from typing import TYPE_CHECKING, Protocol - -from typing_extensions import TypedDict - - -if TYPE_CHECKING: - from pathlib import Path - - - -### CLI Arguments Namespace ### -################################## - -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 - - -### pyproject.toml Nested Config ### -####################################### - -class PyProjectConfigDict(TypedDict): - """Represents the structure of the pyproject.toml configuration file (or at least for what we care about here).""" - tool: ToolConfigDict - - -class ToolConfigDict(TypedDict): - """Represents the 'tool' section within the pyproject.toml.""" - streamdeck: StreamDeckConfigDict - - -class StreamDeckConfigDict(TypedDict): - """Represents the 'streamdeck' configuration section within pyproject.toml.""" - action_scripts: list[str] - - - - - - - - diff --git a/streamdeck/models/configs.py b/streamdeck/models/configs.py new file mode 100644 index 0000000..610b29b --- /dev/null +++ b/streamdeck/models/configs.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +from types import ModuleType +from typing import TYPE_CHECKING, Annotated + +import tomli as toml +from pydantic import ( + BaseModel, + Field, + ImportString, + ValidationInfo, + field_validator, +) + +from streamdeck.actions import ActionBase + + +if TYPE_CHECKING: + from collections.abc import Generator + from pathlib import Path + from typing import Any + + + + +class PyProjectConfigs(BaseModel): + """A Pydantic model for the PyProject.toml configuration file to load a Stream Deck plugin's actions.""" + tool: ToolSection = Field(alias="tool") + + @classmethod + def validate_from_toml_file(cls, filepath: Path, action_scripts: list[str] | None = None) -> PyProjectConfigs: + """Alternative constructor to validate a PyProjectConfigs instance from a TOML file.""" + 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. + ctx = {"action_scripts": action_scripts} if action_scripts else None + + # Return the loaded PyProjectConfigs model instance. + return cls.model_validate(pyproject_configs, context=ctx) + + @property + def streamdeck_plugin_actions(self) -> Generator[type[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): + obj = getattr(loaded_action_script, object_name) + + # Ensure the object isn't a magic method or attribute of the loaded module. + if object_name.startswith("__"): + continue + + yield obj + + +class ToolSection(BaseModel): + """A model class representing the "tool" section in configuration. + + Nothing much to see here, just a wrapper around the model representing the "streamdeck" subsection. + """ + streamdeck: StreamDeckToolConfig + + +class StreamDeckToolConfig(BaseModel, arbitrary_types_allowed=True): + """A model class representing the "streamdeck" subsection in the "tool" section of the PyProject.toml file. + + This section contains the developer's configuration for their Stream Deck plugin. + """ + action_script_modules: Annotated[list[ImportString[ModuleType]], Field(alias="action_scripts")] + """A list of loaded action script modules with all of their objects. + + 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]: + """Filter out non- ActionBase subclasses from the list of objects loaded from each action script module.""" + loaded_modules: list[ModuleType] = [] + + for module in value: + new_module = ModuleType(module.__name__) + + for object_name in dir(module): + obj = getattr(module, object_name) + + if not isinstance(obj, ActionBase): + continue + + setattr(new_module, object_name, obj) + + loaded_modules.append(new_module) + + return loaded_modules + + diff --git a/tests/data/pyproject.toml b/tests/data/pyproject.toml index affb0e8..00f9c53 100644 --- a/tests/data/pyproject.toml +++ b/tests/data/pyproject.toml @@ -3,6 +3,6 @@ [tool.streamdeck] action_scripts = [ - "tests/data/test_action1.py", - "tests/data/test_action2.py", + "tests.data.test_action1", + "tests.data.test_action2", ] \ No newline at end of file diff --git a/tests/data/test_action2.py b/tests/data/test_action2.py index 3c1142a..d676184 100644 --- a/tests/data/test_action2.py +++ b/tests/data/test_action2.py @@ -1,10 +1,17 @@ -from streamdeck.actions import Action +from streamdeck.actions import Action, GlobalAction from streamdeck.models.events import ApplicationDidLaunch test_action2 = Action("my-first-test-action") +test_global_action1 = GlobalAction() + @test_action2.on("applicationDidLaunch") def handle_application_launched_event(event_data: ApplicationDidLaunch) -> None: print("ApplicationDidLaunch event handled.") + + +@test_global_action1.on("applicationDidLaunch") +def handle_application_launched_event_globally(event_data: ApplicationDidLaunch) -> None: + print("Global ApplicationDidLaunch event handled.")