"""Linux: SaraTune on the session bus as an MPRIS media player (org.mpris.MediaPlayer2.SaraTune), which is what the desktop's media controls, media keys, lock screen and playerctl use. Spec: https://specifications.freedesktop.org/mpris-spec/latest/ dbus-fast runs on the same asyncio loop as Flet, so its callbacks can hand work straight to Remote.""" import os from typing import Annotated from dbus_fast import BusType, NameFlag, PropertyAccess, RequestNameReply, Variant from dbus_fast.aio import MessageBus from dbus_fast.annotations import ( DBusBool, DBusDict, DBusDouble, DBusInt64, DBusObjectPath, DBusSignature, DBusStr, ) from dbus_fast.service import ServiceInterface, dbus_method, dbus_property, dbus_signal from Core import AppName, AudioExtensions StrList = Annotated[list[str], DBusSignature("as")] ObjectPath = "/org/mpris/MediaPlayer2" BusName = f"org.mpris.MediaPlayer2.{AppName}" TrackPathBase = "/garden/srhtnin/SaraTune/Track/" NoTrack = "/org/mpris/MediaPlayer2/TrackList/NoTrack" LoopNames = {"Off": "None", "All": "Playlist", "One": "Track"} # SaraTune -> MPRIS MimeTypes = [ "audio/mpeg", "audio/flac", "audio/x-flac", "audio/ogg", "audio/x-vorbis+ogg", "audio/opus", "audio/wav", "audio/x-wav", "audio/mp4", "audio/x-m4a", "audio/aac", "audio/x-aiff", "audio/x-matroska", "audio/webm", ] class RootInterface(ServiceInterface): def __init__(self, Remote): super().__init__("org.mpris.MediaPlayer2") self.Remote = Remote @dbus_method() def Raise(self) -> None: self.Remote.Run(self.Remote.Raise) @dbus_method() def Quit(self) -> None: self.Remote.Run(self.Remote.Quit) @dbus_property(access=PropertyAccess.READ) def CanQuit(self) -> DBusBool: return True @dbus_property(access=PropertyAccess.READ) def CanRaise(self) -> DBusBool: return True @dbus_property(access=PropertyAccess.READ) def HasTrackList(self) -> DBusBool: return False @dbus_property(access=PropertyAccess.READ) def Identity(self) -> DBusStr: return AppName @dbus_property(access=PropertyAccess.READ) def DesktopEntry(self) -> DBusStr: return AppName # SaraTune.desktop: the icon and name the desktop shows @dbus_property(access=PropertyAccess.READ) def SupportedUriSchemes(self) -> StrList: return ["file"] @dbus_property(access=PropertyAccess.READ) def SupportedMimeTypes(self) -> StrList: return MimeTypes class PlayerInterface(ServiceInterface): def __init__(self, Remote): super().__init__("org.mpris.MediaPlayer2.Player") self.Remote = Remote # --- Methods ------------------------------------------------------------ @dbus_method() def Next(self) -> None: self.Remote.Run(self.Remote.Next) @dbus_method() def Previous(self) -> None: self.Remote.Run(self.Remote.Previous) @dbus_method() def Pause(self) -> None: self.Remote.Run(self.Remote.Pause) @dbus_method() def PlayPause(self) -> None: self.Remote.Run(self.Remote.Toggle) @dbus_method() def Stop(self) -> None: self.Remote.Run(self.Remote.StopPlaying) @dbus_method() def Play(self) -> None: self.Remote.Run(self.Remote.Play) @dbus_method() def Seek(self, Offset: DBusInt64) -> None: self.Remote.Run(self.Remote.SeekBy, Offset // 1000) @dbus_method() def SetPosition(self, TrackId: DBusObjectPath, Position: DBusInt64) -> None: # Ignored unless it's for the current track (the spec's rule against stale requests). Duration = self.Remote.ThePlayer.Duration * 1000 if TrackId == self._TrackId() and 0 <= Position <= (Duration or Position): self.Remote.Run(self.Remote.SetPosition, Position // 1000) @dbus_method() def OpenUri(self, Uri: DBusStr) -> None: from urllib.parse import unquote, urlparse Parsed = urlparse(Uri) if Parsed.scheme == "file" and os.path.splitext(Parsed.path)[1].lower() in AudioExtensions: self.Remote.Page.run_task(self.Remote.Raise) self.Remote.Modes["Audio"].TheApp.OpenInAudio([unquote(Parsed.path)]) @dbus_signal() def Seeked(self) -> DBusInt64: return self.Remote.ThePlayer.Position * 1000 # --- Properties --------------------------------------------------------- def _TrackId(self) -> str: return f"{TrackPathBase}{self.Remote.TrackNumber}" if self.Remote.Info else NoTrack @dbus_property(access=PropertyAccess.READ) def PlaybackStatus(self) -> DBusStr: if not self.Remote.Info: return "Stopped" return "Playing" if self.Remote.ThePlayer.Playing else "Paused" @dbus_property() def LoopStatus(self) -> DBusStr: return LoopNames[self.Remote.Repeat] @LoopStatus.setter def LoopStatus(self, Value: DBusStr) -> None: Mode = {Mpris: Ours for Ours, Mpris in LoopNames.items()}.get(Value) if Mode: self.Remote.SetRepeat(Mode) @dbus_property() def Rate(self) -> DBusDouble: return 1.0 @Rate.setter def Rate(self, Value: DBusDouble) -> None: pass # only 1.0 (MinimumRate = MaximumRate) @dbus_property() def Shuffle(self) -> DBusBool: return self.Remote.Shuffle @Shuffle.setter def Shuffle(self, Value: DBusBool) -> None: self.Remote.SetShuffle(Value) @dbus_property(access=PropertyAccess.READ) def Metadata(self) -> DBusDict: return self.MetadataValue() @dbus_property() def Volume(self) -> DBusDouble: return float(self.Remote.ThePlayer.Volume) @Volume.setter def Volume(self, Value: DBusDouble) -> None: self.Remote.SetVolume(Value) @dbus_property(access=PropertyAccess.READ) def Position(self) -> DBusInt64: return self.Remote.ThePlayer.Position * 1000 @dbus_property(access=PropertyAccess.READ) def MinimumRate(self) -> DBusDouble: return 1.0 @dbus_property(access=PropertyAccess.READ) def MaximumRate(self) -> DBusDouble: return 1.0 @dbus_property(access=PropertyAccess.READ) def CanGoNext(self) -> DBusBool: return self.Remote.Info is not None @dbus_property(access=PropertyAccess.READ) def CanGoPrevious(self) -> DBusBool: return self.Remote.Info is not None @dbus_property(access=PropertyAccess.READ) def CanPlay(self) -> DBusBool: return self.Remote.Info is not None @dbus_property(access=PropertyAccess.READ) def CanPause(self) -> DBusBool: return self.Remote.Info is not None @dbus_property(access=PropertyAccess.READ) def CanSeek(self) -> DBusBool: return self.Remote.Info is not None @dbus_property(access=PropertyAccess.READ) def CanControl(self) -> DBusBool: return True def MetadataValue(self) -> dict: Info = self.Remote.Info if Info is None: return {"mpris:trackid": Variant("o", NoTrack)} Data = { "mpris:trackid": Variant("o", self._TrackId()), "xesam:title": Variant("s", Info.DisplayTitle), "xesam:url": Variant("s", f"file://{_Quote(Info.Path)}"), } Length = self.Remote.ThePlayer.Duration or Info.Duration if Length: Data["mpris:length"] = Variant("x", int(Length) * 1000) if Info.Artist or Info.AlbumArtist: Data["xesam:artist"] = Variant("as", [Info.Artist or Info.AlbumArtist]) if Info.Album: Data["xesam:album"] = Variant("s", Info.Album) if Info.AlbumArtist: Data["xesam:albumArtist"] = Variant("as", [Info.AlbumArtist]) if Info.Genre: Data["xesam:genre"] = Variant("as", [Info.Genre]) if Info.TrackNumber: Data["xesam:trackNumber"] = Variant("i", Info.TrackNumber) if Info.DiscNumber: Data["xesam:discNumber"] = Variant("i", Info.DiscNumber) if self.Remote.ArtPath: Data["mpris:artUrl"] = Variant("s", f"file://{_Quote(str(self.Remote.ArtPath))}") return Data def _Quote(PathText: str) -> str: from urllib.parse import quote return quote(PathText) class MprisBackend: def __init__(self, Remote): self.Remote = Remote self.Bus: MessageBus | None = None self.Root = RootInterface(Remote) self.Player = PlayerInterface(Remote) self._LastState: dict = {} async def Start(self) -> None: self.Bus = await MessageBus(bus_type=BusType.SESSION).connect() self.Bus.export(ObjectPath, self.Root) self.Bus.export(ObjectPath, self.Player) # A second SaraTune takes an instance name, as the spec suggests. Reply = await self.Bus.request_name(BusName, NameFlag.DO_NOT_QUEUE) if Reply not in (RequestNameReply.PRIMARY_OWNER, RequestNameReply.ALREADY_OWNER): await self.Bus.request_name(f"{BusName}.instance{os.getpid()}", NameFlag.DO_NOT_QUEUE) async def Stop(self) -> None: if self.Bus is not None: self.Bus.disconnect() self.Bus = None def _State(self) -> dict: P = self.Player return { "PlaybackStatus": P.PlaybackStatus, "LoopStatus": P.LoopStatus, "Shuffle": P.Shuffle, "CanGoNext": P.CanGoNext, "CanGoPrevious": P.CanGoPrevious, "CanPlay": P.CanPlay, "CanPause": P.CanPause, "CanSeek": P.CanSeek, } def StateChanged(self) -> None: """Only what actually changed goes out (Changed fires often).""" if self.Bus is None: return State = self._State() Changed = {Key: Value for Key, Value in State.items() if self._LastState.get(Key) != Value} self._LastState = State if Changed: self.Player.emit_properties_changed(Changed) def TrackChanged(self) -> None: if self.Bus is None: return self._LastState = self._State() self.Player.emit_properties_changed({"Metadata": self.Player.MetadataValue(), **self._LastState}) def Seeked(self) -> None: if self.Bus is not None: self.Player.Seeked() def VolumeChanged(self) -> None: if self.Bus is not None: self.Player.emit_properties_changed({"Volume": float(self.Remote.ThePlayer.Volume)})