# SPDX-License-Identifier: LGPL-2.1-or-later

# Copyright (C) 2020, 2021 igo95862

# This file is part of python-sdbus

# This library is free software; you can redistribute it and/or

# modify it under the terms of the GNU Lesser General Public

# License as published by the Free Software Foundation; either

# version 2.1 of the License, or (at your option) any later version.

# This library is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU

# Lesser General Public License for more details.

# You should have received a copy of the GNU Lesser General Public

# License along with this library; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

from __future__ import annotations

from pathlib import Path

from typing import TYPE_CHECKING

from xml.etree.ElementTree import fromstring as etree_from_str

from xml.etree.ElementTree import parse as etree_from_file

from .dbus_common_funcs import snake_case_to_camel_case

if TYPE_CHECKING:

from collections.abc import Iterable, Iterator

from typing import Literal, Optional, Union

from xml.etree.ElementTree import Element

def _camel_case_to_snake_case_generator(camel: str) -> Iterator[str]:

i = iter(camel)

# First character is directly converted to low

try:

first_char = next(i)

except StopIteration:

raise ValueError('Name too short')

yield first_char.lower()

last_character = first_char

# Yield every character

# if upper is encountered

# yield _ only if previous character

# was not already uppercase or underscore

while True:

try:

c = next(i)

except StopIteration:

return

if c.isupper():

if not last_character.isupper() and not last_character == "_":

yield '_'

yield c.lower()

else:

yield c

last_character = c

def camel_case_to_snake_case(camel: str) -> str:

return ''.join(_camel_case_to_snake_case_generator(camel))

def _iterface_name_to_class_generator(interface_name: str) -> Iterator[str]:

i = iter(interface_name)

try:

first_char = next(i)

except StopIteration:

raise ValueError('Interface name too short')

yield first_char.upper()

up_next_char = False

for _ in range(120):

try:

c = next(i)

except StopIteration:

return

if c == '.':

up_next_char = True

else:

if up_next_char:

yield c.upper()

else:

yield c

up_next_char = False

def interface_name_to_class(interface_name: str) -> str:

return ''.join(_iterface_name_to_class_generator(interface_name))

def parse_str_bool(annotation_value: str) -> bool:

if annotation_value == 'true':

return True

elif annotation_value == 'false':

return False

else:

raise ValueError(f"Unknown bool value: {annotation_value}")

class DbusSigToTyping:

_DBUS_BASIC_SIG_TO_TYPING = {

'y': 'int',

'b': 'bool',

'n': 'int',

'q': 'int',

'i': 'int',

'u': 'int',

'x': 'int',

't': 'int',

'd': 'float',

's': 'str',

'o': 'str',

'g': 'str',

'h': 'int',

}

@classmethod

def typing_basic(cls, char: str) -> str:

return cls._DBUS_BASIC_SIG_TO_TYPING[char]

@staticmethod

def typing_into_tuple(typing_iter: Iterable[str]) -> str:

return f"tuple[{', '.join(typing_iter)}]"

@staticmethod

def slice_container(dbus_sig_iter: Iterator[str], peek_str: str) -> str:

accumulator: list[str] = [peek_str]

round_braces_count = 0

curly_braces_count = 0

if peek_str == '(':

round_braces_count += 1

if peek_str == '{':

curly_braces_count += 1

while True:

try:

char = next(dbus_sig_iter)

except StopIteration:

break

if char == ')':

round_braces_count -= 1

if char == '}':

curly_braces_count -= 1

if char == '(':

round_braces_count += 1

if char == '{':

curly_braces_count += 1

accumulator.append(char)

if char == 'a':

continue

if round_braces_count == 0 and curly_braces_count == 0:

break

return ''.join(accumulator)

@classmethod

def split_sig(cls, sig: str) -> list[str]:

completes: list[str] = []

sig_iter = iter(sig)

while True:

try:

next_char = next(sig_iter)

except StopIteration:

break

if next_char == '(':

next_complete = cls.slice_container(sig_iter, '(')

elif next_char == 'a':

next_complete = cls.slice_container(sig_iter, 'a')

else:

next_complete = next_char

completes.append(next_complete)

return completes

@classmethod

def typing_complete(cls, complete_sig: str) -> str:

if complete_sig == 'v':

return cls.typing_into_tuple(('str', 'Any'))

elif complete_sig == 'ay':

return 'bytes'

elif complete_sig.startswith('a{'):

if complete_sig[-1] != '}':

