linux-voice-assistant/linux_voice_assistant/mpv_player.py

"""
Media player using mpv in a subprocess.
Includes logic to detect the audio server being used (Pirewire, PulseAudio, Alsa).
This refactored version simplifies the audio backend selection logic,
improves readability with docstrings, and standardizes method signatures.
"""
from __future__ import annotations

import logging
import os
from threading import Lock, Timer
from typing import Callable, Iterable, List, Optional, Sequence, Tuple, Union

from mpv import MPV

_LOGGER = logging.getLogger(__name__)

# -----------------------------------------------------------------------------

def _set_player_opt(player: MPV, key: str, value) -> bool:
    """Safely set an mpv option, logging any errors."""
    try:
        player[key] = value
        return True
    except Exception:
        _LOGGER.warning("Failed to set mpv option %r=%r", key, value, exc_info=True)
        return False

def _select_backend(player: MPV, device: Optional[str]) -> None:
    """
    Selects the best available audio output backend for mpv.
    
    The selection follows a clear priority:
    1. A fully qualified device string (e.g., 'alsa/plughw:...') is used directly.
    2. A simple device name tries PipeWire, then PulseAudio, then ALSA.
    3. No device specified (or 'default') tries the default for PipeWire, PulseAudio, and ALSA in order.
    """
    candidates: List[Tuple[str, str]] = []
    
    # Define the preferred order of audio outputs
    preferred_ao_order = ["pipewire", "pulse", "alsa"]

    if device:
        # If a full device path is given (e.g., "alsa/plughw:...")
        for prefix in preferred_ao_order:
            if device.startswith(f"{prefix}/"):
                candidates.append((prefix, device))
                break
        
        # If it's a simple name or "default"
        if not candidates:
            for ao in preferred_ao_order:
                candidates.append((ao, f"{ao}/{device}"))
    else:
        # If no device is specified, try the default for each backend
        for ao in preferred_ao_order:
            candidates.append((ao, f"{ao}/default"))

    # Try to set the audio output from the generated candidates
    for ao, audio_device in candidates:
        try:
            player["ao"] = ao
            player["audio-device"] = audio_device
            _LOGGER.debug("mpv backend selected: ao=%s, audio-device=%s", ao, audio_device)
            return
        except Exception:
            continue
            
    _LOGGER.warning("No suitable audio output backend could be found.")


class MpvMediaPlayer:
    """A media player class that wraps the python-mpv library."""
    def __init__(self, device: Optional[str] = None, initial_volume: float = 1.0) -> None:
        self.player = MPV(
            video=False,
            terminal=False,
            log_handler=self._mpv_log,
            # Set options directly in the constructor
            audio_samplerate=44100,
            audio_channels="stereo",
            keep_open="no",
            network_timeout=7,
            msg_level=os.environ.get("LVA_MPV_MSG_LEVEL", "all=warn"),
        )

        _select_backend(self.player, device)
        self.set_volume(int(initial_volume * 100))

        self.is_playing: bool = False
        self._playlist: List[str] = []
        self._done_callback: Optional[Callable[[], None]] = None
        self._done_callback_lock = Lock()
        self._pre_duck_volume: Optional[int] = None

        self.player.observe_property("idle-active", self._on_idle_active)

    def play(
        self,
        url: Union[str, Sequence[str], bytes],
        done_callback: Optional[Callable[[], None]] = None,
    ) -> None:
        """Plays a URL or a sequence of URLs."""
        self.stop() # Ensure player is in a clean state

        with self._done_callback_lock:
            self._done_callback = done_callback

        playlist: List[str] = []
        if isinstance(url, (list, tuple)):
            playlist = list(url)
        elif isinstance(url, bytes):
            playlist = [url.decode(errors="ignore")]
        elif isinstance(url, str):
            playlist = [url]
        else:
            _LOGGER.error("play() expected str, bytes, or sequence, got %r", type(url))
            self._run_done_callback()
            return
        
        if not playlist:
            self._run_done_callback()
            return
            
        # Load the full playlist into mpv
        self.player.playlist_clear()
        for item in playlist:
            self.player.playlist_append(item)

        self.is_playing = True
        self.player.playlist_pos = 0 # Start playing from the first item
        
    def pause(self) -> None:
        """Pauses playback."""
        self.player.pause = True

    def resume(self) -> None:
        """Resumes playback."""
        self.player.pause = False

    def stop(self) -> None:
        """Stops playback and clears the playlist."""
        if self.is_playing:
            self.is_playing = False
            self.player.playlist_clear()
            self.player.command("stop")
            self._run_done_callback()
            
    def set_volume(self, volume: int) -> None:
        """Sets the player volume from 0 to 100."""
        try:
            self.player.volume = max(0, min(100, volume))
        except Exception:
            _LOGGER.exception("set_volume() failed")

    def duck(self, target_percent: int = 20) -> None:
        """Lowers the volume for an announcement."""
        if self._pre_duck_volume is not None:
            return
        try:
            self._pre_duck_volume = int(self.player.volume)
            self.set_volume(target_percent)
        except Exception:
            _LOGGER.exception("duck() failed")

    def unduck(self) -> None:
        """Restores the volume after an announcement."""
        if self._pre_duck_volume is None:
            return
        try:
            self.set_volume(self._pre_duck_volume)
        finally:
            self._pre_duck_volume = None

    def _on_idle_active(self, _name: str, active: bool) -> None:
        """Callback triggered when mpv enters or leaves the idle state."""
        if active and self.is_playing:
            _LOGGER.debug("mpv became idle; treating as end-of-playback")
            self.is_playing = False
            self._run_done_callback()

    def _run_done_callback(self) -> None:
        """Safely runs the done_callback if it exists."""
        with self._done_callback_lock:
            cb = self._done_callback
            self._done_callback = None
        if cb:
            try:
                cb()
            except Exception:
                _LOGGER.exception("Error in done_callback")

    def _mpv_log(self, level: str, prefix: str, text: str) -> None:
        """Routes mpv's internal logs to our logger."""
        msg = f"mpv[{prefix}]: {text}".rstrip()
        if level == "error":
            _LOGGER.error(msg)
        elif level == "warn":
            _LOGGER.warning(msg)
        elif level == "info":
            _LOGGER.info(msg)
        else:
            _LOGGER.debug(msg)