695 lines
21 KiB
Markdown
695 lines
21 KiB
Markdown
|
# 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).
|