raise ValueError(f"Malformed dict {complete_sig}")

dict_key_sig = complete_sig[2]

dict_key_typing = cls.typing_basic(dict_key_sig)

dict_value_sig = complete_sig[3:-1]

dict_value_typing = cls.typing_complete(dict_value_sig)

return f"dict[{dict_key_typing}, {dict_value_typing}]"

elif complete_sig.startswith('a'):

array_completes = cls.split_sig(complete_sig[1:])

if len(array_completes) != 1:

raise ValueError("Array does not have only "

f"one complete type: {array_completes}")

array_single_complete = array_completes[0]

return f"list[{cls.typing_complete(array_single_complete)}]"

elif complete_sig.startswith('('):

if complete_sig[-1] != ')':

raise ValueError(f"Malformed struct {complete_sig}")

struct_completes = cls.split_sig(complete_sig[1:-1])

struct_typing = (cls.typing_complete(x) for x in struct_completes)

return cls.typing_into_tuple(struct_typing)

else:

return cls.typing_basic(complete_sig)

@classmethod

def result_typing(cls, result_args: list[str]) -> str:

result_len = len(result_args)

if result_len == 0:

return 'None'

elif result_len == 1:

return cls.typing_complete(result_args[0])

else:

return cls.typing_into_tuple(

cls.typing_complete(x) for x in result_args

)

@classmethod

def sig_to_typing(cls, signature: str) -> str:

return cls.result_typing(cls.split_sig(signature))

class DbusMemberAbstract:

def __init__(self, element: Element):

self.method_name = element.attrib['name']

self.python_name = camel_case_to_snake_case(self.method_name)

self.is_deprecated = False

self.is_priveledged = False

self.iter_sub_elements(element)

def _can_use_unpivileged(self) -> bool:

return True

def _flags_iter(self) -> Iterator[str]:

if self.is_deprecated:

yield 'DbusDeprecatedFlag'

if not self.is_priveledged and self._can_use_unpivileged():

yield 'DbusUnprivilegedFlag'

@property

def flags_str(self) -> str:

return ' | '.join(self._flags_iter())

def _parse_arg(self, arg: Element) -> None:

raise NotImplementedError('Member does not have arguments')

def _parse_annotation_data(self,

annotation_name: str,

annotation_value: str) -> None:

if annotation_name == 'org.freedesktop.DBus.Deprecated':

self.is_deprecated = parse_str_bool(annotation_value)

elif annotation_name == 'org.freedesktop.systemd1.Privileged':

self.is_priveledged = parse_str_bool(annotation_value)

else:

...

def _parse_annotation(self, annotation: Element) -> None:

if annotation.tag != 'annotation':

raise ValueError('Uknown element of member: ', annotation.tag)

annotation_name = annotation.attrib['name']

annotation_value = annotation.attrib['value']

self._parse_annotation_data(annotation_name, annotation_value)

def iter_sub_elements(self, element: Element) -> None:

for sub_element in element:

tag = sub_element.tag

if tag == 'annotation':

self._parse_annotation(sub_element)

elif tag == 'arg':

self._parse_arg(sub_element)

else:

raise ValueError(

'Uknown member annotation tag: ', tag)

@property

def wants_rename(self) -> bool:

return self.method_name != snake_case_to_camel_case(self.python_name)

class DbusArgsIntrospection:

def __init__(self, element: Element):

if element.tag != 'arg':

raise ValueError(f"Expected arg tag, got {element.tag}")

try:

self.name: Optional[str] = element.attrib['name']

except KeyError:

self.name = None

self.dbus_type = element.attrib['type']

direction = element.attrib.get('direction')

if direction == 'in':

self.is_input: Optional[bool] = True

elif direction == 'out':

self.is_input = False

elif direction is None:

self.is_input = None

else:

raise ValueError(f'Unknown arg direction {direction}')

@property

def typing(self) -> str:

return DbusSigToTyping.typing_complete(self.dbus_type)

def __repr__(self) -> str:

return (f"D-Bus Arg: {self.name}, "

f"type: {self.dbus_type}, "

f"is input: {self.is_input}")

class DbusMethodInrospection(DbusMemberAbstract):

def __init__(self, element: Element):

if element.tag != 'method':

raise ValueError(f"Expected method tag, got {element.tag}")

self.is_no_reply = False

self.input_args: list[DbusArgsIntrospection] = []

self.result_args: list[DbusArgsIntrospection] = []

super().__init__(element)

