player.py
1 from __future__ import annotations 2 3 import logging 4 import shutil 5 import subprocess 6 7 import mpv 8 9 from linamp.stations import Station 10 11 log = logging.getLogger(__name__) 12 13 14 class AudioPlayer: 15 """Wraps mpv into a simple interface for the TUI layer.""" 16 17 def __init__(self) -> None: 18 self._mpv = mpv.MPV( 19 video=False, 20 terminal=False, 21 input_terminal=False, 22 ytdl_format="bestaudio/best", 23 ) 24 self._current_station: Station | None = None 25 self._stopped = True 26 27 def play(self, station: Station) -> None: 28 self._current_station = station 29 self._stopped = False 30 url = self._resolve_url(station.url) 31 self._mpv.play(url) 32 33 # Domains where mpv's built-in ytdl hook handles playback natively. 34 # We pass these URLs straight through — pre-resolving would produce 35 # temporary signed URLs that expire before mpv can use them. 36 YTDL_DOMAINS = ( 37 "youtube.com", "youtu.be", "youtube-nocookie.com", 38 "music.youtube.com", 39 ) 40 41 @classmethod 42 def _is_ytdl_url(cls, url: str) -> bool: 43 """Check if a URL should be handled by mpv's ytdl hook.""" 44 from urllib.parse import urlparse 45 try: 46 host = urlparse(url).hostname or "" 47 return any(host == d or host.endswith("." + d) for d in cls.YTDL_DOMAINS) 48 except Exception: 49 return False 50 51 @classmethod 52 def _resolve_url(cls, url: str) -> str: 53 """Try yt-dlp to extract a direct stream URL, fall back to raw URL. 54 55 YouTube URLs are passed through directly — mpv's ytdl hook handles 56 them better than pre-resolved temporary URLs that expire quickly. 57 """ 58 # Local file paths — mpv handles these natively 59 if url.startswith("/"): 60 return url 61 # Let mpv's ytdl hook handle known video platforms directly 62 if cls._is_ytdl_url(url): 63 return url 64 # Skip resolution for direct stream URLs (icecast/shoutcast/raw audio) 65 if any(h in url for h in ("ice", "stream", ".mp3", ".aac", ".ogg", ".m3u", ".pls")): 66 return url 67 ytdlp = shutil.which("yt-dlp") 68 if not ytdlp: 69 return url 70 try: 71 result = subprocess.run( 72 [ytdlp, "--no-download", "--print", "urls", "-f", "bestaudio/best", url], 73 capture_output=True, text=True, timeout=15, 74 ) 75 resolved = result.stdout.strip().splitlines() 76 if result.returncode == 0 and resolved and resolved[0]: 77 log.info("yt-dlp resolved %s → %s", url, resolved[0]) 78 return resolved[0] 79 except (subprocess.TimeoutExpired, OSError) as exc: 80 log.debug("yt-dlp failed for %s: %s", url, exc) 81 return url 82 83 def toggle_pause(self) -> None: 84 if self._stopped: 85 return 86 self._mpv.pause = not self._mpv.pause 87 88 def stop(self) -> None: 89 self._mpv.stop() 90 self._stopped = True 91 92 @property 93 def volume(self) -> float: 94 try: 95 return self._mpv.volume or 100.0 96 except Exception: 97 return 100.0 98 99 @volume.setter 100 def volume(self, value: float) -> None: 101 self._mpv.volume = max(0.0, min(150.0, value)) 102 103 def volume_up(self, step: float = 5.0) -> None: 104 self.volume = self.volume + step 105 106 def volume_down(self, step: float = 5.0) -> None: 107 self.volume = self.volume - step 108 109 @property 110 def is_playing(self) -> bool: 111 if self._stopped: 112 return False 113 try: 114 return not self._mpv.pause 115 except Exception: 116 return False 117 118 @property 119 def is_paused(self) -> bool: 120 if self._stopped: 121 return False 122 try: 123 return bool(self._mpv.pause) 124 except Exception: 125 return False 126 127 @property 128 def is_stopped(self) -> bool: 129 return self._stopped 130 131 @property 132 def current_station(self) -> Station | None: 133 return self._current_station 134 135 @property 136 def metadata(self) -> dict: 137 try: 138 return dict(self._mpv.metadata or {}) 139 except Exception: 140 return {} 141 142 @property 143 def icy_title(self) -> str: 144 """Get the ICY stream title (current song on radio).""" 145 meta = self.metadata 146 return meta.get("icy-title", "") 147 148 @property 149 def media_title(self) -> str: 150 """Get mpv's synthesized media title (incorporates stream metadata).""" 151 try: 152 return self._mpv.media_title or "" 153 except Exception: 154 return "" 155 156 @property 157 def idle_active(self) -> bool: 158 """True when mpv has no file loaded (track ended or never started).""" 159 try: 160 return bool(self._mpv.idle_active) 161 except Exception: 162 return True 163 164 @property 165 def time_pos(self) -> float | None: 166 if self._stopped: 167 return None 168 try: 169 return self._mpv.time_pos 170 except Exception: 171 return None 172 173 @property 174 def duration(self) -> float | None: 175 if self._stopped: 176 return None 177 try: 178 return self._mpv.duration 179 except Exception: 180 return None 181 182 def shutdown(self) -> None: 183 self._mpv.terminate()