from dbus_next.service import ServiceInterface, dbus_property, method, signal, Variant from dbus_next import PropertyAccess, RequestNameReply, BusType from dbus_next.aio import MessageBus import asyncio async def setup_mpris(*names, bus_address=None, system=False): # TODO maybe they should all share a bus for speed async def setup(name): if system: bus_type = BusType.SYSTEM else: bus_type = BusType.SESSION bus = await MessageBus(bus_type=bus_type, bus_address=bus_address).connect() player = MprisPlayer(bus) bus.export('/org/mpris/MediaPlayer2', player) bus.export('/org/mpris/MediaPlayer2', MprisRoot()) reply = await bus.request_name(f'org.mpris.MediaPlayer2.{name}') assert reply == RequestNameReply.PRIMARY_OWNER return player players = await asyncio.gather(*(setup(name) for name in names)) await asyncio.gather(*(p.ping() for p in players)) return players async def setup_playerctld(bus_address=None): bus = await MessageBus(bus_address=bus_address).connect() playerctld = PlayerctldInterface(bus) bus.export('/org/mpris/MediaPlayer2', playerctld) reply = await bus.request_name('org.mpris.MediaPlayer2.playerctld') assert reply == RequestNameReply.PRIMARY_OWNER return playerctld class MprisRoot(ServiceInterface): def __init__(self): super().__init__('org.mpris.MediaPlayer2') @method() def Raise(self): return @method() def Quit(self): return @dbus_property(access=PropertyAccess.READ) def CanRaise(self) -> 'b': return False @dbus_property(access=PropertyAccess.READ) def HasTrackList(self) -> 'b': return False @dbus_property(access=PropertyAccess.READ) def Identity(self) -> 's': return 'playerctl test client' @dbus_property(access=PropertyAccess.READ) def SupportedUriSchemes(self) -> 'as': return ['file'] @dbus_property(access=PropertyAccess.READ) def SupportedMimeTypes(self) -> 'as': return ['audio/mp3'] class MprisPlayer(ServiceInterface): def __init__(self, bus): super().__init__('org.mpris.MediaPlayer2.Player') self.counter = 0 self.reset() self.bus = bus def reset(self): # method calls self.next_called = False self.previous_called = False self.pause_called = False self.play_pause_called = False self.stop_called = False self.play_called = False self.seek_called_with = None self.set_position_called_with = None self.open_uri_called_with = None # properties self.playback_status = 'Playing' self.loop_status = 'None' self.rate = 1.0 self.shuffle = False self.metadata = {} self.volume = 1.0 self.position = 0 self.minimum_rate = 1.0 self.maximum_rate = 1.0 self.can_go_next = True self.can_go_previous = True self.can_play = True self.can_pause = True self.can_seek = True self.can_control = True # signals self.seeked_value = 0 async def ping(self): await self.bus.introspect('org.freedesktop.DBus', '/org/freedesktop/DBus') async def set_artist_title(self, artist, title, track_id=None): if track_id is None: self.counter += 1 track_id = '/' + str(self.counter) self.metadata = { 'xesam:title': Variant('s', title), 'xesam:artist': Variant('as', [artist]), 'mpris:trackid': Variant('o', track_id), } self.emit_properties_changed({ 'Metadata': self.metadata, }) await self.ping() async def clear_metadata(self): self.counter += 1 self.metadata = { 'mpris:trackid': Variant('o', '/' + str(self.counter)), } self.emit_properties_changed({ 'Metadata': self.metadata, }) await self.ping() async def disconnect(self): self.bus.disconnect() await self.bus.wait_for_disconnect() @method() def Next(self): self.next_called = True @method() def Previous(self): self.previous_called = True @method() def Pause(self): self.pause_called = True @method() def PlayPause(self): self.play_pause_called = True @method() def Stop(self): self.stop_called = True @method() def Play(self): self.play_called = True @method() def Seek(self, offset: 'x'): self.seek_called_with = offset @method() def SetPosition(self, track_id: 'o', position: 'x'): self.set_position_called_with = (track_id, position) @method() def OpenUri(self, uri: 's'): self.open_uri_called_with = uri @signal() def Seeked(self) -> 'x': return self.seeked_value @dbus_property(access=PropertyAccess.READ) def PlaybackStatus(self) -> 's': return self.playback_status @dbus_property() def LoopStatus(self) -> 's': return self.loop_status @LoopStatus.setter def LoopStatus(self, status: 's'): self.loop_status = status @dbus_property() def Rate(self) -> 'd': return self.rate @Rate.setter def Rate(self, rate: 'd'): self.rate = rate @dbus_property() def Shuffle(self) -> 'b': return self.shuffle @Shuffle.setter def Shuffle(self, shuffle: 'b'): self.shuffle = shuffle @dbus_property(access=PropertyAccess.READ) def Metadata(self) -> 'a{sv}': return self.metadata @dbus_property() def Volume(self) -> 'd': return self.volume @Volume.setter def Volume(self, volume: 'd'): self.volume = volume @dbus_property(access=PropertyAccess.READ) def Position(self) -> 'x': return self.position @dbus_property(access=PropertyAccess.READ) def MinimumRate(self) -> 'd': return self.minimum_rate @dbus_property(access=PropertyAccess.READ) def MaximumRate(self) -> 'd': return self.maximum_rate @dbus_property(access=PropertyAccess.READ) def CanGoNext(self) -> 'b': return self.can_go_next @dbus_property(access=PropertyAccess.READ) def CanGoPrevious(self) -> 'b': return self.can_go_previous @dbus_property(access=PropertyAccess.READ) def CanPlay(self) -> 'b': return self.can_play @dbus_property(access=PropertyAccess.READ) def CanPause(self) -> 'b': return self.can_pause @dbus_property(access=PropertyAccess.READ) def CanSeek(self) -> 'b': return self.can_seek @dbus_property(access=PropertyAccess.READ) def CanControl(self) -> 'b': return self.can_control class PlayerctldInterface(ServiceInterface): '''just enough of playerctld for testing''' def __init__(self, bus): super().__init__('com.github.altdesktop.playerctld') self.bus = bus self.player_names = [] @dbus_property(access=PropertyAccess.READ) def PlayerNames(self) -> 'as': return self.player_names async def disconnect(self): self.bus.disconnect() await self.bus.wait_for_disconnect()