diff --git a/docs/superpowers/plans/2026-04-15-media-pause-windows-smtc.md b/docs/superpowers/plans/2026-04-15-media-pause-windows-smtc.md new file mode 100644 index 0000000..90df6e2 --- /dev/null +++ b/docs/superpowers/plans/2026-04-15-media-pause-windows-smtc.md @@ -0,0 +1,694 @@ +# Windows SMTC Media-Controller Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Windows-Implementierung der Media-Pause-Funktion via Global System Media Transport Controls (GSMTC / SMTC) — pausiert alle laufenden Mediaplayer bei Aufnahmebeginn und setzt sie danach fort. + +**Architecture:** `SmtcController` in `whisper_local/media/_smtc.py` implementiert das bestehende `MediaController`-Protocol. Die Factory in `__init__.py` bekommt einen `win32`-Dispatch-Zweig. Sessions werden über ihre AUMID (`source_app_user_model_id`) identifiziert; ein Circuit-Breaker verhindert Reconnect-Versuche wenn SMTC nicht erreichbar ist. + +**Tech Stack:** Python 3.13, `pywinrt` (`winrt-Windows.Media.Control`, `winrt-Windows.Foundation`, `winrt-Windows.Foundation.Collections`), `pytest-asyncio`, `unittest.mock` + +--- + +## Dateien + +| Datei | Aktion | Zweck | +|---|---|---| +| `pyproject.toml` | Modify | pywinrt-Dependencies hinzufügen | +| `whisper_local/media/_smtc.py` | Create | SmtcController-Implementierung | +| `whisper_local/media/__init__.py` | Modify | win32-Dispatch-Zweig ergänzen | +| `tests/test_media_smtc.py` | Create | Tests für SmtcController | +| `tests/test_media_factory.py` | Modify | win32-Factory-Test + Noop-Test-Fix | + +--- + +## Task 1: pywinrt-Dependencies hinzufügen + +**Files:** +- Modify: `pyproject.toml` + +- [ ] **Schritt 1: Dependencies ergänzen** + +In `pyproject.toml` die `dependencies`-Liste um drei Einträge erweitern (nach der `pywin32`-Zeile): + +```toml +dependencies = [ + "faster-whisper>=1.1.0", + "sounddevice>=0.5.0", + "numpy>=2.0.0", + "evdev>=1.7.0; sys_platform == 'linux'", + "PyGObject>=3.50; sys_platform == 'linux'", + "dbus-next>=0.2.3; sys_platform == 'linux'", + "pynput>=1.7.0; sys_platform == 'win32'", + "pywin32>=306; sys_platform == 'win32'", + "winrt-Windows.Media.Control>=3.2.1; sys_platform == 'win32'", + "winrt-Windows.Foundation>=3.2.1; sys_platform == 'win32'", + "winrt-Windows.Foundation.Collections>=3.2.1; sys_platform == 'win32'", + "pystray>=0.19.0", + "Pillow>=10.0.0", + "sv-ttk>=2.6.0", + "darkdetect>=0.8.0", +] +``` + +- [ ] **Schritt 2: Sync ausführen** + +```bash +uv sync +``` + +Erwartete Ausgabe: Pakete `winrt-runtime`, `winrt-windows-media-control`, `winrt-windows-foundation`, `winrt-windows-foundation-collections` werden aufgelöst (sind ggf. bereits installiert). + +- [ ] **Schritt 3: Import prüfen** + +```bash +uv run python -c "import winrt.windows.media.control; print('OK')" +``` + +Erwartete Ausgabe: `OK` + +- [ ] **Schritt 4: Committen** + +```bash +git add pyproject.toml uv.lock +git commit -m "build: pywinrt als win32-Dependency hinzufügen" +``` + +--- + +## Task 2: SmtcController — Skeleton + Circuit-Breaker (TDD) + +**Files:** +- Create: `whisper_local/media/_smtc.py` +- Create: `tests/test_media_smtc.py` + +- [ ] **Schritt 1: Testdatei mit circuit-breaker-Tests anlegen** + +Datei `tests/test_media_smtc.py` erstellen: + +```python +"""Tests für SmtcController (Windows/SMTC).""" + +import sys +from unittest.mock import AsyncMock, MagicMock + +import pytest + +pytestmark = pytest.mark.skipif( + sys.platform != "win32", reason="SMTC is Windows-only" +) + +from winrt.windows.media.control import ( + GlobalSystemMediaTransportControlsSessionPlaybackStatus as Status, +) + +PLAYING = Status.PLAYING +PAUSED = Status.PAUSED + + +def _make_session(aumid: str, status) -> MagicMock: + """Erzeugt eine gemockte SMTC-Session mit gegebenem PlaybackStatus.""" + session = MagicMock() + session.source_app_user_model_id = aumid + info = MagicMock() + info.playback_status = status + session.get_playback_info = MagicMock(return_value=info) + session.try_pause_async = AsyncMock() + session.try_play_async = AsyncMock() + return session + + +def _make_manager(sessions: list) -> MagicMock: + """Erzeugt einen gemockten SMTC-Manager mit gegebenen Sessions.""" + manager = MagicMock() + manager.get_sessions = MagicMock(return_value=sessions) + return manager + + +@pytest.mark.asyncio +async def test_pause_is_noop_when_smtc_unreachable(monkeypatch, caplog): + from whisper_local.media._smtc import SmtcController + + controller = SmtcController() + monkeypatch.setattr( + controller, + "_ensure_manager", + AsyncMock(side_effect=RuntimeError("kein SMTC")), + ) + + with caplog.at_level("WARNING"): + await controller.pause() + + assert controller._paused == [] + assert any("SMTC" in r.message or "smtc" in r.message.lower() for r in caplog.records) + + +@pytest.mark.asyncio +async def test_pause_skips_reconnect_after_smtc_failure(monkeypatch): + from whisper_local.media._smtc import SmtcController + + call_count = 0 + + async def failing_ensure(): + nonlocal call_count + call_count += 1 + raise RuntimeError("kein SMTC") + + controller = SmtcController() + monkeypatch.setattr(controller, "_ensure_manager", failing_ensure) + + await controller.pause() + await controller.pause() + await controller.pause() + + assert call_count == 1 +``` + +- [ ] **Schritt 2: Tests ausführen — müssen FAIL sein** + +```bash +uv run pytest tests/test_media_smtc.py -v +``` + +Erwartete Ausgabe: `ImportError` oder `ModuleNotFoundError` (Datei existiert noch nicht). + +- [ ] **Schritt 3: SmtcController-Skeleton implementieren** + +Datei `whisper_local/media/_smtc.py` erstellen: + +```python +"""Windows SMTC-Implementierung via pywinrt.""" + +import logging +from typing import Any + +logger = logging.getLogger(__name__) + + +class SmtcController: + def __init__(self) -> None: + self._paused: list[str] = [] + self._manager: Any = None + self._broken: bool = False + + async def _ensure_manager(self) -> Any: + if self._broken: + raise RuntimeError("SMTC nicht verfügbar") + if self._manager is None: + from winrt.windows.media.control import ( + GlobalSystemMediaTransportControlsSessionManager, + ) + self._manager = ( + await GlobalSystemMediaTransportControlsSessionManager.request_async() + ) + return self._manager + + async def pause(self) -> None: + try: + await self._ensure_manager() + except Exception as e: + if not self._broken: + logger.warning( + "SMTC nicht erreichbar, Media-Pause dauerhaft deaktiviert: %s", e + ) + self._broken = True + self._paused = [] + return + + async def resume(self) -> None: + pass +``` + +- [ ] **Schritt 4: Tests ausführen — müssen PASS sein** + +```bash +uv run pytest tests/test_media_smtc.py::test_pause_is_noop_when_smtc_unreachable tests/test_media_smtc.py::test_pause_skips_reconnect_after_smtc_failure -v +``` + +Erwartete Ausgabe: 2 passed + +- [ ] **Schritt 5: Committen** + +```bash +git add whisper_local/media/_smtc.py tests/test_media_smtc.py +git commit -m "feat(media): SmtcController Skeleton mit circuit-breaker" +``` + +--- + +## Task 3: pause() — Session-Erkennung und Pausieren (TDD) + +**Files:** +- Modify: `tests/test_media_smtc.py` +- Modify: `whisper_local/media/_smtc.py` + +- [ ] **Schritt 1: Tests für pause() ergänzen** + +Am Ende von `tests/test_media_smtc.py` anfügen: + +```python +@pytest.mark.asyncio +async def test_pause_with_no_sessions_is_noop(monkeypatch): + from whisper_local.media._smtc import SmtcController + + controller = SmtcController() + monkeypatch.setattr( + controller, "_ensure_manager", AsyncMock(return_value=_make_manager([])) + ) + + await controller.pause() + + assert controller._paused == [] + + +@pytest.mark.asyncio +async def test_pause_pauses_all_playing_sessions(monkeypatch): + from whisper_local.media._smtc import SmtcController + + s1 = _make_session("Spotify", PLAYING) + s2 = _make_session("msedge", PLAYING) + controller = SmtcController() + monkeypatch.setattr( + controller, + "_ensure_manager", + AsyncMock(return_value=_make_manager([s1, s2])), + ) + + await controller.pause() + + s1.try_pause_async.assert_awaited_once() + s2.try_pause_async.assert_awaited_once() + assert controller._paused == ["Spotify", "msedge"] + + +@pytest.mark.asyncio +async def test_pause_skips_already_paused_sessions(monkeypatch): + from whisper_local.media._smtc import SmtcController + + playing = _make_session("Spotify", PLAYING) + already_paused = _make_session("msedge", PAUSED) + controller = SmtcController() + monkeypatch.setattr( + controller, + "_ensure_manager", + AsyncMock(return_value=_make_manager([playing, already_paused])), + ) + + await controller.pause() + + playing.try_pause_async.assert_awaited_once() + already_paused.try_pause_async.assert_not_awaited() + assert controller._paused == ["Spotify"] + + +@pytest.mark.asyncio +async def test_pause_logs_and_continues_when_session_fails(monkeypatch, caplog): + from whisper_local.media._smtc import SmtcController + + broken = _make_session("broken", PLAYING) + broken.try_pause_async = AsyncMock(side_effect=RuntimeError("Verbindung verloren")) + good = _make_session("Spotify", PLAYING) + controller = SmtcController() + monkeypatch.setattr( + controller, + "_ensure_manager", + AsyncMock(return_value=_make_manager([broken, good])), + ) + + with caplog.at_level("WARNING"): + await controller.pause() + + good.try_pause_async.assert_awaited_once() + assert controller._paused == ["Spotify"] + assert any("broken" in r.message for r in caplog.records) +``` + +- [ ] **Schritt 2: Tests ausführen — müssen FAIL sein** + +```bash +uv run pytest tests/test_media_smtc.py::test_pause_with_no_sessions_is_noop tests/test_media_smtc.py::test_pause_pauses_all_playing_sessions tests/test_media_smtc.py::test_pause_skips_already_paused_sessions tests/test_media_smtc.py::test_pause_logs_and_continues_when_session_fails -v +``` + +Erwartete Ausgabe: 4 FAIL (pause() tut bislang nichts außer Manager holen). + +- [ ] **Schritt 3: pause() und _pause_session() implementieren** + +In `whisper_local/media/_smtc.py` die `pause()`-Methode und `_pause_session()` ersetzen/ergänzen: + +```python +"""Windows SMTC-Implementierung via pywinrt.""" + +import logging +from typing import Any + +logger = logging.getLogger(__name__) + + +class SmtcController: + def __init__(self) -> None: + self._paused: list[str] = [] + self._manager: Any = None + self._broken: bool = False + + async def _ensure_manager(self) -> Any: + if self._broken: + raise RuntimeError("SMTC nicht verfügbar") + if self._manager is None: + from winrt.windows.media.control import ( + GlobalSystemMediaTransportControlsSessionManager, + ) + self._manager = ( + await GlobalSystemMediaTransportControlsSessionManager.request_async() + ) + return self._manager + + async def _pause_session(self, session: Any) -> str | None: + """Pausiert eine Session wenn sie spielt. Gibt AUMID zurück, sonst None.""" + from winrt.windows.media.control import ( + GlobalSystemMediaTransportControlsSessionPlaybackStatus, + ) + aumid = session.source_app_user_model_id + try: + info = session.get_playback_info() + if ( + info.playback_status + != GlobalSystemMediaTransportControlsSessionPlaybackStatus.PLAYING + ): + return None + await session.try_pause_async() + return aumid + except Exception as e: + logger.warning("Konnte Session %s nicht pausieren: %s", aumid, e) + return None + + async def pause(self) -> None: + try: + manager = await self._ensure_manager() + except Exception as e: + if not self._broken: + logger.warning( + "SMTC nicht erreichbar, Media-Pause dauerhaft deaktiviert: %s", e + ) + self._broken = True + self._paused = [] + return + + sessions = list(manager.get_sessions()) + paused = [] + for session in sessions: + result = await self._pause_session(session) + if result is not None: + paused.append(result) + self._paused = paused + + async def resume(self) -> None: + pass +``` + +- [ ] **Schritt 4: Alle bisherigen Tests ausführen — müssen PASS sein** + +```bash +uv run pytest tests/test_media_smtc.py -v +``` + +Erwartete Ausgabe: 6 passed + +- [ ] **Schritt 5: Committen** + +```bash +git add whisper_local/media/_smtc.py tests/test_media_smtc.py +git commit -m "feat(media): SmtcController.pause() erkennt und pausiert PLAYING-Sessions" +``` + +--- + +## Task 4: resume() implementieren (TDD) + +**Files:** +- Modify: `tests/test_media_smtc.py` +- Modify: `whisper_local/media/_smtc.py` + +- [ ] **Schritt 1: Tests für resume() ergänzen** + +Am Ende von `tests/test_media_smtc.py` anfügen: + +```python +@pytest.mark.asyncio +async def test_resume_with_empty_paused_list_is_noop(monkeypatch): + from whisper_local.media._smtc import SmtcController + + controller = SmtcController() + controller._paused = [] + ensure = AsyncMock() + monkeypatch.setattr(controller, "_ensure_manager", ensure) + + await controller.resume() + + ensure.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_resume_plays_only_previously_paused(monkeypatch): + from whisper_local.media._smtc import SmtcController + + spotify = _make_session("Spotify", PAUSED) + edge = _make_session("msedge", PAUSED) + controller = SmtcController() + controller._paused = ["Spotify"] + monkeypatch.setattr( + controller, + "_ensure_manager", + AsyncMock(return_value=_make_manager([spotify, edge])), + ) + + await controller.resume() + + spotify.try_play_async.assert_awaited_once() + edge.try_play_async.assert_not_awaited() + assert controller._paused == [] + + +@pytest.mark.asyncio +async def test_resume_skips_disappeared_session(monkeypatch, caplog): + from whisper_local.media._smtc import SmtcController + + still_there = _make_session("Spotify", PAUSED) + controller = SmtcController() + controller._paused = ["gone_app", "Spotify"] + monkeypatch.setattr( + controller, + "_ensure_manager", + AsyncMock(return_value=_make_manager([still_there])), + ) + + with caplog.at_level("WARNING"): + await controller.resume() + + still_there.try_play_async.assert_awaited_once() + assert controller._paused == [] + assert any("gone_app" in r.message for r in caplog.records) + + +@pytest.mark.asyncio +async def test_resume_logs_and_continues_when_session_fails(monkeypatch, caplog): + from whisper_local.media._smtc import SmtcController + + broken = _make_session("broken", PAUSED) + broken.try_play_async = AsyncMock(side_effect=RuntimeError("Verbindung verloren")) + good = _make_session("Spotify", PAUSED) + controller = SmtcController() + controller._paused = ["broken", "Spotify"] + monkeypatch.setattr( + controller, + "_ensure_manager", + AsyncMock(return_value=_make_manager([broken, good])), + ) + + with caplog.at_level("WARNING"): + await controller.resume() + + good.try_play_async.assert_awaited_once() + assert controller._paused == [] + assert any("broken" in r.message for r in caplog.records) +``` + +- [ ] **Schritt 2: Tests ausführen — müssen FAIL sein** + +```bash +uv run pytest tests/test_media_smtc.py::test_resume_with_empty_paused_list_is_noop tests/test_media_smtc.py::test_resume_plays_only_previously_paused tests/test_media_smtc.py::test_resume_skips_disappeared_session tests/test_media_smtc.py::test_resume_logs_and_continues_when_session_fails -v +``` + +Erwartete Ausgabe: 4 FAIL (resume() ist noch ein No-Op). + +- [ ] **Schritt 3: resume() implementieren** + +In `whisper_local/media/_smtc.py` die `resume()`-Methode ersetzen: + +```python + async def resume(self) -> None: + if not self._paused: + return + try: + manager = await self._ensure_manager() + current = { + s.source_app_user_model_id: s for s in manager.get_sessions() + } + except Exception as e: + logger.warning("SMTC nicht erreichbar beim Fortsetzen: %s", e) + self._paused = [] + return + to_resume = self._paused + self._paused = [] + for aumid in to_resume: + session = current.get(aumid) + if session is None: + logger.warning( + "Session %s nicht mehr vorhanden, wird übersprungen", aumid + ) + continue + try: + await session.try_play_async() + except Exception as e: + logger.warning("Konnte Session %s nicht fortsetzen: %s", aumid, e) +``` + +- [ ] **Schritt 4: Alle Tests ausführen — müssen PASS sein** + +```bash +uv run pytest tests/test_media_smtc.py -v +``` + +Erwartete Ausgabe: 10 passed + +- [ ] **Schritt 5: Committen** + +```bash +git add whisper_local/media/_smtc.py tests/test_media_smtc.py +git commit -m "feat(media): SmtcController.resume() stellt nur eigene Pausen wieder her" +``` + +--- + +## Task 5: Factory-Dispatch für win32 + Protocol-Check (TDD) + +**Files:** +- Modify: `tests/test_media_factory.py` +- Modify: `whisper_local/media/__init__.py` + +- [ ] **Schritt 1: Factory-Tests aktualisieren** + +In `tests/test_media_factory.py`: + +1. Den bestehenden Test `test_factory_returns_noop_on_non_linux` umbenennen und auf `"darwin"` patchen (bisher war `"win32"` der Testfall — der hat jetzt einen eigenen Test): + +```python +def test_factory_returns_noop_on_other_platforms(): + with patch.object(sys, "platform", "darwin"): + controller = create_media_controller(enabled=True) + assert isinstance(controller, NoopController) +``` + +2. Neuen Test für win32 ergänzen (am Ende der Datei): + +```python +@pytest.mark.skipif(sys.platform != "win32", reason="SmtcController nur auf Windows") +def test_factory_returns_smtc_on_win32_when_enabled(): + from whisper_local.media._smtc import SmtcController + + controller = create_media_controller(enabled=True) + assert isinstance(controller, SmtcController) +``` + +- [ ] **Schritt 2: Tests ausführen — win32-Test muss FAIL sein** + +```bash +uv run pytest tests/test_media_factory.py -v +``` + +Erwartete Ausgabe: `test_factory_returns_smtc_on_win32_when_enabled` FAIL (Factory gibt noch `NoopController` zurück), alle anderen PASS. + +- [ ] **Schritt 3: Factory um win32-Dispatch erweitern** + +In `whisper_local/media/__init__.py` den `win32`-Zweig vor dem Noop-Fallback einfügen: + +```python +"""Media-Steuerung — plattformspezifische Backends hinter gemeinsamem Interface.""" + +import sys +from typing import Protocol, runtime_checkable + + +@runtime_checkable +class MediaController(Protocol): + async def pause(self) -> None: ... + async def resume(self) -> None: ... + + +def create_media_controller(enabled: bool) -> MediaController: + """Erstellt den plattformspezifischen Media-Controller. + + `enabled=False` → immer NoopController. Auf nicht unterstützten Plattformen + wird ebenfalls der NoopController zurückgegeben. + """ + if not enabled: + from whisper_local.media._noop import NoopController + return NoopController() + if sys.platform == "linux": + from whisper_local.media._mpris import MprisController + return MprisController() + if sys.platform == "win32": + from whisper_local.media._smtc import SmtcController + return SmtcController() + from whisper_local.media._noop import NoopController + return NoopController() +``` + +- [ ] **Schritt 4: Alle Factory-Tests ausführen — müssen PASS sein** + +```bash +uv run pytest tests/test_media_factory.py -v +``` + +Erwartete Ausgabe: alle passed + +- [ ] **Schritt 5: Committen** + +```bash +git add whisper_local/media/__init__.py tests/test_media_factory.py +git commit -m "feat(media): Factory dispatcht auf win32 zum SmtcController" +``` + +--- + +## Task 6: Gesamttest + Protocol-Konformität prüfen + +**Files:** keine Änderungen + +- [ ] **Schritt 1: Alle Tests ausführen** + +```bash +uv run pytest tests/test_media_smtc.py tests/test_media_factory.py -v +``` + +Erwartete Ausgabe: alle Tests grün, kein Skip außer den Linux-spezifischen (auf Windows). + +- [ ] **Schritt 2: Protocol-Konformität prüfen** + +```bash +uv run python -c " +from whisper_local.media import MediaController, create_media_controller +ctrl = create_media_controller(enabled=True) +assert isinstance(ctrl, MediaController), f'Protocol nicht erfüllt: {type(ctrl)}' +print('OK:', type(ctrl).__name__, 'erfüllt MediaController-Protocol') +" +``` + +Erwartete Ausgabe: `OK: SmtcController erfüllt MediaController-Protocol` + +- [ ] **Schritt 3: Vollständige Testsuite ausführen** + +```bash +uv run pytest -v +``` + +Erwartete Ausgabe: alle Tests grün (Linux-Tests werden auf Windows übersprungen).