# Source code for bleak.backends.bluezdbus.signals

    from __future__ import annotations

    import sys
    from typing import TYPE_CHECKING

    if TYPE_CHECKING:
        if sys.platform != "linux":
            assert False, "This backend is only available on Linux"

    import re
    from typing import Any, Optional

    from dbus_fast.aio.message_bus import MessageBus
    from dbus_fast.errors import InvalidObjectPathError
    from dbus_fast.message import Message
    from dbus_fast.validators import (
        assert_interface_name_valid,
        assert_member_name_valid,
        assert_object_path_valid,
    )

    # TODO: this stuff should be improved and submitted upstream to dbus-next
    # https://github.com/altdesktop/python-dbus-next/issues/53

    _message_types = ["signal", "method_call", "method_return", "error"]

    class InvalidMessageTypeError(TypeError):
        def __init__(self, type: str):
            super().__init__(f"invalid message type: {type}")

    def is_message_type_valid(type: str) -> bool:
        """Whether this is a valid message type.

        .. seealso:: https://dbus.freedesktop.org/doc/dbus-specification.html#message-bus-routing-match-rules

        :param type: The message type to validate.
        :type name: str

        :returns: Whether the name is a valid message type.
        :rtype: bool
        """
        return type in _message_types

    def assert_bus_name_valid(type: str) -> None:
        """Raise an error if this is not a valid message type.

        .. seealso:: https://dbus.freedesktop.org/doc/dbus-specification.html#message-bus-routing-match-rules

        :param type: The message type to validate.
        :type name: str

        :raises:
            - :class:`InvalidMessageTypeError` - If this is not a valid message type.
        """
        if not is_message_type_valid(type):
            raise InvalidMessageTypeError(type)

    class MatchRules:
        """D-Bus signal match rules.

        .. seealso:: https://dbus.freedesktop.org/doc/dbus-specification.html#message-bus-routing-match-rules
        """

        def __init__(
            self,
            type: str = "signal",
            sender: Optional[str] = None,
            interface: Optional[str] = None,
            member: Optional[str] = None,
            path: Optional[str] = None,
            path_namespace: Optional[str] = None,
            destination: Optional[str] = None,
            arg0namespace: Optional[str] = None,
            **kwargs: Any,
        ):
            assert_bus_name_valid(type)
            self.type: str = type

            if sender:
                assert_bus_name_valid(sender)
                self.sender: Optional[str] = sender
            else:
                self.sender = None

            if interface:
                assert_interface_name_valid(interface)
                self.interface: Optional[str] = interface
            else:
                self.interface = None

            if member:
                assert_member_name_valid(member)
                self.member: Optional[str] = member
            else:
                self.member = None

            if path:
                assert_object_path_valid(path)
                self.path: Optional[str] = path
            else:
                self.path = None

            if path_namespace:
                assert_object_path_valid(path_namespace)
                self.path_namespace: Optional[str] = path_namespace
            else:
                self.path_namespace = None

            if path and path_namespace:
                raise TypeError(
                    "message rules cannot have both 'path' and 'path_namespace' at the same time"
                )

            if destination:
                assert_bus_name_valid(destination)
                self.destination: Optional[str] = destination
            else:
                self.destination = None

            if arg0namespace:
                assert_bus_name_valid(arg0namespace)
                self.arg0namespace: Optional[str] = arg0namespace
            else:
                self.arg0namespace = None

            if kwargs:
                for k, v in kwargs.items():
                    if re.match(r"^arg\d+$", k):
                        if not isinstance(v, str):
                            raise TypeError(f"kwarg '{k}' must have a str value")
                    elif re.match(r"^arg\d+path$", k):
                        if not isinstance(v, str):
                            raise InvalidObjectPathError(v)
                        assert_object_path_valid(v[:-1] if v.endswith("/") else v)
                    else:
                        raise ValueError("kwargs must be in the form 'arg0' or 'arg0path'")
                self.args: Optional[dict[str, str]] = kwargs
            else:
                self.args = None

        @staticmethod
        def parse(rules: str) -> MatchRules:
            return MatchRules(**dict(r.split("=") for r in rules.split(",")))

        def __str__(self) -> str:
            rules = [f"type={self.type}"]

            if self.sender:
                rules.append(f"sender={self.sender}")

            if self.interface:
                rules.append(f"interface={self.interface}")

            if self.member:
                rules.append(f"member={self.member}")

            if self.path:
                rules.append(f"path={self.path}")

            if self.path_namespace:
                rules.append(f"path_namespace={self.path_namespace}")

            if self.destination:
                rules.append(f"destination={self.destination}")

            if self.args:
                for k, v in self.args.items():
                    rules.append(f"{k}={v}")

            if self.arg0namespace:
                rules.append(f"arg0namespace={self.arg0namespace}")

            return ",".join(rules)

        def __repr__(self) -> str:
            return f"MatchRules({self})"

    async def add_match(bus: MessageBus, rules: MatchRules) -> Message:
        """Calls org.freedesktop.DBus.AddMatch using ``rules``."""
        reply = await bus.call(
            Message(
                destination="org.freedesktop.DBus",
                interface="org.freedesktop.DBus",
                path="/org/freedesktop/DBus",
                member="AddMatch",
                signature="s",
                body=[str(rules)],
            )
        )

        return reply

    async def remove_match(bus: MessageBus, rules: MatchRules) -> Message:
        """Calls org.freedesktop.DBus.RemoveMatch using ``rules``."""
        reply = await bus.call(
            Message(
                destination="org.freedesktop.DBus",
                interface="org.freedesktop.DBus",
                path="/org/freedesktop/DBus",
                member="RemoveMatch",
                signature="s",
                body=[str(rules)],
            )
        )

        return reply