nc/linux-voice-assistant
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TTS does not lower mpv music volume while speaking

Browse Source 
- Fixes #126
- callback was missing
This commit is contained in:
Philipp Grafe committed 2026-02-08 18:29:40 +01:00
1 parent 9b52f61a9f
commit f5b76b8921
2 files changed
+90 -68
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

linux_voice_assistant/mpv_player.py
+33 -8
View File
@@ -1,4 +1,4 @@
import logging
# mpv_player.py
from typing import Union, List, Callable, Optional
from .player.libmpv import LibMpvPlayer
@@ -6,11 +6,13 @@ from .player.libmpv import LibMpvPlayer
class MpvMediaPlayer:
"""
LVA MediaPlayer implementation using libmpv.
Linux Voice Assistant MediaPlayer implementation based on libmpv.
This class provides the MediaPlayer interface expected by LVA and
delegates all playback logic to LibMpvPlayer.
"""
def __init__(self, device: str | None = None):
self._log = logging.getLogger(self.__class__.__name__)
def __init__(self, device: str | None = None) -> None:
self._player = LibMpvPlayer(device=device)
self._done_callback: Optional[Callable[[], None]] = None
@@ -18,33 +20,56 @@ class MpvMediaPlayer:
self,
url: Union[str, List[str]],
done_callback: Optional[Callable[[], None]] = None,
stop_first: bool = True,
stop_first: bool = False,
) -> None:
self._log.info("Playing %s", url)
"""
Play a media URL.
Args:
url: Media URL or list of URLs (LVA currently uses a single URL).
done_callback: Optional callback invoked when playback finishes.
stop_first: Kept for API compatibility; currently unused.
"""
# LVA currently only uses single URLs
if isinstance(url, list):
url = url[0]
self._done_callback = done_callback
self._player.play(url, paused=False)
self._player.play(url, done_callback=done_callback, stop_first=stop_first)
def pause(self) -> None:
"""Pause playback."""
self._player.pause()
def resume(self) -> None:
"""Resume playback."""
self._player.resume()
def stop(self) -> None:
"""Stop playback and invoke the done callback if present."""
self._player.stop()
if self._done_callback:
self._done_callback()
self._done_callback = None
def set_volume(self, volume: float) -> None:
"""
Set playback volume.
Args:
volume: Volume in percent (0.0–100.0).
"""
self._player.set_volume(volume)
def duck(self, factor: float = 0.5) -> None:
"""
Temporarily reduce volume.
Args:
factor: Volume multiplier (0.0–1.0).
"""
self._player.duck(factor)
def unduck(self) -> None:
self._player.unduck()
"""Restore volume after ducking."""
self._player.unduck()
linux_voice_assistant/player/libmpv.py
+57 -60
View File
@@ -1,6 +1,6 @@
import threading
import logging
from typing import Optional
from typing import Optional, Callable
import mpv
from .base import AudioPlayer
@@ -9,28 +9,23 @@ from .state import PlayerState
class LibMpvPlayer(AudioPlayer):
"""
Audio player implementation for Linux Voice Assistant based on libmpv.
AudioPlayer implementation for Linux Voice Assistant using libmpv.
Features:
- Thread-safe state handling
- Explicit pause / resume control
- Proper volume handling with ducking support
- Compatible with LVA AudioPlayer interface
Responsibilities:
- mpv lifecycle and playback control
- thread-safe state management
- volume handling with ducking support
"""
def __init__(self, device: Optional[str] = None) -> None:
"""
Initialize the mpv-based audio player.
:param device: Optional mpv audio output device name
"""
self._log = logging.getLogger(self.__class__.__name__)
self._state: PlayerState = PlayerState.IDLE
self._state_lock = threading.Lock()
# Volume handling
self._user_volume: float = 100.0 # User volume (0.0 – 100.0)
self._duck_factor: float = 1.0 # Ducking factor (0.0 – 1.0)
self._user_volume: float = 100.0 # 0.0 – 100.0
self._duck_factor: float = 1.0 # 0.0 – 1.0
self._done_callback: Optional[Callable[[], None]] = None
# mpv setup
self._mpv = mpv.MPV(
@@ -40,59 +35,53 @@ class LibMpvPlayer(AudioPlayer):
)
if device:
self._log.info("Using audio device: %s", device)
self._mpv["audio-device"] = device
# -------- Core Playback Methods --------
self._mpv.event_callback("end-file")(self._on_end_file)
def play(self, url: str, paused: bool = False) -> None:
# -------- Playback control --------
def play(
self,
url: str,
done_callback: Optional[Callable[[], None]] = None,
stop_first: bool = True,
) -> None:
"""
Start playback of a media URL.
:param url: Media URL or file path
:param paused: If True, playback starts paused
Args:
url: Media URL or local file path.
done_callback: Optional callback invoked when playback finishes.
stop_first: If True, start playback in paused state.
"""
with self._state_lock:
self._duck_factor = 1.0 # auto-unduck on new playback
self._apply_volume()
self._set_state(PlayerState.LOADING)
self._mpv.pause = paused
self._log.info("Loading media: %s (paused=%s)", url, paused)
self._done_callback = done_callback
self._mpv.pause = stop_first
self._mpv.play(url)
def pause(self) -> None:
"""
Pause playback.
"""
"""Pause playback."""
with self._state_lock:
self._mpv.pause = True
self._set_state(PlayerState.PAUSED)
def resume(self) -> None:
"""
Resume playback if paused.
"""
"""Resume playback if paused."""
with self._state_lock:
self._duck_factor = 1.0 # ensure unduck
self._apply_volume()
self._mpv.pause = False
self._set_state(PlayerState.PLAYING)
def stop(self) -> None:
"""
Stop playback and reset player state to IDLE.
"""
"""Stop playback and reset state to IDLE."""
with self._state_lock:
self._mpv.stop()
self._set_state(PlayerState.IDLE)
def state(self) -> PlayerState:
"""
Get the current player state.
:return: Current PlayerState
"""
"""Return the current player state."""
with self._state_lock:
return self._state
@@ -100,9 +89,10 @@ class LibMpvPlayer(AudioPlayer):
def set_volume(self, volume: float) -> None:
"""
Set the user volume.
Set user volume.
:param volume: Volume level (0.0 – 100.0)
Args:
volume: Volume level (0.0–100.0).
"""
with self._state_lock:
self._user_volume = max(0.0, min(100.0, float(volume)))
@@ -110,46 +100,53 @@ class LibMpvPlayer(AudioPlayer):
def duck(self, factor: float = 0.5) -> None:
"""
Temporarily reduce volume by a ducking factor.
Reduce volume temporarily by a ducking factor.
:param factor: Ducking factor (0.0 – 1.0)
Args:
factor: Ducking factor (0.0–1.0).
"""
with self._state_lock:
self._duck_factor = max(0.0, min(1.0, float(factor)))
self._apply_volume()
def unduck(self) -> None:
"""
Restore volume to the user-defined level.
"""
"""Restore volume to the user-defined level."""
with self._state_lock:
self._duck_factor = 1.0
self._apply_volume()
# -------- Internal Helpers --------
# -------- Internal helpers --------
def _apply_volume(self) -> None:
"""
Apply effective volume (user volume × duck factor) to mpv.
"""
"""Apply effective volume (user volume × duck factor) to mpv."""
effective = self._user_volume * self._duck_factor
self._mpv.volume = max(0.0, min(100.0, effective))
def _on_end_file(self, event) -> None:
callback: Optional[Callable[[], None]] = None
with self._state_lock:
self._set_state(PlayerState.IDLE)
callback = self._done_callback
self._done_callback = None
if callback is not None:
try:
callback()
except RuntimeError:
# Callback errors must never break the player
pass
def _on_mpv_log(self, level: str, prefix: str, text: str) -> None:
"""
Handle mpv log messages.
Errors and fatal messages transition the player into ERROR state.
Error and fatal messages transition the player into ERROR state.
"""
if level in ("error", "fatal"):
self._log.error("[mpv] %s", text.strip())
with self._state_lock:
self._set_state(PlayerState.ERROR)
def _set_state(self, new_state: PlayerState) -> None:
"""
Update internal player state with logging.
"""
if self._state != new_state:
self._log.debug("State %s → %s", self._state.name, new_state.name)
self._state = new_state
"""Update internal player state."""
self._state = new_state