def _flags_iter(self) -> Iterator[str]:

if self.is_no_reply:

yield 'DbusNoReplyFlag'

yield from super()._flags_iter()

def _parse_arg(self, arg: Element) -> None:

new_arg = DbusArgsIntrospection(arg)

if new_arg.is_input or new_arg.is_input is None:

self.input_args.append(new_arg)

elif not new_arg.is_input:

self.result_args.append(new_arg)

else:

raise ValueError('Malformed arg direction')

@property

def dbus_input_signature(self) -> str:

return ''.join(

x.dbus_type for x in self.input_args

)

@property

def dbus_result_signature(self) -> str:

return ''.join(

x.dbus_type if not x.is_input else ''

for x in self.result_args

)

@property

def args_names_and_typing(self) -> list[tuple[str, str]]:

arg_names: list[tuple[str, str]] = []

for i, input_arg in enumerate(self.input_args):

if input_arg.name is not None:

input_arg_name = camel_case_to_snake_case(input_arg.name)

else:

input_arg_name = f"arg_{i}"

arg_names.append((input_arg_name, input_arg.typing))

return arg_names

@property

def result_typing(self) -> str:

return DbusSigToTyping.result_typing(

[x.dbus_type for x in self.result_args])

@property

def is_results_args_valid_names(self) -> bool:

return all(r.name is not None for r in self.result_args)

@property

def result_args_names_repr(self) -> str:

return repr(tuple(r.name for r in self.result_args))

def __repr__(self) -> str:

return (f"D-Bus Method: {self.method_name}, "

f"args: {self.args_names_and_typing}, "

f"result: {self.dbus_result_signature}")

class DbusPropertyIntrospection(DbusMemberAbstract):

_EMITS_CHANGED_MAP: dict[

Union[bool, Literal['const', 'invalidates']], str

] = {

True: 'DbusPropertyEmitsChangeFlag',

'invalidates': 'DbusPropertyEmitsInvalidationFlag',

'const': 'DbusPropertyConstFlag',

}

def __init__(self, element: Element):

if element.tag != 'property':

raise ValueError(f"Expected property tag, got {element.tag}")

self.dbus_signature = element.attrib['type']

self.emits_changed: Union[bool, Literal['const', 'invalidates']] = True

self.is_explicit = False

access_type = element.attrib['access']

if access_type == 'readwrite' or access_type == 'write':

self.is_read_only = False

elif access_type == 'read':

self.is_read_only = True

else:

raise ValueError(f"Unknown property access {access_type}")

super().__init__(element)

def _can_use_unpivileged(self) -> bool:

# Only properties that have setters defined can use the

# unprivileged flags. The code generator does NOT generate

# setters.

return False

def _flags_iter(self) -> Iterator[str]:

emits_changed_str = self._EMITS_CHANGED_MAP.get(self.emits_changed)

if emits_changed_str is not None:

yield emits_changed_str

yield from super()._flags_iter()

def _parse_annotation_data(self,

annotation_name: str,

annotation_value: str) -> None:

if annotation_name == ('org.freedesktop.DBus.Property'

'.EmitsChangedSignal'):

if annotation_value == 'true':

self.emits_changed = True

elif annotation_value == 'false':

self.emits_changed = False

elif annotation_value == 'const':

self.emits_changed = 'const'

elif annotation_value == 'invalidates':

self.emits_changed = 'invalidates'

else:

raise ValueError('Unknown EmitsChanged value',

annotation_value)

elif annotation_name == 'org.freedesktop.systemd1.Explicit':

self.is_explicit = parse_str_bool(annotation_value)

super()._parse_annotation_data(annotation_name, annotation_value)

@property

def typing(self) -> str:

return DbusSigToTyping.typing_complete(self.dbus_signature)

class DbusSignalIntrospection(DbusMemberAbstract):

def __init__(self, element: Element):

if element.tag != 'signal':

raise ValueError(f"Expected signal tag, got {element.tag}")

self.args: list[DbusArgsIntrospection] = []

super().__init__(element)

def _can_use_unpivileged(self) -> bool:

return False

def _parse_arg(self, arg: Element) -> None:

new_arg = DbusArgsIntrospection(arg)

if new_arg.is_input:

raise ValueError('Signal argument cannot be in', new_arg)

self.args.append(new_arg)

@property

def dbus_signature(self) -> str:

return ''.join(x.dbus_type for x in self.args)

@property

def typing(self) -> str:

