Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@
"pydantic_core >= 2.23.4",
"tomli >= 2.0.2",
"websockets >= 13.1",
"typer >= 0.15.1",
]

[project.optional-dependencies]
Expand Down
111 changes: 30 additions & 81 deletions streamdeck/__main__.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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:
Expand All @@ -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()



7 changes: 3 additions & 4 deletions streamdeck/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand All @@ -21,7 +21,6 @@
if TYPE_CHECKING:
from typing import Any, Literal

from streamdeck.actions import Action
from streamdeck.models.events import EventBase


Expand Down Expand Up @@ -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)
Expand Down
36 changes: 21 additions & 15 deletions streamdeck/models/configs.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
ImportString,
ValidationInfo,
field_validator,
model_validator,
)

from streamdeck.actions import ActionBase
Expand All @@ -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):
Expand Down Expand Up @@ -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]:
Expand Down