"""SaraTune as a media player the system knows about: the desktop's media controls, media keys and lock screen show what's playing and control it. Remote is the platform-neutral part: it follows whichever mode played last and turns system commands into calls on that mode. The backends only talk to Remote: Mpris.py on Linux (D-Bus), Smtc.py on Windows (System Media Transport Controls). Neither is required: if one can't start, SaraTune just plays without it.""" import asyncio import sys from pathlib import Path from typing import Awaitable, Callable, Optional import flet as ft from Covers import CoverPath from Metadata import Track from Player import Player ArtSize = 512 # pixels; the cover thumbnail handed to the system SeekStep = 10_000 # ms, for Windows' fast-forward / rewind buttons class Remote: def __init__(self, Page: ft.Page, Players: dict[str, Player], Modes: dict, Start: str, SetVolume: Callable[[float], None], Quit: Callable[[], Awaitable[None]]): self.Page = Page self.Players = Players self.Modes = Modes self.ActiveName = Start # the mode the controls show self.SetVolumeEverywhere = SetVolume self._Quit = Quit self.TrackNumber = 0 # changes with every new track; MPRIS track IDs are made from it self.ArtPath: Optional[Path] = None self._Shown: Optional[tuple] = None # (mode, track) the backend last heard about self.Backend = None for Name, ThePlayer in Players.items(): ThePlayer.Watchers.append(lambda Kind, Name=Name: self._PlayerEvent(Name, Kind)) async def Start(self) -> None: try: if sys.platform.startswith("linux"): from Mpris import MprisBackend self.Backend = MprisBackend(self) elif sys.platform == "win32": from Smtc import SmtcBackend self.Backend = SmtcBackend(self) if self.Backend is not None: await self.Backend.Start() except Exception as Error: # no session bus, no WinRT, ...: play without system controls print(f"SaraTune: system media controls unavailable: {Error!r}", file=sys.stderr) self.Backend = None return await self._TrackChanged() async def Stop(self) -> None: if self.Backend is not None: try: await self.Backend.Stop() except Exception: pass # --- What's playing ----------------------------------------------------- @property def Mode(self): return self.Modes[self.ActiveName] @property def ThePlayer(self) -> Player: return self.Players[self.ActiveName] @property def Info(self) -> Optional[Track]: return self.Mode.NowPlaying if self.ThePlayer.Source else None @property def Shuffle(self) -> bool: return bool(getattr(self.Mode, "Queue", None) and self.Mode.Queue.Shuffle) @property def Repeat(self) -> str: """Off, All or One. Audio mode's loop counts as One.""" Queue = getattr(self.Mode, "Queue", None) if Queue is not None: return Queue.Repeat return "One" if self.ThePlayer.Loop else "Off" def _PlayerEvent(self, Name: str, Kind: str) -> None: if self.Backend is None: return if Kind == "Started" and Name != self.ActiveName: self.ActiveName = Name self.Page.run_task(self._TrackChanged) return if Name != self.ActiveName: return if Kind == "Changed": self.Backend.StateChanged() elif Kind == "Seeked": self.Backend.Seeked() elif Kind == "Volume": self.Backend.VolumeChanged() def MediaChanged(self) -> None: """A mode's NowPlaying changed (Main hooks this to App.MediaChanged).""" if self.Backend is not None: self.Page.run_task(self._TrackChanged) def ModesChanged(self) -> None: """Shuffle, repeat or loop changed in the app.""" if self.Backend is not None: self.Backend.StateChanged() async def _TrackChanged(self) -> None: Info = self.Info Key = (self.ActiveName, Info.Path if Info else None, id(Info)) if Key != self._Shown: self._Shown = Key self.TrackNumber += 1 self.ArtPath = await asyncio.to_thread(CoverPath, Info.Path, ArtSize) if Info else None if self._Shown != Key: return # changed again while the cover was being made if self.Backend is not None: self.Backend.TrackChanged() # --- Commands from the system ------------------------------------------- def Run(self, Coroutine, *Arguments) -> None: """Backends call this from their callbacks; the work runs on Flet's loop.""" self.Page.run_task(Coroutine, *Arguments) async def Play(self) -> None: if not self.ThePlayer.Playing: await self.ThePlayer.Toggle() async def Pause(self) -> None: await self.ThePlayer.Pause() async def Toggle(self) -> None: await self.ThePlayer.Toggle() async def StopPlaying(self) -> None: """MPRIS Stop: pause and go back to the start, keeping the track.""" await self.ThePlayer.Pause() await self.ThePlayer.Seek(0) async def Next(self) -> None: await self.Mode.Next() async def Previous(self) -> None: await self.Mode.Previous() async def SeekBy(self, Milliseconds: int) -> None: Target = self.ThePlayer.Position + Milliseconds if self.ThePlayer.Duration and Target >= self.ThePlayer.Duration: await self.Mode.Next() # MPRIS: seeking past the end acts like Next else: await self.ThePlayer.Seek(max(0, Target)) async def SetPosition(self, Milliseconds: int) -> None: await self.ThePlayer.Seek(Milliseconds) def SetVolume(self, Volume: float) -> None: self.SetVolumeEverywhere(max(0.0, min(1.0, Volume))) for Mode in self.Modes.values(): Mode.Bar.SetVolume(self.ThePlayer.Volume) Mode._Show(Mode.Bar) def SetShuffle(self, Enabled: bool) -> None: """The modes tell the backend back through ModesChanged.""" if hasattr(self.Mode, "SetShuffle"): self.Mode.SetShuffle(Enabled) else: self.ModesChanged() # Audio mode has no shuffle: put the system's switch back def SetRepeat(self, Mode: str) -> None: if hasattr(self.Mode, "SetRepeat"): self.Mode.SetRepeat(Mode) else: self.Mode.SetLoop(Mode == "One") async def Raise(self) -> None: await self.Page.window.to_front() async def Quit(self) -> None: await self._Quit()