""" Media player using mpv in a subprocess. 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)