return DbusSigToTyping.result_typing(

[x.dbus_type for x in self.args])

@property

def is_args_valid_names(self) -> bool:

return all(a.name is not None for a in self.args)

@property

def args_names_repr(self) -> str:

return repr(tuple(a.name for a in self.args))

class DbusInterfaceIntrospection:

def __init__(self, element: Element):

if element.tag != 'interface':

raise ValueError(f"Expected interface tag, got {element.tag}")

self.interface_name = element.attrib['name']

self.python_name = interface_name_to_class(

self.interface_name) + 'Interface'

self.is_deprecated = False

self.c_name: Optional[str] = None

self.methods: list[DbusMethodInrospection] = []

self.properties: list[DbusPropertyIntrospection] = []

self.signals: list[DbusSignalIntrospection] = []

for dbus_member in element:

if dbus_member.tag == 'method':

self.methods.append(DbusMethodInrospection(dbus_member))

elif dbus_member.tag == 'property':

self.properties.append(DbusPropertyIntrospection(dbus_member))

elif dbus_member.tag == 'signal':

self.signals.append(DbusSignalIntrospection(dbus_member))

elif dbus_member.tag == 'annotation':

annotation_name = dbus_member.attrib['name']

annotation_value = dbus_member.attrib['value']

if annotation_name == 'org.freedesktop.DBus.Deprecated':

self.is_deprecated = parse_str_bool(annotation_value)

elif annotation_name == 'org.freedesktop.DBus.GLib.CSymbol':

self.c_name = annotation_value

else:

...

else:

raise ValueError(f'Unknown D-Bus member {dbus_member}')

@property

def has_members(self) -> bool:

return any((self.methods, self.properties, self.signals))

SKIP_INTERFACES = {

'org.freedesktop.DBus.Properties',

'org.freedesktop.DBus.Introspectable',

'org.freedesktop.DBus.Peer',

'org.freedesktop.DBus.ObjectManager',

}

