docs: Implementation-Plan für Media-Pause während Aufnahme
9 Tasks in TDD-Reihenfolge: Config-Feld, Protocol/Factory/Noop, dbus-next Dependency, MprisController (pause, resume, Fehlertoleranz), App-Integration, Tray-Checkbox, Beispiel-Config. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,984 @@
|
||||
# Media-Pause während Aufnahme — 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:** Während einer Whisper-Aufnahme werden alle aktuell abspielenden MPRIS-Player (Browser-Video, Spotify, VLC, mpv, …) automatisch pausiert und am Ende der Aufnahme wieder fortgesetzt.
|
||||
|
||||
**Architecture:** Neues Modul `whisper_local/media/` mit `MediaController`-Protocol und Factory-Dispatch analog zu `inserter`/`hotkey`. Linux-Impl `MprisController` nutzt `dbus-next` (async, pure Python), Windows/Opt-out fällt auf `NoopController` zurück. `App.on_press` / `App.on_release` pausieren/resumen die Wiedergabe rund um `recorder.start()` / `recorder.stop()`.
|
||||
|
||||
**Tech Stack:** Python 3.13, asyncio, dbus-next (Linux-only), pytest/pytest-asyncio, tkinter (Settings-UI).
|
||||
|
||||
**Spec:** [docs/superpowers/specs/2026-04-14-media-pause-during-recording-design.md](../specs/2026-04-14-media-pause-during-recording-design.md)
|
||||
|
||||
---
|
||||
|
||||
## Datei-Übersicht
|
||||
|
||||
**Neu:**
|
||||
- `whisper_local/media/__init__.py` — Protocol `MediaController` + Factory `create_media_controller()`
|
||||
- `whisper_local/media/_noop.py` — `NoopController` (Windows, Opt-out)
|
||||
- `whisper_local/media/_mpris.py` — `MprisController` (Linux, D-Bus / MPRIS)
|
||||
- `tests/test_media_factory.py` — Factory-Dispatch-Tests
|
||||
- `tests/test_media_mpris.py` — MprisController-Verhalten mit gemocktem D-Bus
|
||||
|
||||
**Geändert:**
|
||||
- `whisper_local/config.py` — Feld `pause_media_during_recording: bool = True`, Laden/Speichern
|
||||
- `whisper_local/__main__.py` — `App`-Integration: Controller erstellen, in `on_press`/`on_release` verwenden, bei Config-Reload neu erstellen
|
||||
- `whisper_local/tray/_settings.py` — Neue Checkbox „Medienwiedergabe während Aufnahme pausieren"
|
||||
- `pyproject.toml` — Dependency `dbus-next ; sys_platform == "linux"`
|
||||
- `config.example.toml` — Neuer Abschnitt `[media]` (wenn vorhanden; sonst überspringen)
|
||||
|
||||
---
|
||||
|
||||
## Task 1: Config um `pause_media_during_recording` erweitern
|
||||
|
||||
**Files:**
|
||||
- Modify: `whisper_local/config.py`
|
||||
- Test: `tests/test_config.py` (erstellen falls nicht vorhanden)
|
||||
|
||||
- [ ] **Step 1: Failing Test schreiben**
|
||||
|
||||
Datei: `tests/test_config.py` — falls Datei existiert, die neuen Tests anhängen; falls nicht, komplett neu anlegen:
|
||||
|
||||
```python
|
||||
"""Tests für whisper_local.config."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from whisper_local.config import Config, load_config, save_config
|
||||
|
||||
|
||||
def test_default_pause_media_during_recording_is_true():
|
||||
config = Config()
|
||||
assert config.pause_media_during_recording is True
|
||||
|
||||
|
||||
def test_load_config_reads_pause_media_false(tmp_path: Path):
|
||||
cfg_path = tmp_path / "config.toml"
|
||||
cfg_path.write_text("[media]\npause_during_recording = false\n", encoding="utf-8")
|
||||
|
||||
config = load_config(cfg_path)
|
||||
|
||||
assert config.pause_media_during_recording is False
|
||||
|
||||
|
||||
def test_save_config_roundtrip_preserves_pause_media(tmp_path: Path):
|
||||
cfg_path = tmp_path / "config.toml"
|
||||
original = Config(pause_media_during_recording=False)
|
||||
|
||||
save_config(original, cfg_path)
|
||||
loaded = load_config(cfg_path)
|
||||
|
||||
assert loaded.pause_media_during_recording is False
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Tests laufen lassen — müssen fehlschlagen**
|
||||
|
||||
Run: `uv run pytest tests/test_config.py -v`
|
||||
Expected: FAIL — `Config` hat das Feld nicht, oder `load_config` / `save_config` kennen den Abschnitt nicht.
|
||||
|
||||
- [ ] **Step 3: `Config`-Dataclass-Feld ergänzen**
|
||||
|
||||
In `whisper_local/config.py` den `@dataclass Config` erweitern (Feld am Ende hinzufügen, Reihenfolge respektieren):
|
||||
|
||||
```python
|
||||
@dataclass
|
||||
class Config:
|
||||
hotkey: str = "KEY_F12"
|
||||
whisper_model: str = "small"
|
||||
language: str = "de"
|
||||
compute_type: str = "int8"
|
||||
sample_rate: int = 16000
|
||||
channels: int = 1
|
||||
min_duration: float = 0.5
|
||||
microphone: str = ""
|
||||
pause_media_during_recording: bool = True
|
||||
```
|
||||
|
||||
- [ ] **Step 4: `load_config` erweitern**
|
||||
|
||||
In `whisper_local/config.py` nach dem `audio_section`-Block ergänzen:
|
||||
|
||||
```python
|
||||
media_section = data.get("media", {})
|
||||
if "pause_during_recording" in media_section:
|
||||
config.pause_media_during_recording = bool(
|
||||
media_section["pause_during_recording"]
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] **Step 5: `save_config` erweitern**
|
||||
|
||||
In `whisper_local/config.py` den `content`-String in `save_config` erweitern — neuer Abschnitt am Ende:
|
||||
|
||||
```python
|
||||
def save_config(config: Config, path: Path = DEFAULT_CONFIG_PATH) -> None:
|
||||
"""Schreibt Config als TOML-Datei. Erstellt Verzeichnisse bei Bedarf."""
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
pause_media_str = "true" if config.pause_media_during_recording else "false"
|
||||
content = (
|
||||
f'[hotkey]\nkey = "{_toml_str(config.hotkey)}"\n\n'
|
||||
f'[whisper]\nmodel = "{_toml_str(config.whisper_model)}"\n'
|
||||
f'language = "{_toml_str(config.language)}"\n'
|
||||
f'compute_type = "{_toml_str(config.compute_type)}"\n\n'
|
||||
f'[audio]\nsample_rate = {config.sample_rate}\n'
|
||||
f'channels = {config.channels}\n'
|
||||
f'min_duration = {config.min_duration}\n'
|
||||
f'device = "{_toml_str(config.microphone)}"\n\n'
|
||||
f'[media]\npause_during_recording = {pause_media_str}\n'
|
||||
)
|
||||
path.write_text(content, encoding="utf-8")
|
||||
```
|
||||
|
||||
- [ ] **Step 6: Tests erneut laufen lassen — alle grün**
|
||||
|
||||
Run: `uv run pytest tests/test_config.py -v`
|
||||
Expected: PASS (3 neue Tests grün).
|
||||
|
||||
- [ ] **Step 7: Commit**
|
||||
|
||||
```bash
|
||||
git add whisper_local/config.py tests/test_config.py
|
||||
git commit -m "feat(config): pause_media_during_recording-Flag"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 2: Protocol, Noop-Controller und Factory
|
||||
|
||||
**Files:**
|
||||
- Create: `whisper_local/media/__init__.py`
|
||||
- Create: `whisper_local/media/_noop.py`
|
||||
- Create: `tests/test_media_factory.py`
|
||||
|
||||
- [ ] **Step 1: Failing Tests schreiben**
|
||||
|
||||
Datei: `tests/test_media_factory.py`:
|
||||
|
||||
```python
|
||||
"""Tests für whisper_local.media Factory-Dispatch."""
|
||||
|
||||
import sys
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from whisper_local.media import MediaController, create_media_controller
|
||||
from whisper_local.media._noop import NoopController
|
||||
|
||||
|
||||
def test_factory_returns_noop_when_disabled():
|
||||
controller = create_media_controller(enabled=False)
|
||||
assert isinstance(controller, NoopController)
|
||||
|
||||
|
||||
def test_factory_returns_noop_on_non_linux():
|
||||
with patch.object(sys, "platform", "win32"):
|
||||
controller = create_media_controller(enabled=True)
|
||||
assert isinstance(controller, NoopController)
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform != "linux", reason="MPRIS is Linux-only")
|
||||
def test_factory_returns_mpris_on_linux_when_enabled():
|
||||
from whisper_local.media._mpris import MprisController
|
||||
|
||||
controller = create_media_controller(enabled=True)
|
||||
assert isinstance(controller, MprisController)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_noop_controller_pause_is_noop():
|
||||
controller = NoopController()
|
||||
await controller.pause() # Darf nicht werfen
|
||||
await controller.resume() # Darf nicht werfen
|
||||
|
||||
|
||||
def test_noop_controller_satisfies_protocol():
|
||||
controller = NoopController()
|
||||
assert isinstance(controller, MediaController)
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Tests laufen lassen — müssen fehlschlagen**
|
||||
|
||||
Run: `uv run pytest tests/test_media_factory.py -v`
|
||||
Expected: FAIL — Modul `whisper_local.media` existiert nicht.
|
||||
|
||||
- [ ] **Step 3: `NoopController` implementieren**
|
||||
|
||||
Datei: `whisper_local/media/_noop.py`:
|
||||
|
||||
```python
|
||||
"""No-Op-Fallback für Plattformen ohne MPRIS oder wenn das Feature aus ist."""
|
||||
|
||||
|
||||
class NoopController:
|
||||
async def pause(self) -> None:
|
||||
return None
|
||||
|
||||
async def resume(self) -> None:
|
||||
return None
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Protocol + Factory implementieren**
|
||||
|
||||
Datei: `whisper_local/media/__init__.py`:
|
||||
|
||||
```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-Linux-Plattformen
|
||||
wird aktuell 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()
|
||||
from whisper_local.media._noop import NoopController
|
||||
return NoopController()
|
||||
```
|
||||
|
||||
- [ ] **Step 5: Platzhalter für `MprisController` anlegen**
|
||||
|
||||
Damit der Linux-Import nicht sofort bricht (die echte Implementierung kommt in den folgenden Tasks), Datei `whisper_local/media/_mpris.py` mit minimalem Stub anlegen:
|
||||
|
||||
```python
|
||||
"""Linux-MPRIS-Implementierung. Vollständige Logik folgt in späteren Tasks."""
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MprisController:
|
||||
def __init__(self) -> None:
|
||||
self._paused: list[str] = []
|
||||
|
||||
async def pause(self) -> None:
|
||||
return None
|
||||
|
||||
async def resume(self) -> None:
|
||||
return None
|
||||
```
|
||||
|
||||
- [ ] **Step 6: Tests laufen lassen — alle grün**
|
||||
|
||||
Run: `uv run pytest tests/test_media_factory.py -v`
|
||||
Expected: PASS (5 Tests grün; auf Nicht-Linux läuft der gate-skipte Test nicht).
|
||||
|
||||
- [ ] **Step 7: Commit**
|
||||
|
||||
```bash
|
||||
git add whisper_local/media/ tests/test_media_factory.py
|
||||
git commit -m "feat(media): Protocol, Factory und Noop-Controller"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 3: `dbus-next` als Linux-Dependency aufnehmen
|
||||
|
||||
**Files:**
|
||||
- Modify: `pyproject.toml`
|
||||
- Modify: `uv.lock` (automatisch durch `uv sync`)
|
||||
|
||||
- [ ] **Step 1: `pyproject.toml` erweitern**
|
||||
|
||||
Den `dependencies`-Block in `pyproject.toml` um einen Eintrag erweitern:
|
||||
|
||||
```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'",
|
||||
"pystray>=0.19.0",
|
||||
"Pillow>=10.0.0",
|
||||
"sv-ttk>=2.6.0",
|
||||
"darkdetect>=0.8.0",
|
||||
]
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Lockfile aktualisieren**
|
||||
|
||||
Run: `uv sync`
|
||||
Expected: `uv.lock` wird aktualisiert, `dbus-next` wird (unter Linux) installiert.
|
||||
|
||||
- [ ] **Step 3: Verfügbarkeit verifizieren**
|
||||
|
||||
Run: `uv run python -c "from dbus_next.aio import MessageBus; print('ok')"`
|
||||
Expected: `ok`. Falls ImportError: Dependency nicht installiert — `uv sync` erneut prüfen.
|
||||
|
||||
- [ ] **Step 4: Commit**
|
||||
|
||||
```bash
|
||||
git add pyproject.toml uv.lock
|
||||
git commit -m "build: dbus-next als Linux-Dependency hinzufügen"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 4: `MprisController.pause()` — Player finden und pausieren
|
||||
|
||||
**Files:**
|
||||
- Modify: `whisper_local/media/_mpris.py`
|
||||
- Create: `tests/test_media_mpris.py`
|
||||
|
||||
- [ ] **Step 1: Failing Test schreiben**
|
||||
|
||||
Datei: `tests/test_media_mpris.py`:
|
||||
|
||||
```python
|
||||
"""Tests für MprisController (Linux/MPRIS)."""
|
||||
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.skipif(sys.platform != "linux", reason="MPRIS is Linux-only")
|
||||
|
||||
|
||||
def _make_player(status: str) -> MagicMock:
|
||||
"""Erzeugt einen gemockten Player-Proxy mit gegebenem PlaybackStatus."""
|
||||
player = MagicMock()
|
||||
player.get_playback_status = AsyncMock(return_value=status)
|
||||
player.call_pause = AsyncMock()
|
||||
player.call_play = AsyncMock()
|
||||
return player
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pause_with_no_players_is_noop(monkeypatch):
|
||||
from whisper_local.media._mpris import MprisController
|
||||
|
||||
controller = MprisController()
|
||||
monkeypatch.setattr(
|
||||
controller, "_list_player_names", AsyncMock(return_value=[])
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
controller, "_get_player_interface", AsyncMock()
|
||||
)
|
||||
|
||||
await controller.pause()
|
||||
|
||||
assert controller._paused == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pause_pauses_only_playing_players(monkeypatch):
|
||||
from whisper_local.media._mpris import MprisController
|
||||
|
||||
playing = _make_player("Playing")
|
||||
paused = _make_player("Paused")
|
||||
|
||||
async def fake_get_player(name: str):
|
||||
return {"org.mpris.MediaPlayer2.spotify": playing,
|
||||
"org.mpris.MediaPlayer2.vlc": paused}[name]
|
||||
|
||||
controller = MprisController()
|
||||
monkeypatch.setattr(
|
||||
controller,
|
||||
"_list_player_names",
|
||||
AsyncMock(return_value=[
|
||||
"org.mpris.MediaPlayer2.spotify",
|
||||
"org.mpris.MediaPlayer2.vlc",
|
||||
]),
|
||||
)
|
||||
monkeypatch.setattr(controller, "_get_player_interface", fake_get_player)
|
||||
|
||||
await controller.pause()
|
||||
|
||||
playing.call_pause.assert_awaited_once()
|
||||
paused.call_pause.assert_not_awaited()
|
||||
assert controller._paused == ["org.mpris.MediaPlayer2.spotify"]
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Test laufen lassen — muss fehlschlagen**
|
||||
|
||||
Run: `uv run pytest tests/test_media_mpris.py -v`
|
||||
Expected: FAIL — `_list_player_names` / `_get_player_interface` gibt es noch nicht, `pause()` ist leer.
|
||||
|
||||
- [ ] **Step 3: `MprisController.pause()` implementieren**
|
||||
|
||||
`whisper_local/media/_mpris.py` komplett ersetzen:
|
||||
|
||||
```python
|
||||
"""Linux-MPRIS-Implementierung via dbus-next."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
MPRIS_PREFIX = "org.mpris.MediaPlayer2."
|
||||
MPRIS_PATH = "/org/mpris/MediaPlayer2"
|
||||
PLAYER_IFACE = "org.mpris.MediaPlayer2.Player"
|
||||
DBUS_SERVICE = "org.freedesktop.DBus"
|
||||
DBUS_PATH = "/org/freedesktop/DBus"
|
||||
DBUS_IFACE = "org.freedesktop.DBus"
|
||||
|
||||
|
||||
class MprisController:
|
||||
def __init__(self) -> None:
|
||||
self._paused: list[str] = []
|
||||
self._bus: Any = None
|
||||
|
||||
async def _ensure_bus(self) -> Any:
|
||||
if self._bus is None:
|
||||
from dbus_next.aio import MessageBus
|
||||
self._bus = await MessageBus().connect()
|
||||
return self._bus
|
||||
|
||||
async def _list_player_names(self) -> list[str]:
|
||||
bus = await self._ensure_bus()
|
||||
intro = await bus.introspect(DBUS_SERVICE, DBUS_PATH)
|
||||
obj = bus.get_proxy_object(DBUS_SERVICE, DBUS_PATH, intro)
|
||||
iface = obj.get_interface(DBUS_IFACE)
|
||||
names = await iface.call_list_names()
|
||||
return [n for n in names if n.startswith(MPRIS_PREFIX)]
|
||||
|
||||
async def _get_player_interface(self, bus_name: str) -> Any:
|
||||
bus = await self._ensure_bus()
|
||||
intro = await bus.introspect(bus_name, MPRIS_PATH)
|
||||
obj = bus.get_proxy_object(bus_name, MPRIS_PATH, intro)
|
||||
return obj.get_interface(PLAYER_IFACE)
|
||||
|
||||
async def _pause_player(self, bus_name: str) -> str | None:
|
||||
"""Pausiert einen Player, wenn er im Status 'Playing' ist.
|
||||
Gibt den Bus-Namen zurück, wenn pausiert wurde, sonst None.
|
||||
"""
|
||||
player = await self._get_player_interface(bus_name)
|
||||
status = await player.get_playback_status()
|
||||
if status != "Playing":
|
||||
return None
|
||||
await player.call_pause()
|
||||
return bus_name
|
||||
|
||||
async def pause(self) -> None:
|
||||
names = await self._list_player_names()
|
||||
results = await asyncio.gather(
|
||||
*(self._pause_player(n) for n in names),
|
||||
return_exceptions=True,
|
||||
)
|
||||
self._paused = [
|
||||
name for name in results
|
||||
if isinstance(name, str)
|
||||
]
|
||||
|
||||
async def resume(self) -> None:
|
||||
return None
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Test laufen lassen — grün**
|
||||
|
||||
Run: `uv run pytest tests/test_media_mpris.py -v`
|
||||
Expected: PASS (2 Tests grün).
|
||||
|
||||
- [ ] **Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add whisper_local/media/_mpris.py tests/test_media_mpris.py
|
||||
git commit -m "feat(media): MprisController.pause() via dbus-next"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 5: `MprisController.resume()` — nur eigene Pausen zurücknehmen
|
||||
|
||||
**Files:**
|
||||
- Modify: `whisper_local/media/_mpris.py`
|
||||
- Modify: `tests/test_media_mpris.py`
|
||||
|
||||
- [ ] **Step 1: Failing Tests schreiben**
|
||||
|
||||
An `tests/test_media_mpris.py` anhängen:
|
||||
|
||||
```python
|
||||
@pytest.mark.asyncio
|
||||
async def test_resume_plays_only_previously_paused(monkeypatch):
|
||||
from whisper_local.media._mpris import MprisController
|
||||
|
||||
spotify = _make_player("Paused")
|
||||
vlc = _make_player("Paused")
|
||||
|
||||
async def fake_get_player(name: str):
|
||||
return {"org.mpris.MediaPlayer2.spotify": spotify,
|
||||
"org.mpris.MediaPlayer2.vlc": vlc}[name]
|
||||
|
||||
controller = MprisController()
|
||||
controller._paused = ["org.mpris.MediaPlayer2.spotify"]
|
||||
monkeypatch.setattr(controller, "_get_player_interface", fake_get_player)
|
||||
|
||||
await controller.resume()
|
||||
|
||||
spotify.call_play.assert_awaited_once()
|
||||
vlc.call_play.assert_not_awaited()
|
||||
assert controller._paused == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resume_with_empty_paused_list_is_noop(monkeypatch):
|
||||
from whisper_local.media._mpris import MprisController
|
||||
|
||||
controller = MprisController()
|
||||
controller._paused = []
|
||||
get_player = AsyncMock()
|
||||
monkeypatch.setattr(controller, "_get_player_interface", get_player)
|
||||
|
||||
await controller.resume()
|
||||
|
||||
get_player.assert_not_awaited()
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Tests laufen lassen — müssen fehlschlagen**
|
||||
|
||||
Run: `uv run pytest tests/test_media_mpris.py -v`
|
||||
Expected: FAIL — `resume()` ruft noch nichts auf.
|
||||
|
||||
- [ ] **Step 3: `resume()` implementieren**
|
||||
|
||||
In `whisper_local/media/_mpris.py` die Methode `resume` ersetzen und eine Hilfsmethode `_resume_player` ergänzen:
|
||||
|
||||
```python
|
||||
async def _resume_player(self, bus_name: str) -> None:
|
||||
player = await self._get_player_interface(bus_name)
|
||||
await player.call_play()
|
||||
|
||||
async def resume(self) -> None:
|
||||
if not self._paused:
|
||||
return
|
||||
to_resume = self._paused
|
||||
self._paused = []
|
||||
await asyncio.gather(
|
||||
*(self._resume_player(n) for n in to_resume),
|
||||
return_exceptions=True,
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Tests laufen lassen — alle grün**
|
||||
|
||||
Run: `uv run pytest tests/test_media_mpris.py -v`
|
||||
Expected: PASS (4 Tests grün).
|
||||
|
||||
- [ ] **Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add whisper_local/media/_mpris.py tests/test_media_mpris.py
|
||||
git commit -m "feat(media): MprisController.resume() stellt nur eigene Pausen wieder her"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 6: Fehlertoleranz — verschwindende Player und D-Bus nicht erreichbar
|
||||
|
||||
**Files:**
|
||||
- Modify: `whisper_local/media/_mpris.py`
|
||||
- Modify: `tests/test_media_mpris.py`
|
||||
|
||||
- [ ] **Step 1: Failing Tests schreiben**
|
||||
|
||||
An `tests/test_media_mpris.py` anhängen:
|
||||
|
||||
```python
|
||||
@pytest.mark.asyncio
|
||||
async def test_pause_logs_and_continues_when_single_player_fails(monkeypatch, caplog):
|
||||
from whisper_local.media._mpris import MprisController
|
||||
|
||||
good = _make_player("Playing")
|
||||
|
||||
async def fake_get_player(name: str):
|
||||
if name.endswith("broken"):
|
||||
raise RuntimeError("player disappeared")
|
||||
return good
|
||||
|
||||
controller = MprisController()
|
||||
monkeypatch.setattr(
|
||||
controller,
|
||||
"_list_player_names",
|
||||
AsyncMock(return_value=[
|
||||
"org.mpris.MediaPlayer2.broken",
|
||||
"org.mpris.MediaPlayer2.good",
|
||||
]),
|
||||
)
|
||||
monkeypatch.setattr(controller, "_get_player_interface", fake_get_player)
|
||||
|
||||
with caplog.at_level("WARNING"):
|
||||
await controller.pause()
|
||||
|
||||
good.call_pause.assert_awaited_once()
|
||||
assert controller._paused == ["org.mpris.MediaPlayer2.good"]
|
||||
assert any("broken" in r.message for r in caplog.records)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resume_logs_and_continues_when_single_player_fails(monkeypatch, caplog):
|
||||
from whisper_local.media._mpris import MprisController
|
||||
|
||||
good = _make_player("Paused")
|
||||
|
||||
async def fake_get_player(name: str):
|
||||
if name.endswith("broken"):
|
||||
raise RuntimeError("player disappeared")
|
||||
return good
|
||||
|
||||
controller = MprisController()
|
||||
controller._paused = [
|
||||
"org.mpris.MediaPlayer2.broken",
|
||||
"org.mpris.MediaPlayer2.good",
|
||||
]
|
||||
monkeypatch.setattr(controller, "_get_player_interface", fake_get_player)
|
||||
|
||||
with caplog.at_level("WARNING"):
|
||||
await controller.resume()
|
||||
|
||||
good.call_play.assert_awaited_once()
|
||||
assert controller._paused == []
|
||||
assert any("broken" in r.message for r in caplog.records)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pause_is_noop_when_bus_unreachable(monkeypatch, caplog):
|
||||
from whisper_local.media._mpris import MprisController
|
||||
|
||||
async def failing_connect(self):
|
||||
raise RuntimeError("no session bus")
|
||||
|
||||
monkeypatch.setattr(
|
||||
"dbus_next.aio.MessageBus.connect", failing_connect, raising=False
|
||||
)
|
||||
controller = MprisController()
|
||||
|
||||
with caplog.at_level("WARNING"):
|
||||
await controller.pause()
|
||||
|
||||
assert controller._paused == []
|
||||
assert any("D-Bus" in r.message or "bus" in r.message.lower()
|
||||
for r in caplog.records)
|
||||
```
|
||||
|
||||
- [ ] **Step 2: Tests laufen lassen — müssen fehlschlagen**
|
||||
|
||||
Run: `uv run pytest tests/test_media_mpris.py -v`
|
||||
Expected: FAIL — noch kein Logging für Player-Fehler, noch kein Swallow der Bus-Verbindungs-Exception.
|
||||
|
||||
- [ ] **Step 3: `_pause_player` / `_resume_player` mit Logging absichern**
|
||||
|
||||
In `whisper_local/media/_mpris.py` die Methoden `_pause_player` und `_resume_player` ersetzen und `pause` / `resume` für Bus-Fehler absichern:
|
||||
|
||||
```python
|
||||
async def _pause_player(self, bus_name: str) -> str | None:
|
||||
try:
|
||||
player = await self._get_player_interface(bus_name)
|
||||
status = await player.get_playback_status()
|
||||
if status != "Playing":
|
||||
return None
|
||||
await player.call_pause()
|
||||
return bus_name
|
||||
except Exception as e:
|
||||
logger.warning("Konnte Player %s nicht pausieren: %s", bus_name, e)
|
||||
return None
|
||||
|
||||
async def _resume_player(self, bus_name: str) -> None:
|
||||
try:
|
||||
player = await self._get_player_interface(bus_name)
|
||||
await player.call_play()
|
||||
except Exception as e:
|
||||
logger.warning("Konnte Player %s nicht fortsetzen: %s", bus_name, e)
|
||||
|
||||
async def pause(self) -> None:
|
||||
try:
|
||||
names = await self._list_player_names()
|
||||
except Exception as e:
|
||||
logger.warning("D-Bus nicht erreichbar, überspringe Media-Pause: %s", e)
|
||||
self._paused = []
|
||||
return
|
||||
results = await asyncio.gather(
|
||||
*(self._pause_player(n) for n in names),
|
||||
return_exceptions=True,
|
||||
)
|
||||
self._paused = [name for name in results if isinstance(name, str)]
|
||||
|
||||
async def resume(self) -> None:
|
||||
if not self._paused:
|
||||
return
|
||||
to_resume = self._paused
|
||||
self._paused = []
|
||||
await asyncio.gather(
|
||||
*(self._resume_player(n) for n in to_resume),
|
||||
return_exceptions=True,
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Tests laufen lassen — alle grün**
|
||||
|
||||
Run: `uv run pytest tests/test_media_mpris.py -v`
|
||||
Expected: PASS (7 Tests grün).
|
||||
|
||||
- [ ] **Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add whisper_local/media/_mpris.py tests/test_media_mpris.py
|
||||
git commit -m "feat(media): MprisController fängt Player- und Bus-Fehler sauber ab"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 7: App-Integration in `__main__.py`
|
||||
|
||||
**Files:**
|
||||
- Modify: `whisper_local/__main__.py`
|
||||
|
||||
- [ ] **Step 1: Import und Controller-Erstellung in `App.__init__`**
|
||||
|
||||
In `whisper_local/__main__.py` den Import-Block oben ergänzen:
|
||||
|
||||
```python
|
||||
from whisper_local.media import create_media_controller
|
||||
```
|
||||
|
||||
Und in `App.__init__` — nach `self.inserter = create_inserter()` und vor `self.hotkey = create_listener(...)`:
|
||||
|
||||
```python
|
||||
self.media = create_media_controller(
|
||||
enabled=config.pause_media_during_recording
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] **Step 2: `on_press` erweitern**
|
||||
|
||||
Die `on_press`-Methode in `whisper_local/__main__.py` ersetzen:
|
||||
|
||||
```python
|
||||
async def on_press(self) -> None:
|
||||
"""Callback: Hotkey gedrückt — Medien pausieren + Aufnahme starten."""
|
||||
logger.info("Aufnahme startet...")
|
||||
self.tray.set_state(AppState.RECORDING)
|
||||
await self.media.pause()
|
||||
self.recorder.start()
|
||||
```
|
||||
|
||||
- [ ] **Step 3: `on_release` erweitern (Resume via try/finally)**
|
||||
|
||||
Die `on_release`-Methode in `whisper_local/__main__.py` ersetzen:
|
||||
|
||||
```python
|
||||
async def on_release(self) -> None:
|
||||
"""Callback: Hotkey losgelassen — Aufnahme stoppen, Medien fortsetzen, transkribieren, einfügen."""
|
||||
try:
|
||||
audio = self.recorder.stop()
|
||||
finally:
|
||||
try:
|
||||
await self.media.resume()
|
||||
except Exception as e:
|
||||
logger.warning("Fehler beim Fortsetzen der Medienwiedergabe: %s", e)
|
||||
|
||||
if audio is None:
|
||||
logger.info("Keine Audio-Daten, übersprungen")
|
||||
self.tray.set_state(AppState.WAITING)
|
||||
return
|
||||
|
||||
logger.info("Transkribiere...")
|
||||
self.tray.set_state(AppState.TRANSCRIBING)
|
||||
text = self.transcriber.transcribe(audio)
|
||||
if text:
|
||||
await self.inserter.insert(text)
|
||||
self.tray.set_state(AppState.WAITING)
|
||||
```
|
||||
|
||||
- [ ] **Step 4: Controller bei Config-Reload neu erstellen**
|
||||
|
||||
Die `_on_config_reload`-Methode in `whisper_local/__main__.py` ersetzen:
|
||||
|
||||
```python
|
||||
def _on_config_reload(self, new_config: Config) -> None:
|
||||
"""Übernimmt neue Konfiguration ohne App-Neustart."""
|
||||
self._config = new_config
|
||||
self.recorder = Recorder(
|
||||
sample_rate=new_config.sample_rate,
|
||||
channels=new_config.channels,
|
||||
min_duration=new_config.min_duration,
|
||||
device=new_config.microphone or None,
|
||||
)
|
||||
self.media = create_media_controller(
|
||||
enabled=new_config.pause_media_during_recording
|
||||
)
|
||||
if self._loop is not None:
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._restart_hotkey(new_config.hotkey), self._loop
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] **Step 5: Bestehende Tests laufen lassen — nichts gebrochen**
|
||||
|
||||
Run: `uv run pytest -v`
|
||||
Expected: PASS — alle bisherigen Tests weiterhin grün, neue Media-Tests weiterhin grün.
|
||||
|
||||
- [ ] **Step 6: Manueller Smoke-Test (Linux mit MPRIS-Player)**
|
||||
|
||||
Falls ein MPRIS-Player verfügbar ist (Spotify / Firefox mit Video / VLC):
|
||||
|
||||
1. Player starten und Wiedergabe starten.
|
||||
2. `uv run whisper-local` starten.
|
||||
3. Hotkey halten — Wiedergabe sollte pausieren.
|
||||
4. Hotkey loslassen — Wiedergabe sollte sofort fortsetzen.
|
||||
5. Während Transkription läuft, darf die Musik nicht wieder abbrechen.
|
||||
|
||||
Falls kein Player verfügbar: Schritt überspringen, `logger.info` im Log beobachten (keine Fehler).
|
||||
|
||||
- [ ] **Step 7: Commit**
|
||||
|
||||
```bash
|
||||
git add whisper_local/__main__.py
|
||||
git commit -m "feat(app): Medien pausieren bei Aufnahmestart, fortsetzen bei Stopp"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 8: Tray-Settings-Checkbox
|
||||
|
||||
**Files:**
|
||||
- Modify: `whisper_local/tray/_settings.py`
|
||||
|
||||
- [ ] **Step 1: `SettingsDialog._run` um Checkbox erweitern**
|
||||
|
||||
In `whisper_local/tray/_settings.py` — im Aufbau des Dialogs nach dem Mikrofon-Block, **vor** dem Button-Block, ergänzen:
|
||||
|
||||
```python
|
||||
# --- Medien-Pause ---
|
||||
pause_media_var = tk.BooleanVar(
|
||||
value=self._config.pause_media_during_recording
|
||||
)
|
||||
ttk.Checkbutton(
|
||||
frame,
|
||||
text="Medienwiedergabe während Aufnahme pausieren",
|
||||
variable=pause_media_var,
|
||||
).grid(row=3, column=0, columnspan=3, sticky=tk.W, pady=4)
|
||||
```
|
||||
|
||||
Und den Button-Block-Row-Index anpassen — `btn_frame.grid(...)` nutzt aktuell `row=3`; das muss auf `row=4`:
|
||||
|
||||
```python
|
||||
btn_frame = ttk.Frame(frame)
|
||||
btn_frame.grid(row=4, column=0, columnspan=3, pady=12, sticky=tk.E)
|
||||
```
|
||||
|
||||
- [ ] **Step 2: `save()` anpassen**
|
||||
|
||||
In `whisper_local/tray/_settings.py` die lokale Funktion `save` innerhalb von `_run` ersetzen:
|
||||
|
||||
```python
|
||||
def save():
|
||||
new_config = Config(
|
||||
hotkey=hotkey_var.get(),
|
||||
whisper_model=self._config.whisper_model,
|
||||
language=self._config.language,
|
||||
compute_type=self._config.compute_type,
|
||||
sample_rate=self._config.sample_rate,
|
||||
channels=self._config.channels,
|
||||
min_duration=self._config.min_duration,
|
||||
microphone="" if mic_var.get() == "Standard" else mic_var.get(),
|
||||
pause_media_during_recording=pause_media_var.get(),
|
||||
)
|
||||
save_config(new_config)
|
||||
self._on_save(new_config)
|
||||
self._cancel_event.set()
|
||||
root.destroy()
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Manueller Smoke-Test**
|
||||
|
||||
Run: `uv run whisper-local`, Tray-Icon rechtsklicken → Einstellungen → Checkbox sichtbar, umschaltbar, Speichern schreibt Wert in die Config-Datei.
|
||||
|
||||
Verifizieren:
|
||||
|
||||
```bash
|
||||
cat ~/.config/whisper-local/config.toml
|
||||
```
|
||||
|
||||
Expected: Abschnitt `[media]\npause_during_recording = true|false\n` entsprechend der Wahl vorhanden.
|
||||
|
||||
- [ ] **Step 4: Bestehende Tests laufen lassen**
|
||||
|
||||
Run: `uv run pytest -v`
|
||||
Expected: PASS — alles weiterhin grün.
|
||||
|
||||
- [ ] **Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add whisper_local/tray/_settings.py
|
||||
git commit -m "feat(settings): Checkbox für Medien-Pause während Aufnahme"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Task 9: Beispiel-Config aktualisieren
|
||||
|
||||
**Files:**
|
||||
- Modify: `config.example.toml`
|
||||
|
||||
- [ ] **Step 1: Prüfen, ob Datei existiert**
|
||||
|
||||
Run: `ls config.example.toml`
|
||||
|
||||
Falls Datei nicht existiert: Task überspringen und direkt zum nächsten Task (oder Abschluss).
|
||||
|
||||
- [ ] **Step 2: Abschnitt `[media]` ergänzen**
|
||||
|
||||
Am Ende von `config.example.toml` anfügen:
|
||||
|
||||
```toml
|
||||
|
||||
[media]
|
||||
# Pausiert MPRIS-Medienplayer (Spotify, VLC, Browser-Video, …) während
|
||||
# einer Aufnahme und setzt sie danach wieder fort. Nur Linux.
|
||||
pause_during_recording = true
|
||||
```
|
||||
|
||||
- [ ] **Step 3: Commit**
|
||||
|
||||
```bash
|
||||
git add config.example.toml
|
||||
git commit -m "docs: [media]-Abschnitt in config.example.toml"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Abschluss-Check
|
||||
|
||||
- [ ] **Alle Tests laufen**
|
||||
|
||||
Run: `uv run pytest -v`
|
||||
Expected: Alle Tests grün, keine Warnings bezüglich unhandled exceptions / unawaited coroutines.
|
||||
|
||||
- [ ] **Manueller End-to-End-Test (Linux)**
|
||||
|
||||
1. Musik/Video in einem MPRIS-Player starten.
|
||||
2. `uv run whisper-local`.
|
||||
3. Hotkey halten, etwas einsprechen, loslassen.
|
||||
4. Beobachten: Musik pausiert beim Drücken, startet beim Loslassen.
|
||||
5. Eingefügter Text erscheint im aktiven Fenster.
|
||||
6. Einstellungen öffnen, Checkbox deaktivieren, Speichern.
|
||||
7. Hotkey erneut halten — Musik bleibt jetzt an.
|
||||
|
||||
- [ ] **Dokumentation prüfen**
|
||||
|
||||
Spec-Dokument unter `docs/superpowers/specs/2026-04-14-media-pause-during-recording-design.md` ist aktuell, keine Abweichungen zwischen Spec und Implementierung.
|
||||
Reference in New Issue
Block a user