File indexing completed on 2024-06-23 05:30:42
0001 #!/usr/bin/env python3 0002 0003 # SPDX-FileCopyrightText: 2019 The GNOME Music developers 0004 # SPDX-FileCopyrightText: 2023 Fushan Wen <qydwhotmail@gmail.com> 0005 # SPDX-License-Identifier: GPL-2.0-or-later 0006 0007 # pylint: disable=too-many-arguments 0008 0009 # For FreeBSD CI which only has python 3.9 0010 from __future__ import annotations 0011 0012 import logging 0013 import json 0014 import signal 0015 import sys 0016 import threading 0017 from os import getcwd, getpid, path 0018 from typing import Any, Final 0019 0020 sys.path.append(path.dirname(path.abspath(__file__))) 0021 0022 from gi.repository import Gio, GLib 0023 from GLibMainLoopThread import GLibMainLoopThread 0024 0025 0026 def read_player_metadata(json_dict: dict[str, Any]) -> list[dict[str, GLib.Variant]]: 0027 song_list: list[dict[str, int | str | list[str]]] = json_dict["metadata"] 0028 assert len(song_list) > 0 0029 _metadata: list[dict[str, GLib.Variant]] = [] 0030 for song in song_list: 0031 song_dict: dict[str, GLib.Variant] = { 0032 "mpris:trackid": GLib.Variant('o', song["mpris:trackid"]), 0033 "xesam:url": GLib.Variant('s', "file://" + path.join(getcwd(), song["xesam:url"])), 0034 "mpris:length": GLib.Variant('x', int(song["mpris:length"])), # ms 0035 "xesam:title": GLib.Variant('s', song["xesam:title"]), 0036 "xesam:artist": GLib.Variant('as', song["xesam:artist"]), 0037 "mpris:artUrl": GLib.Variant('s', "") 0038 } 0039 if "xesam:album" in song.keys(): 0040 song_dict["xesam:album"] = GLib.Variant('s', song["xesam:album"]) 0041 if "mpris:artUrl" in song.keys(): 0042 song_dict["mpris:artUrl"] = GLib.Variant('s', "file://" + path.join(getcwd(), song["mpris:artUrl"])) 0043 if "kde:pid" in song.keys(): 0044 song_dict["kde:pid"] = GLib.Variant('u', song["kde:pid"]) 0045 _metadata.append(song_dict) 0046 0047 return _metadata 0048 0049 0050 def read_base_properties(json_dict: dict[str, Any]) -> dict[str, GLib.Variant]: 0051 prop_dict: dict[str, bool | str | list[str]] = json_dict["base_properties"] 0052 assert len(prop_dict) > 0 0053 _base_properties: dict[str, GLib.Variant] = { 0054 "CanQuit": GLib.Variant('b', prop_dict["CanQuit"]), 0055 "Fullscreen": GLib.Variant('b', prop_dict["Fullscreen"]), 0056 "CanSetFullscreen": GLib.Variant('b', prop_dict["CanSetFullscreen"]), 0057 "CanRaise": GLib.Variant('b', prop_dict["CanRaise"]), 0058 "HasTrackList": GLib.Variant('b', prop_dict["HasTrackList"]), 0059 "Identity": GLib.Variant('s', prop_dict["Identity"]), 0060 "DesktopEntry": GLib.Variant('s', prop_dict["DesktopEntry"]), 0061 "SupportedUriSchemes": GLib.Variant('as', prop_dict["SupportedUriSchemes"]), 0062 "SupportedMimeTypes": GLib.Variant('as', prop_dict["SupportedMimeTypes"]), 0063 } 0064 return _base_properties 0065 0066 0067 def read_player_properties(json_dict: dict[str, Any], _metadata: dict[str, GLib.Variant]) -> dict[str, GLib.Variant]: 0068 prop_dict: dict[str, int | float | bool | str | dict] = json_dict["player_properties"] 0069 assert len(prop_dict) > 0 0070 _player_properties: dict[str, GLib.Variant] = { 0071 "PlaybackStatus": GLib.Variant("s", prop_dict["PlaybackStatus"]), 0072 "LoopStatus": GLib.Variant("s", prop_dict["LoopStatus"]), 0073 "Rate": GLib.Variant("d", prop_dict["Rate"]), 0074 "Shuffle": GLib.Variant("b", prop_dict["Shuffle"]), 0075 "Metadata": GLib.Variant("a{sv}", _metadata), 0076 "Position": GLib.Variant("x", prop_dict["Position"]), 0077 "MinimumRate": GLib.Variant("d", prop_dict["MinimumRate"]), 0078 "MaximumRate": GLib.Variant("d", prop_dict["MaximumRate"]), 0079 "Volume": GLib.Variant("d", prop_dict["Volume"]), 0080 "CanGoNext": GLib.Variant("b", prop_dict["CanGoNext"]), 0081 "CanGoPrevious": GLib.Variant("b", prop_dict["CanGoPrevious"]), 0082 "CanPlay": GLib.Variant("b", prop_dict["CanPlay"]), 0083 "CanPause": GLib.Variant("b", prop_dict["CanPause"]), 0084 "CanSeek": GLib.Variant("b", prop_dict["CanSeek"]), 0085 "CanControl": GLib.Variant("b", prop_dict["CanControl"]), 0086 } 0087 return _player_properties 0088 0089 0090 class Mpris2: 0091 """ 0092 MPRIS2 interface implemented in GDBus, since dbus-python does not support property 0093 """ 0094 0095 BASE_IFACE: Final = "org.mpris.MediaPlayer2" 0096 OBJECT_PATH: Final = "/org/mpris/MediaPlayer2" 0097 PLAYER_IFACE: Final = GLib.Variant('s', "org.mpris.MediaPlayer2.Player") 0098 APP_INTERFACE: Final = f"org.mpris.MediaPlayer2.appiumtest.instance{str(getpid())}" 0099 0100 connection: Gio.DBusConnection | None = None 0101 0102 def __init__(self, metadata: list[dict[str, GLib.Variant]], base_properties: dict[str, GLib.Variant], player_properties: dict[str, GLib.Variant], current_index: int) -> None: 0103 self.metadata: list[dict[str, GLib.Variant]] = metadata 0104 self.base_properties: dict[str, GLib.Variant] = base_properties 0105 self.player_properties: dict[str, GLib.Variant] = player_properties 0106 self.current_index: int = current_index 0107 0108 self.registered_event = threading.Event() 0109 self.playback_status_set_event = threading.Event() 0110 self.metadata_updated_event = threading.Event() 0111 0112 self.__owner_id: int = Gio.bus_own_name(Gio.BusType.SESSION, self.APP_INTERFACE, Gio.BusNameOwnerFlags.NONE, self.on_bus_acquired, None, None) 0113 assert self.__owner_id > 0 0114 0115 self.__prop_reg_id: int = 0 0116 self.__player_reg_id: int = 0 0117 self.__base_reg_id: int = 0 0118 0119 def quit(self) -> None: 0120 if self.connection is None: 0121 return 0122 self.connection.unregister_object(self.__prop_reg_id) 0123 self.__prop_reg_id = 0 0124 self.connection.unregister_object(self.__player_reg_id) 0125 self.__player_reg_id = 0 0126 self.connection.unregister_object(self.__base_reg_id) 0127 self.__base_reg_id = 0 0128 Gio.bus_unown_name(self.__owner_id) 0129 self.connection.flush_sync(None) # Otherwise flaky 0130 logging.info("Player exit") 0131 0132 def on_bus_acquired(self, connection: Gio.DBusConnection, name: str, *args) -> None: 0133 """ 0134 Interface is ready, now register objects. 0135 """ 0136 self.connection = connection 0137 0138 with open("../libkmpris/dbus/org.freedesktop.DBus.Properties.xml", encoding="utf-8") as handler: 0139 properties_introspection_xml: str = '\n'.join(handler.readlines()) 0140 introspection_data = Gio.DBusNodeInfo.new_for_xml(properties_introspection_xml) 0141 self.__prop_reg_id = connection.register_object(self.OBJECT_PATH, introspection_data.interfaces[0], self.properties_handle_method_call, None, None) 0142 assert self.__prop_reg_id > 0 0143 0144 with open("../libkmpris/dbus/org.mpris.MediaPlayer2.Player.xml", encoding="utf-8") as handler: 0145 player_introspection_xml: str = '\n'.join(handler.readlines()) 0146 introspection_data = Gio.DBusNodeInfo.new_for_xml(player_introspection_xml) 0147 self.__player_reg_id = connection.register_object(self.OBJECT_PATH, introspection_data.interfaces[0], self.player_handle_method_call, self.player_handle_get_property, self.player_handle_set_property) 0148 assert self.__player_reg_id != 0 0149 0150 with open("../libkmpris/dbus/org.mpris.MediaPlayer2.xml", encoding="utf-8") as handler: 0151 interface_introspection_xml: str = '\n'.join(handler.readlines()) 0152 introspection_data = Gio.DBusNodeInfo.new_for_xml(interface_introspection_xml) 0153 self.__base_reg_id = connection.register_object(self.OBJECT_PATH, introspection_data.interfaces[0], self.interface_handle_method_call, self.interface_handle_get_property, self.interface_handle_set_property) 0154 assert self.__base_reg_id != 0 0155 0156 print("MPRIS registered", file=sys.stderr, flush=True) 0157 self.registered_event.set() 0158 0159 def properties_handle_method_call(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, method_name: str, parameters: GLib.Variant, invocation: Gio.DBusMethodInvocation) -> None: 0160 """ 0161 Handles method calls for org.freedesktop.DBus.Properties 0162 """ 0163 assert interface_name == "org.freedesktop.DBus.Properties", f"Wrong interface name {interface_name} from {sender}" 0164 0165 if method_name == "Get": 0166 ret: Any = None 0167 interface: str = parameters[0] 0168 property_name: str = parameters[1] 0169 if interface == self.BASE_IFACE: 0170 ret = self.base_properties[property_name] 0171 elif interface == self.PLAYER_IFACE.get_string(): 0172 ret = self.player_properties[property_name] 0173 else: 0174 assert False, f"Unknown interface {interface}" 0175 0176 print(f"Get: {interface} {property_name} {ret}", file=sys.stderr, flush=True) 0177 # https://bugzilla.gnome.org/show_bug.cgi?id=765603 0178 invocation.return_value(GLib.Variant.new_tuple(ret)) 0179 0180 elif method_name == "GetAll": 0181 interface = parameters[0] 0182 if interface == self.BASE_IFACE: 0183 ret = GLib.Variant('a{sv}', self.base_properties) 0184 elif interface == self.PLAYER_IFACE.get_string(): 0185 ret = GLib.Variant('a{sv}', self.player_properties) 0186 else: 0187 assert False, f"Unknown interface {interface}" 0188 0189 print(f"GetAll: {interface} {ret}", file=sys.stderr, flush=True) 0190 invocation.return_value(GLib.Variant.new_tuple(ret)) 0191 0192 elif method_name == "Set": 0193 interface = parameters[0] 0194 property_name = parameters[1] 0195 value: Any = parameters[2] 0196 0197 if interface != self.PLAYER_IFACE.get_string(): 0198 assert False, f"Wrong interface {interface}" 0199 0200 if property_name == "Rate": 0201 self.set_rate(value, connection, object_path) 0202 elif property_name == "LoopStatus": 0203 self.set_repeat(value, connection, object_path) 0204 elif property_name == "Shuffle": 0205 self.set_shuffle(value, connection, object_path) 0206 elif property_name == "Volume": 0207 self.set_volume(value, connection, object_path) 0208 else: 0209 assert False, f"Unknown property {property_name}" 0210 0211 print(f"Set: {interface} {property_name} {value}", file=sys.stderr, flush=True) 0212 0213 else: 0214 logging.error("Unhandled method: %s", method_name) 0215 invocation.return_error_literal(Gio.dbus_error_quark(), Gio.DBusError.UNKNOWN_METHOD, f"Unknown method {method_name}") 0216 0217 def set_playing(self, playing: bool, connection: Gio.DBusConnection, object_path: str) -> None: 0218 """ 0219 Changes the playing state 0220 """ 0221 self.player_properties["PlaybackStatus"] = GLib.Variant('s', "Playing" if playing else "Paused") 0222 changed_properties = GLib.Variant('a{sv}', { 0223 "PlaybackStatus": self.player_properties["PlaybackStatus"], 0224 }) 0225 Gio.DBusConnection.emit_signal(connection, None, object_path, "org.freedesktop.DBus.Properties", "PropertiesChanged", GLib.Variant.new_tuple(self.PLAYER_IFACE, changed_properties, GLib.Variant('as', ()))) 0226 self.playback_status_set_event.set() 0227 0228 def stop(self, connection: Gio.DBusConnection, object_path: str) -> None: 0229 """ 0230 Changes the playing state 0231 """ 0232 self.player_properties["PlaybackStatus"] = GLib.Variant('s', "Stopped") 0233 self.player_properties["Metadata"] = GLib.Variant('a{sv}', {}) 0234 changed_properties = GLib.Variant('a{sv}', { 0235 "PlaybackStatus": self.player_properties["PlaybackStatus"], 0236 "Metadata": self.player_properties["Metadata"], 0237 }) 0238 Gio.DBusConnection.emit_signal(connection, None, object_path, "org.freedesktop.DBus.Properties", "PropertiesChanged", GLib.Variant.new_tuple(self.PLAYER_IFACE, changed_properties, GLib.Variant('as', ()))) 0239 0240 def set_rate(self, rate: float, connection: Gio.DBusConnection, object_path: str) -> None: 0241 """ 0242 Changes the shuffle state 0243 """ 0244 assert isinstance(rate, float), f"argument is not a float but {type(rate)}" 0245 assert self.player_properties["MinimumRate"].get_double() <= rate <= self.player_properties["MaximumRate"].get_double(), f"Rate {rate} is out of bounds" 0246 self.player_properties["Rate"] = GLib.Variant('d', rate) 0247 changed_properties = GLib.Variant('a{sv}', { 0248 "Rate": self.player_properties["Rate"], 0249 }) 0250 Gio.DBusConnection.emit_signal(connection, None, object_path, "org.freedesktop.DBus.Properties", "PropertiesChanged", GLib.Variant.new_tuple(self.PLAYER_IFACE, changed_properties, GLib.Variant('as', ()))) 0251 0252 def set_shuffle(self, shuffle: bool, connection: Gio.DBusConnection, object_path: str) -> None: 0253 """ 0254 Changes the shuffle state 0255 """ 0256 assert isinstance(shuffle, bool), f"argument is not a boolean but {type(shuffle)}" 0257 self.player_properties["Shuffle"] = GLib.Variant('b', shuffle) 0258 changed_properties = GLib.Variant('a{sv}', { 0259 "Shuffle": self.player_properties["Shuffle"], 0260 }) 0261 Gio.DBusConnection.emit_signal(connection, None, object_path, "org.freedesktop.DBus.Properties", "PropertiesChanged", GLib.Variant.new_tuple(self.PLAYER_IFACE, changed_properties, GLib.Variant('as', ()))) 0262 0263 def set_repeat(self, repeat: str, connection: Gio.DBusConnection, object_path: str) -> None: 0264 """ 0265 Changes the loop state 0266 """ 0267 assert isinstance(repeat, str), f"argument is not a string but {type(repeat)}" 0268 self.player_properties["LoopStatus"] = GLib.Variant('s', repeat) 0269 changed_properties = GLib.Variant('a{sv}', { 0270 "LoopStatus": self.player_properties["LoopStatus"], 0271 }) 0272 Gio.DBusConnection.emit_signal(connection, None, object_path, "org.freedesktop.DBus.Properties", "PropertiesChanged", GLib.Variant.new_tuple(self.PLAYER_IFACE, changed_properties, GLib.Variant('as', ()))) 0273 0274 def set_volume(self, volume: float, connection: Gio.DBusConnection, object_path: str) -> None: 0275 """ 0276 Adjusts the volume 0277 """ 0278 assert isinstance(volume, float) and 0 <= volume <= 1, f"Invalid volume {volume} of type {type(volume)}" 0279 self.player_properties["Volume"] = GLib.Variant('d', volume) 0280 changed_properties = GLib.Variant('a{sv}', { 0281 "Volume": self.player_properties["Volume"], 0282 }) 0283 Gio.DBusConnection.emit_signal(connection, None, object_path, "org.freedesktop.DBus.Properties", "PropertiesChanged", GLib.Variant.new_tuple(self.PLAYER_IFACE, changed_properties, GLib.Variant('as', ()))) 0284 0285 def player_handle_method_call(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, method_name: str, parameters: GLib.Variant, invocation: Gio.DBusMethodInvocation) -> None: 0286 """ 0287 Handles method calls for org.mpris.MediaPlayer2.Player 0288 """ 0289 assert interface_name == "org.mpris.MediaPlayer2.Player", f"Wrong interface name {interface_name} from {sender}" 0290 0291 print(f"player_handle_method_call method_name: {method_name}", file=sys.stderr, flush=True) 0292 0293 if method_name == "Next" or method_name == "Previous": 0294 self.current_index += 1 if method_name == "Next" else -1 0295 assert 0 <= self.current_index < len(self.metadata) 0296 self.player_properties["Metadata"] = GLib.Variant('a{sv}', self.metadata[self.current_index]) 0297 self.player_properties["CanGoPrevious"] = GLib.Variant('b', self.current_index > 0) 0298 self.player_properties["CanGoNext"] = GLib.Variant('b', self.current_index < len(self.metadata) - 1) 0299 changed_properties = GLib.Variant('a{sv}', { 0300 "Metadata": self.player_properties["Metadata"], 0301 "CanGoPrevious": self.player_properties["CanGoPrevious"], 0302 "CanGoNext": self.player_properties["CanGoNext"], 0303 }) 0304 Gio.DBusConnection.emit_signal(connection, None, object_path, "org.freedesktop.DBus.Properties", "PropertiesChanged", GLib.Variant.new_tuple(self.PLAYER_IFACE, changed_properties, GLib.Variant('as', ()))) 0305 self.metadata_updated_event.set() 0306 0307 elif method_name == "Pause": 0308 self.set_playing(False, connection, object_path) 0309 0310 elif method_name == "PlayPause": 0311 self.set_playing(self.player_properties["PlaybackStatus"].get_string() != "Playing", connection, object_path) 0312 0313 elif method_name == "Stop": 0314 self.stop(connection, object_path) 0315 0316 elif method_name == "Play": 0317 self.set_playing(True, connection, object_path) 0318 0319 elif method_name == "Seek": 0320 offset: int = parameters[0] 0321 length: int = int(self.metadata[self.current_index]["mpris:length"]) 0322 position: int = self.player_properties["Position"].get_int64() 0323 assert 0 <= position + offset <= length 0324 self.player_properties["Position"] = GLib.Variant('x', position + offset) 0325 changed_properties = GLib.Variant('a{sv}', { 0326 'Position': self.player_properties["Position"], 0327 }) 0328 Gio.DBusConnection.emit_signal(connection, None, object_path, self.PLAYER_IFACE.get_string(), "Seeked", GLib.Variant.new_tuple(self.player_properties["Position"])) 0329 Gio.DBusConnection.emit_signal(connection, None, object_path, "org.freedesktop.DBus.Properties", "PropertiesChanged", GLib.Variant.new_tuple(self.PLAYER_IFACE, changed_properties, GLib.Variant('as', ()))) 0330 0331 elif method_name == "SetPosition": 0332 assert parameters[0] == self.metadata[self.current_index]["mpris:trackid"].get_string(), f"expected trackid: {parameters[0]}, actual trackid: {self.metadata[self.current_index]['mpris:trackid'].get_string()}" 0333 self.player_properties["Position"] = GLib.Variant('x', parameters[1]) 0334 changed_properties = GLib.Variant('a{sv}', { 0335 'Position': self.player_properties["Position"], 0336 }) 0337 Gio.DBusConnection.emit_signal(connection, None, object_path, "org.freedesktop.DBus.Properties", "PropertiesChanged", GLib.Variant.new_tuple(self.PLAYER_IFACE, changed_properties, GLib.Variant('as', ()))) 0338 0339 elif method_name == "OpenUri": 0340 print("OpenUri", file=sys.stderr, flush=True) 0341 0342 else: 0343 # In case the interface adds new methods, fail here for easier discovery 0344 logging.error("%s does not exist", method_name) 0345 invocation.return_error_literal(Gio.dbus_error_quark(), Gio.DBusError.NOT_SUPPORTED, f"Invalid method {method_name}") 0346 0347 def player_handle_get_property(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, value: Any): 0348 """ 0349 Handles properties for org.mpris.MediaPlayer2.Player 0350 """ 0351 assert interface_name == "org.mpris.MediaPlayer2.Player", f"Wrong interface name {interface_name} from {sender}" 0352 assert value in self.player_properties.keys(), f"{value} does not exist" 0353 0354 return self.player_properties[value] 0355 0356 def player_handle_set_property(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, key: str, value: Any) -> bool: 0357 """ 0358 Handles properties for org.mpris.MediaPlayer2.Player 0359 """ 0360 assert interface_name == "org.mpris.MediaPlayer2.Player", f"Wrong interface name {interface_name} from {sender}" 0361 assert key in self.player_properties.keys(), f"{key} does not exist" 0362 0363 print(f"player_handle_set_property key: {key}, value: {value}", file=sys.stderr, flush=True) 0364 0365 if key == "Rate": 0366 self.set_rate(value, connection, object_path) 0367 elif key == "LoopStatus": 0368 self.set_repeat(value, connection, object_path) 0369 elif key == "Shuffle": 0370 self.set_shuffle(value, connection, object_path) 0371 elif key == "Volume": 0372 self.set_volume(value, connection, object_path) 0373 else: 0374 return False 0375 0376 # What is the correct thing to return here on success? It appears that 0377 # we need to return something other than None or what would be evaluated 0378 # to False for this call back to be successful. 0379 return True 0380 0381 def interface_handle_method_call(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, method_name: str, parameters: GLib.Variant, invocation: Gio.DBusMethodInvocation) -> None: 0382 """ 0383 Handles method calls for org.mpris.MediaPlayer2 0384 """ 0385 assert interface_name == "org.mpris.MediaPlayer2", f"Wrong interface name {interface_name} from {sender}" 0386 0387 if method_name == "Raise": 0388 print("Raise", file=sys.stderr, flush=True) 0389 elif method_name == "Quit": 0390 print("Quit", file=sys.stderr, flush=True) 0391 else: 0392 logging.error("%s does not exist", method_name) 0393 invocation.return_error_literal(Gio.dbus_error_quark(), Gio.DBusError.NOT_SUPPORTED, f"Invalid method {method_name}") 0394 0395 def interface_handle_get_property(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, value: Any): 0396 """ 0397 Handles properties for org.mpris.MediaPlayer2 0398 """ 0399 assert interface_name == "org.mpris.MediaPlayer2", f"Wrong interface name {interface_name} from {sender}" 0400 assert value in self.base_properties.keys(), f"{value} does not exist" 0401 0402 return self.base_properties[value] 0403 0404 def interface_handle_set_property(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, key: str, value: Any) -> bool: 0405 """ 0406 Handles properties for org.mpris.MediaPlayer2 0407 """ 0408 assert interface_name == "org.mpris.MediaPlayer2", f"Wrong interface name {interface_name} from {sender}" 0409 assert key in self.base_properties.keys(), f"{key} does not exist" 0410 0411 return True 0412 0413 0414 class InvalidMpris2(Mpris2): 0415 """ 0416 MPRIS2 interfaces with invalid responses 0417 """ 0418 0419 def __init__(self) -> None: 0420 super().__init__([], {}, {}, 0) 0421 0422 def properties_handle_method_call(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, method_name: str, parameters: GLib.Variant, invocation: Gio.DBusMethodInvocation) -> None: 0423 """ 0424 @override 0425 """ 0426 invocation.return_error_literal(Gio.dbus_error_quark(), Gio.DBusError.NOT_SUPPORTED, "") 0427 0428 def player_handle_method_call(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, method_name: str, parameters: GLib.Variant, invocation: Gio.DBusMethodInvocation) -> None: 0429 """ 0430 @override 0431 """ 0432 invocation.return_error_literal(Gio.dbus_error_quark(), Gio.DBusError.NOT_SUPPORTED, "") 0433 0434 def player_handle_get_property(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, value: Any): 0435 """ 0436 @override 0437 """ 0438 return None 0439 0440 def player_handle_set_property(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, key: str, value: Any) -> bool: 0441 """ 0442 @override 0443 """ 0444 return False 0445 0446 def interface_handle_method_call(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, method_name: str, parameters: GLib.Variant, invocation: Gio.DBusMethodInvocation) -> None: 0447 """ 0448 @override 0449 """ 0450 invocation.return_error_literal(Gio.dbus_error_quark(), Gio.DBusError.NOT_SUPPORTED, "") 0451 0452 def interface_handle_get_property(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, value: Any): 0453 """ 0454 @override 0455 """ 0456 return None 0457 0458 def interface_handle_set_property(self, connection: Gio.DBusConnection, sender: str, object_path: str, interface_name: str, key: str, value: Any) -> bool: 0459 """ 0460 @override 0461 """ 0462 return False 0463 0464 0465 player: Mpris2 0466 loopThread: GLibMainLoopThread 0467 0468 0469 def __on_terminate(signal, frame) -> None: 0470 player.quit() 0471 loopThread.quit() 0472 if loopThread.is_alive(): 0473 loopThread.join(timeout=10) 0474 0475 0476 if __name__ == '__main__': 0477 assert len(sys.argv) >= 2, "Insufficient arguments" 0478 signal.signal(signal.SIGTERM, __on_terminate) 0479 0480 json_path: str = sys.argv.pop() 0481 with open(json_path, "r", encoding="utf-8") as f: 0482 json_dict: dict[str, list | dict] = json.load(f) 0483 metadata: list[dict[str, GLib.Variant]] = read_player_metadata(json_dict) 0484 base_properties: dict[str, GLib.Variant] = read_base_properties(json_dict) 0485 current_index: int = 0 0486 player_properties: dict[str, GLib.Variant] = read_player_properties(json_dict, metadata[current_index]) 0487 0488 loopThread = GLibMainLoopThread() 0489 loopThread.start() 0490 player = Mpris2(metadata, base_properties, player_properties, current_index) 0491 logging.info("Player %s started", base_properties['Identity'].get_string())