sarah/SaraTune
1
0 Fork 0
mirror of https://git.srhtnin.garden/sarah/SaraTune.git synced 2026-10-14 02:18:11 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
d7f5eba1c23873360f39c4f3de6b1fd914b458ce
SaraTune/Mpris.py
T

302 lines
10 KiB
Python
Raw Blame History

"""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)})
Reference in new issue View Git Blame Copy Permalink