diff --git a/README.md b/README.md index c3bd7e7..44a20ec 100644 --- a/README.md +++ b/README.md @@ -77,9 +77,20 @@ uvx --from streamdeck-mcp streamdeck-mcp-usb - Generated icons are stored in `~/.streamdeck-mcp/generated-icons/`. - Generated shell scripts are stored in `~/StreamDeckScripts/`. +## Editing Workflow (Important) + +The Elgato desktop app keeps every profile in memory and rewrites the on-disk manifests from that snapshot when it quits, so any edit made while the app is running is wiped the next time it closes. The profile writer enforces a quit → write → relaunch cycle: + +1. Ensure the Elgato app is not running, or pass `auto_quit_app: true` to `streamdeck_write_page` to have it quit the app for you (AppleScript first, `killall` fallback). +2. Make as many `streamdeck_write_page` calls as you need — the app stays quit across them. +3. Call `streamdeck_restart_app` when you are done. The device re-reads the manifests on launch and your changes appear. + +`streamdeck_write_page` raises a `StreamDeckAppRunningError` when the app is running and `auto_quit_app` is not set, so you cannot accidentally write changes that will be silently discarded. + +If your Elgato app is installed somewhere other than `/Applications/Elgato Stream Deck.app`, set `STREAMDECK_APP_PATH` to the bundle path. + ## Usage Notes -- After writing profiles, the Elgato desktop app may need a restart to pick up the new manifests. - `streamdeck_create_action` is the safest way to build shell-command buttons because it writes a standalone script and returns the native Open action block for it. - The profile writer does not require exclusive USB access. diff --git a/profile_manager.py b/profile_manager.py index ed26eab..10f6599 100644 --- a/profile_manager.py +++ b/profile_manager.py @@ -20,6 +20,7 @@ import string import subprocess import sys +import time import uuid from dataclasses import dataclass from pathlib import Path @@ -70,6 +71,13 @@ "UI Stream Deck": {KEYPAD: (4, 2)}, } +# The Elgato Stream Deck desktop app caches every profile in memory and rewrites the +# on-disk manifests when it quits, so any edit made while it is running gets clobbered +# the next time the user closes or restarts the app. +STREAM_DECK_APP_PROCESS_NAMES = ("Stream Deck", "Elgato Stream Deck") +DEFAULT_STREAM_DECK_APP_PATH = Path("/Applications/Elgato Stream Deck.app") +STREAM_DECK_APP_PATH_ENV = "STREAMDECK_APP_PATH" + HEX_COLOR_PATTERN = re.compile(r"^#[0-9a-fA-F]{6}$") POSITION_PATTERN = re.compile(r"^\d+,\d+$") UUID_PATTERN = re.compile( @@ -102,6 +110,16 @@ class ProfileValidationError(ProfileManagerError): """Raised when inputs for profile operations are invalid.""" +class StreamDeckAppRunningError(ProfileManagerError): + """Raised when a write is attempted while the Elgato desktop app is running. + + The app rewrites every profile manifest from its in-memory snapshot on quit, so + writes made while it is running are silently discarded. Callers must quit the + app first (pass `auto_quit_app=True` to `write_page`) and then call + `restart_app` once their edits are complete to see the changes. + """ + + @dataclass class PageRef: """Resolved page directory metadata.""" @@ -277,6 +295,79 @@ def _count_icons(page_dir: Path) -> int: return len([path for path in images_dir.iterdir() if path.is_file()]) +def _resolve_app_path() -> Path: + override = os.environ.get(STREAM_DECK_APP_PATH_ENV) + if override: + return Path(override).expanduser() + return DEFAULT_STREAM_DECK_APP_PATH + + +def is_stream_deck_app_running() -> bool: + """Return True if the Elgato Stream Deck desktop app is currently running.""" + + if sys.platform != "darwin": + return False + + for name in STREAM_DECK_APP_PROCESS_NAMES: + result = subprocess.run( + ["pgrep", "-x", name], + capture_output=True, + text=True, + check=False, + ) + if result.returncode == 0 and result.stdout.strip(): + return True + return False + + +def stop_stream_deck_app(*, graceful_timeout: float = 3.0) -> dict[str, Any]: + """Quit the Elgato Stream Deck desktop app. + + Tries an AppleScript quit first so the app can persist any unrelated state, then + falls back to `killall` if it does not exit in time. Returns a small report about + which path was taken. + """ + + if sys.platform != "darwin": + return {"stopped": False, "graceful": [], "forced": [], "reason": "non-darwin platform"} + + if not is_stream_deck_app_running(): + return {"stopped": False, "graceful": [], "forced": [], "reason": "not running"} + + graceful: list[str] = [] + for name in STREAM_DECK_APP_PROCESS_NAMES: + result = subprocess.run( + ["osascript", "-e", f'tell application "{name}" to quit'], + capture_output=True, + text=True, + check=False, + ) + if result.returncode == 0: + graceful.append(name) + + deadline = time.monotonic() + graceful_timeout + while time.monotonic() < deadline and is_stream_deck_app_running(): + time.sleep(0.2) + + forced: list[str] = [] + if is_stream_deck_app_running(): + for name in STREAM_DECK_APP_PROCESS_NAMES: + result = subprocess.run( + ["killall", name], + capture_output=True, + text=True, + check=False, + ) + if result.returncode == 0: + forced.append(name) + + return { + "stopped": not is_stream_deck_app_running(), + "graceful": graceful, + "forced": forced, + } + + class ProfileManager: """Read and write Elgato Stream Deck profiles.""" @@ -415,9 +506,30 @@ def write_page( clear_existing: bool = True, create_new: bool = False, make_current: bool = False, + auto_quit_app: bool = False, ) -> dict[str, Any]: """Create a page or rewrite an existing page manifest.""" + app_stop_report: dict[str, Any] | None = None + if is_stream_deck_app_running(): + if not auto_quit_app: + raise StreamDeckAppRunningError( + "The Elgato Stream Deck app is running and will overwrite this " + "edit on quit. Retry with auto_quit_app=True to quit it first, " + "then call streamdeck_restart_app once your edits are complete " + "to apply the changes." + ) + app_stop_report = stop_stream_deck_app() + stop_failed = not app_stop_report.get("stopped", False) + still_running = is_stream_deck_app_running() + if stop_failed or still_running: + reason = app_stop_report.get("reason", "") + detail = f" Reason: {reason}." if reason else "" + raise StreamDeckAppRunningError( + f"The Elgato Stream Deck app could not be stopped.{detail} Aborting " + "page write because the running app may overwrite these edits on quit." + ) + profile_dir, profile_manifest = self._resolve_profile( profile_name=profile_name, profile_id=profile_id ) @@ -518,6 +630,7 @@ def write_page( "button_count": total_button_count, "page_name": page_manifest.get("Name", ""), "manifest_path": str(page_dir / "manifest.json"), + "app_quit": app_stop_report, } def create_icon( @@ -599,40 +712,34 @@ def restart_app(self) -> dict[str, Any]: "streamdeck_restart_app is currently only supported on macOS." ) - killed = [] - for app_name in ("Elgato Stream Deck", "Stream Deck"): - result = subprocess.run( - ["killall", app_name], - capture_output=True, - text=True, - check=False, + app_path = _resolve_app_path() + if not app_path.exists(): + raise ProfileManagerError( + f"Stream Deck app not found at {app_path}. " + f"Set {STREAM_DECK_APP_PATH_ENV} to override the default install path." ) - if result.returncode == 0: - killed.append(app_name) - open_result = subprocess.run( - ["open", "-a", "Stream Deck"], + stop_report = stop_stream_deck_app() + + # `open -a ` relies on LaunchServices name lookup, which returns error + # -600 on some systems even when the bundle is present. Launching by explicit + # path bypasses that lookup. + result = subprocess.run( + ["open", str(app_path)], capture_output=True, text=True, check=False, ) - if open_result.returncode != 0: - fallback = subprocess.run( - ["open", "-a", "Elgato Stream Deck"], - capture_output=True, - text=True, - check=False, - ) - open_result = fallback - - if open_result.returncode != 0: + if result.returncode != 0: + message = result.stderr.strip() or result.stdout.strip() raise ProfileManagerError( - open_result.stderr.strip() or "Failed to relaunch Stream Deck app." + f"Failed to relaunch Stream Deck ({app_path}): {message or 'unknown error'}" ) return { - "killed": killed, "restarted": True, + "app_path": str(app_path), + "stop": stop_report, } def _resolve_profile( diff --git a/profile_server.py b/profile_server.py index 9fce0e5..e372b06 100644 --- a/profile_server.py +++ b/profile_server.py @@ -23,6 +23,7 @@ ProfileManagerError, ProfileNotFoundError, ProfileValidationError, + StreamDeckAppRunningError, ) logging.basicConfig( @@ -189,7 +190,12 @@ async def list_tools() -> list[Tool]: Tool( name="streamdeck_write_page", description=( - "Create a new page or replace/update an existing Stream Deck desktop page manifest." + "Create a new page or replace/update an existing Stream Deck desktop " + "page manifest. IMPORTANT: the Elgato desktop app overwrites profile " + "manifests from its in-memory state on quit, so writes made while the " + "app is running are lost. This tool refuses to write when the app is " + "running unless auto_quit_app=True is passed. Call streamdeck_restart_app " + "once your edits are complete to make the changes visible on the device." ), inputSchema={ "type": "object", @@ -227,6 +233,16 @@ async def list_tools() -> list[Tool]: "When true, make the page the active current page after writing." ), }, + "auto_quit_app": { + "type": "boolean", + "description": ( + "If true and the Elgato Stream Deck desktop app is " + "running, quit it (graceful AppleScript first, then " + "killall) before writing. Required when the app is " + "running or the write will raise an error. Defaults to " + "false so callers must explicitly consent to quitting it." + ), + }, }, }, ), @@ -322,6 +338,7 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: clear_existing=arguments.get("clear_existing", True), create_new=arguments.get("create_new", False), make_current=arguments.get("make_current", False), + auto_quit_app=arguments.get("auto_quit_app", False), ) return [TextContent(type="text", text=json.dumps(result, indent=2))] @@ -350,6 +367,8 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: return [TextContent(type="text", text=f"❌ Unknown tool: {name}")] + except StreamDeckAppRunningError as exc: + return [TextContent(type="text", text=f"⚠️ {exc}")] except ( ProfileManagerError, ProfileNotFoundError, diff --git a/tests/test_profile_manager.py b/tests/test_profile_manager.py index 7646649..0d0232b 100644 --- a/tests/test_profile_manager.py +++ b/tests/test_profile_manager.py @@ -9,7 +9,21 @@ import pytest -from profile_manager import PageNotFoundError, ProfileManager, ProfileValidationError +from profile_manager import ( + PageNotFoundError, + ProfileManager, + ProfileManagerError, + ProfileValidationError, + StreamDeckAppRunningError, +) + + +@pytest.fixture(autouse=True) +def _stub_stream_deck_app_not_running(): + """Default: act as if the Stream Deck app is not running so tests hit the write path.""" + + with patch("profile_manager.is_stream_deck_app_running", return_value=False): + yield def _write_json(path: Path, payload: dict) -> None: @@ -584,3 +598,134 @@ def test_write_page_clear_existing_with_empty_buttons_clears_keypad( profile_name="Default Profile", directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB" ) assert page["buttons"] == [] + + +def test_write_page_refuses_while_app_running(sample_profiles_v3: Path, tmp_path: Path) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + + with patch("profile_manager.is_stream_deck_app_running", return_value=True): + with pytest.raises(StreamDeckAppRunningError, match="running"): + manager.write_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + buttons=[{"key": 0, "title": "x", "action_type": "next_page"}], + ) + + +def test_write_page_auto_quit_app_stops_then_writes( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + stop_report = {"stopped": True, "graceful": ["Stream Deck"], "forced": []} + + with ( + patch("profile_manager.is_stream_deck_app_running", side_effect=[True, False]), + patch("profile_manager.stop_stream_deck_app", return_value=stop_report) as stop_mock, + ): + result = manager.write_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + buttons=[{"key": 0, "title": "x", "action_type": "next_page"}], + auto_quit_app=True, + ) + + stop_mock.assert_called_once() + assert result["app_quit"] == stop_report + page = manager.read_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + ) + assert page["buttons"][0]["title"] == "x" + + +def test_write_page_auto_quit_app_raises_when_stop_fails( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + stop_report = {"stopped": False, "graceful": [], "forced": [], "reason": "killall failed"} + + with ( + patch("profile_manager.is_stream_deck_app_running", return_value=True), + patch("profile_manager.stop_stream_deck_app", return_value=stop_report), + ): + with pytest.raises(StreamDeckAppRunningError, match="could not be stopped"): + manager.write_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + buttons=[{"key": 0, "title": "x", "action_type": "next_page"}], + auto_quit_app=True, + ) + + +def test_write_page_records_no_quit_when_app_not_running( + sample_profiles_v3: Path, tmp_path: Path +) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + + # autouse fixture already pins is_stream_deck_app_running to False + result = manager.write_page( + profile_name="Default Profile", + directory_id="BBBBBBBB-BBBB-BBBB-BBBB-BBBBBBBBBBBB", + buttons=[{"key": 0, "title": "x", "action_type": "next_page"}], + ) + assert result["app_quit"] is None + + +def test_restart_app_launches_by_explicit_path(sample_profiles_v3: Path, tmp_path: Path) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + + fake_app = tmp_path / "Fake Stream Deck.app" + fake_app.mkdir() + open_call = {"args": None} + + def _fake_run(cmd, **kwargs): + open_call["args"] = cmd + return type("R", (), {"returncode": 0, "stdout": "", "stderr": ""})() + + with ( + patch("profile_manager.sys.platform", "darwin"), + patch.dict("os.environ", {"STREAMDECK_APP_PATH": str(fake_app)}), + patch("profile_manager.stop_stream_deck_app", return_value={"stopped": True}), + patch("profile_manager.subprocess.run", side_effect=_fake_run), + ): + result = manager.restart_app() + + assert open_call["args"] == ["open", str(fake_app)] + assert result["app_path"] == str(fake_app) + assert result["restarted"] is True + + +def test_restart_app_errors_when_app_missing(sample_profiles_v3: Path, tmp_path: Path) -> None: + manager = ProfileManager( + profiles_dir=sample_profiles_v3, + scripts_dir=tmp_path / "scripts", + generated_icons_dir=tmp_path / "icons", + ) + missing_app = tmp_path / "nope.app" + + with ( + patch("profile_manager.sys.platform", "darwin"), + patch.dict("os.environ", {"STREAMDECK_APP_PATH": str(missing_app)}), + ): + with pytest.raises(ProfileManagerError, match="not found"): + manager.restart_app()