INTERFACE_TEMPLATES: dict[str, str] = {

"generic_no_members": """\

... # Interface has no members

""",

"generic_method_flags": (

"""\

{% if method.dbus_input_signature %}

input_signature="{{ method.dbus_input_signature }}",

{% endif %}

{% if method.dbus_result_signature %}

result_signature="{{ method.dbus_result_signature }}",

{% endif %}

{% if method.flags_str %}

flags={{ method.flags_str }},

{% endif %}

{% if method.wants_rename %}

method_name="{{method.method_name}}",

{% endif %}

"""

),

"generic_property_flags": (

"""\

{% if a_property.dbus_signature %}

property_signature="{{ a_property.dbus_signature }}",

{% endif %}

{% if a_property.flags_str %}

flags={{ a_property.flags_str }},

{% endif %}

{% if a_property.wants_rename %}

property_name="{{a_property.method_name}}",

{% endif %}

"""

),

"generic_header": """\

from __future__ import annotations

from typing import Any

""",

"async_imports_header": """from sdbus import (

DbusDeprecatedFlag,

DbusInterfaceCommonAsync,

DbusNoReplyFlag,

DbusPropertyConstFlag,

DbusPropertyEmitsChangeFlag,

DbusPropertyEmitsInvalidationFlag,

DbusPropertyExplicitFlag,

DbusUnprivilegedFlag,

dbus_method_async,

dbus_property_async,

dbus_signal_async,

)

""",

"async_main": (

"""\

{% if include_import_header %}

{% include 'generic_header' %}

{% include 'async_imports_header' %}

{% endif %}

{% for interface in interfaces %}

{% include 'async_interface' %}

{% endfor %}

"""

),

"async_interface": (

"""\

class {{ interface.python_name }}(

DbusInterfaceCommonAsync,

interface_name="{{ interface.interface_name }}",

):

{% filter indent(first=True) %}

{% if interface.has_members %}

{% for method in interface.methods %}

{% include 'async_method' %}

{% endfor %}

{% for a_property in interface.properties %}

{% include 'async_property' %}

{% endfor %}

{% for signal in interface.signals %}

{% include 'async_signal' %}

{% endfor %}

{% else %}

{% include 'generic_no_members' %}

{% endif %}

{% endfilter %}

"""

),

"async_method": (

"""\

@dbus_method_async(

{% filter indent(first=True) %}

{% include 'generic_method_flags' %}

{% endfilter %}

{% if method.is_results_args_valid_names %}

result_args_names={{method.result_args_names_repr}},

{% endif %}

)

async def {{ method.python_name }}(

self,

{% for arg_name, arg_type in method.args_names_and_typing %}

{{ arg_name }}: {{ arg_type }},

{% endfor %}

) -> {{ method.result_typing }}:

raise NotImplementedError

"""

),

"async_property": (

"""\

@dbus_property_async(

{% filter indent(first=True) %}

{% include 'generic_property_flags' %}

{% endfilter %}

)

def {{ a_property.python_name }}(self) -> {{ a_property.typing }}:

raise NotImplementedError

"""

),

"async_signal": (

"""\

@dbus_signal_async(

{% if signal.dbus_signature %}

signal_signature="{{ signal.dbus_signature }}",

{% endif %}

{% if signal.is_args_valid_names %}

signal_args_names={{signal.args_names_repr}},

{% endif %}

{% if signal.flags_str %}

flags={{ signal.flags_str }},

{% endif %}

{% if signal.wants_rename %}

signal_name=signal.method_name,

{% endif %}

)

def {{ signal.python_name }}(self) -> {{ signal.typing }}:

raise NotImplementedError

"""

),

"blocking_imports_header": """\

from sdbus import (

DbusDeprecatedFlag,

DbusInterfaceCommon,

DbusNoReplyFlag,

DbusPropertyConstFlag,

DbusPropertyEmitsChangeFlag,

DbusPropertyEmitsInvalidationFlag,

DbusPropertyExplicitFlag,

DbusUnprivilegedFlag,

dbus_method,

dbus_property,

)

""",

"blocking_main": (

"""\

{% if include_import_header %}

{% include 'generic_header' %}

{% include 'blocking_imports_header' %}

{% endif %}

{% for interface in interfaces %}

{% include 'blocking_interface' %}

{% endfor %}

"""

),

"blocking_interface": (

"""\

class {{ interface.python_name }}(

DbusInterfaceCommon,

interface_name="{{ interface.interface_name }}",

):

{% filter indent(first=True) %}

{% if interface.has_members %}

{% for method in interface.methods %}

{% include 'blocking_method' %}

{% endfor %}

{% for a_property in interface.properties %}

{% include 'blocking_property' %}

{% endfor %}

{% else %}

{% include 'generic_no_members' %}

{% endif %}

{% endfilter %}

"""

),

"blocking_method": (

"""\

@dbus_method(

{% filter indent(first=True) %}

{% include 'generic_method_flags' %}

{% endfilter %}

)

def {{ method.python_name }}(

self,

{% for arg_name, arg_type in method.args_names_and_typing %}

{{ arg_name }}: {{ arg_type }},

{% endfor %}

) -> {{ method.result_typing }}:

raise NotImplementedError

"""

),

"blocking_property": (

"""\

@dbus_property(

{% filter indent(first=True) %}

{% include 'generic_property_flags' %}

{% endfilter %}

)

def {{ a_property.python_name }}(self) -> {{ a_property.typing }}:

raise NotImplementedError

"""

),

}

def xml_to_interfaces_introspection(

root: Element) -> list[DbusInterfaceIntrospection]:

list_of_interface_introspection: list[DbusInterfaceIntrospection] = []

if root.tag != 'node':

raise ValueError(f"Expected node tag got {root.tag}")

for interface in root:

if interface.tag == 'node':

continue

if interface.attrib['name'] in SKIP_INTERFACES:

continue

list_of_interface_introspection.append(

DbusInterfaceIntrospection(interface))

return list_of_interface_introspection

def interfaces_from_file(filename_or_path: Union[str, Path]

) -> list[DbusInterfaceIntrospection]:

etree = etree_from_file(filename_or_path)

return xml_to_interfaces_introspection(etree.getroot())

def interfaces_from_str(xml_str: str) -> list[DbusInterfaceIntrospection]:

etree = etree_from_str(xml_str)

return xml_to_interfaces_introspection(etree)

def generate_py_file(

interfaces: list[DbusInterfaceIntrospection],

include_import_header: bool = True,

do_async: bool = True,

) -> str:

from jinja2 import DictLoader

from jinja2 import Environment as JinjaEnv

template_name = "async_main" if do_async else "blocking_main"

env = JinjaEnv(

loader=DictLoader(INTERFACE_TEMPLATES),

trim_blocks=True,

lstrip_blocks=True,

autoescape=False,

)

return env.get_template(template_name).render(

interfaces=interfaces,

include_import_header=include_import_header,

)