forked from Wobbl/Wobuzz
97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
|
#!/usr/bin/python3
|
||
|
|
|
||
|
|
from jeepney import DBusAddress, Message, MessageType, HeaderFields, new_error, new_signal
|
||
|
|
|
||
|
|
|
||
|
|
SERVICE_NAME = "org.mpris.MediaPlayer2.wobuzz"
|
||
|
|
OBJECT_PATH = "/org/mpris/MediaPlayer2"
|
||
|
|
PROPERTIES_INTERFACE = "org.freedesktop.DBus.Properties"
|
||
|
|
MPRIS_ROOT_INTERFACE = "org.mpris.MediaPlayer2"
|
||
|
|
MPRIS_PLAYER_INTERFACE = "org.mpris.MediaPlayer2.Player"
|
||
|
|
|
||
|
|
|
||
|
|
class DBusErrors:
|
||
|
|
@classmethod
|
||
|
|
def unknownMethod(cls, method: str) -> tuple[str, str, tuple[str]]:
|
||
|
|
return "org.freedesktop.DBus.Error.UnknownMethod", "s", (f"No such method '{method}'.",)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def unknownProperty(cls, prop: str) -> tuple[str, str, tuple[str]]:
|
||
|
|
return "org.freedesktop.DBus.Error.InvalidArgs", "s", (f"No such property '{prop}'",)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def invalidArgs(cls, prop: str=None, interface: str=None):
|
||
|
|
if prop is None and interface is None:
|
||
|
|
return "org.freedesktop.DBus.Error.InvalidArgs"
|
||
|
|
|
||
|
|
if interface is None:
|
||
|
|
return "org.freedesktop.DBus.Error.InvalidArgs", "s", (f"No such property '{prop}'.",)
|
||
|
|
|
||
|
|
if prop is None:
|
||
|
|
return "org.freedesktop.DBus.Error.InvalidArgs", "s", (f"No such interface '{interface}'.",)
|
||
|
|
|
||
|
|
return (
|
||
|
|
"org.freedesktop.DBus.Error.InvalidArgs",
|
||
|
|
"s",
|
||
|
|
(f"No such property '{prop}' on interface '{interface}'.",)
|
||
|
|
)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def unknownObject(cls, path: str):
|
||
|
|
return "org.freedesktop.DBus.Error.UnknownObject", "s", (f"No such object on path '{path}'.",)
|
||
|
|
|
||
|
|
|
||
|
|
class DBusInterface:
|
||
|
|
def __init__(self, server, interface: str):
|
||
|
|
self.server = server
|
||
|
|
self.interface = interface
|
||
|
|
|
||
|
|
def __setattr__(self, key: str, value) -> None:
|
||
|
|
super().__setattr__(key, value)
|
||
|
|
|
||
|
|
if not key[0].isupper() and not callable(value) and hasattr(self, key.title()):
|
||
|
|
getter = getattr(self, key.title())
|
||
|
|
|
||
|
|
if callable(getter):
|
||
|
|
if hasattr(self.server, "bus"):
|
||
|
|
self.server.properties_interface.properties_changed(self.interface)
|
||
|
|
|
||
|
|
def handle_message(self, msg: Message):
|
||
|
|
return_msg = None
|
||
|
|
|
||
|
|
match msg.header.message_type:
|
||
|
|
case MessageType.method_call:
|
||
|
|
method_name: str = msg.header.fields[HeaderFields.member]
|
||
|
|
|
||
|
|
if hasattr(self, method_name) and method_name[0].isupper():
|
||
|
|
method = getattr(self, msg.header.fields[HeaderFields.member])
|
||
|
|
|
||
|
|
if callable(method):
|
||
|
|
return_msg = method(msg)
|
||
|
|
|
||
|
|
else:
|
||
|
|
return_msg = new_error(msg, *DBusErrors.unknownMethod(method_name))
|
||
|
|
|
||
|
|
else:
|
||
|
|
return_msg = new_error(msg, *DBusErrors.unknownMethod(method_name))
|
||
|
|
|
||
|
|
if return_msg is not None:
|
||
|
|
self.server.bus.send_message(return_msg)
|
||
|
|
|
||
|
|
def get(self, msg: Message):
|
||
|
|
prop_name: str = msg.body[1]
|
||
|
|
|
||
|
|
if prop_name[0].isupper() and hasattr(self, prop_name):
|
||
|
|
prop = getattr(self, prop_name)
|
||
|
|
|
||
|
|
if callable(prop):
|
||
|
|
return prop()
|
||
|
|
|
||
|
|
return new_error(msg, *DBusErrors.unknownProperty(prop_name))
|
||
|
|
|
||
|
|
else:
|
||
|
|
return new_error(msg, *DBusErrors.unknownProperty(prop_name))
|
||
|
|
|
||
|
|
def get_all(self) -> tuple[dict[str: tuple[str: any]]]:
|
||
|
|
raise NotImplementedError
|