alien-everywhere/shimming/alienaudioservice/alienaudio-shim.py

#!/usr/bin/python

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from enum import IntEnum, IntFlag
import logging
from typing import Any, Callable, Coroutine, Optional

from sdbus import (
    DbusInterfaceCommonAsync,
    dbus_method_async,
    dbus_signal_async,
    get_current_message,
    sd_bus_open_system,
)

import pulsectl
import pulsectl_asyncio


# Resource policy manager D-BUS parameters
POLICY_DBUS_SERVICE = "org.maemo.resource.manager"
POLICY_DBUS_PATH = "/org/maemo/resource/manager"
POLICY_DBUS_INTERFACE = "org.maemo.resource.manager"

CLIENT_DBUS_INTERFACE = "org.maemo.resource.client"

logging.basicConfig(level=logging.DEBUG)

log = logging.getLogger(__name__)


class Client(DbusInterfaceCommonAsync, interface_name=CLIENT_DBUS_INTERFACE):
    @dbus_method_async("iuuu", method_name="advice")
    async def advice(self, type: int, id: int, reqno: int, resources: int) -> None:
        raise NotImplementedError

    @dbus_method_async("iuuu", method_name="grant")
    async def grant(self, type: int, id: int, reqno: int, resources: int) -> None:
        raise NotImplementedError


@dataclass
class Resource:
    mandatory: int
    optional: int
    dbus: Client
    is_alien_call: bool = False


class DummyPolicyDaemon(DbusInterfaceCommonAsync, interface_name=POLICY_DBUS_INTERFACE):
    def __init__(self, routeManager: RouteManagerShim) -> None:
        super().__init__()

        self.route_manager = routeManager
        self.background_tasks: set[asyncio.Task[None]] = set()
        self.granted = True
        self.clients: dict[str, dict[str, Resource]] = {}

    def path(self, id: int) -> str:
        return f"/org/maemo/resource/client{id}"

    async def advice(self, sender: str, id: int, reqno: int, resources: int) -> None:
        type = 6

        print("")
        print("ADVICE")
        print("type:     ", str(type))
        print("id:       ", str(id))
        print("reqno:    ", str(reqno))
        print("resources:", str(resources))

        path = self.path(id)

        if sender in self.clients:
            if path in self.clients[sender]:
                resource = self.clients[sender][path]
                try:
                    await resource.dbus.advice(type, id, reqno, resources)
                    log.info("Reply")
                except:
                    log.exception("Something went wrong")

    async def grant(self, sender: str, id: int, reqno: int, resources: int) -> None:
        type = 5

        print("")
        print("GRANT")
        print("type:     ", str(type))
        print("id:       ", str(id))
        print("reqno:    ", str(reqno))
        print("resources:", str(resources))

        path = self.path(id)

        if sender in self.clients:
            if path in self.clients[sender]:
                resource = self.clients[sender][path]
                try:
                    await resource.dbus.grant(type, id, reqno, resources)
                except:
                    log.exception("Something went wrong")

    def idle_response(
        self,
        func: Callable[[str, int, int, int], Coroutine[None, None, None]],
        sender: str,
        id: int,
        reqno: int,
        resources: int,
    ) -> None:
        async def wrap() -> None:
            await asyncio.sleep(1)
            await func(sender, id, reqno, resources)

        task = asyncio.create_task(wrap())
        task.add_done_callback(self.background_tasks.discard)
        self.background_tasks.add(task)

    @dbus_method_async("iuuuuuussu", "iuuis", method_name="register")
    async def register(
        self,
        type: int,
        id: int,
        reqno: int,
        mandatory: int,
        optional: int,
        share: int,
        mask: int,
        app_id: str,
        klass: str,
        mode: int,
    ) -> tuple[int, int, int, int, str]:
        try:
            sender = get_current_message().sender

            error_code = 0
            error_string = "OK"

            print("")
            print("REGISTER", sender)
            print("type:     ", str(type))
            print("id:       ", str(id))
            print("reqno:    ", str(reqno))
            print("mandatory:", str(mandatory))
            print("optional: ", str(optional))
            print("share:    ", str(share))
            print("mask:     ", str(mask))
            print("app_id:   ", app_id)
            print("class:    ", klass)
            print("mode:     ", str(mode))

            client = {}

            if sender in self.clients:
                client = self.clients[sender]

            path = self.path(id)

            assert sender is not None

            proxy = Client.new_proxy(sender, path, bus=sd_bus_open_system())

            resource = Resource(mandatory, optional, proxy)

            client[path] = resource
            self.clients[sender] = client

            self.idle_response(self.advice, sender, id, reqno, resource.mandatory)

            return (9, id, reqno, error_code, error_string)

        except:
            log.exception('')

    @dbus_method_async("iuuuuuussu", "iuuis", method_name="update")
    async def update(
        self,
        type: int,
        id: int,
        reqno: int,
        mandatory: int,
        optional: int,
        share: int,
        mask: int,
        app_id: str,
        klass: str,
        mode: int,
    ) -> tuple[int, int, int, int, str]:
        try:
            sender = get_current_message().sender

            error_code = 0
            error_string = "OK"

            print("")
            print("UPDATE", sender)
            print("type:     ", str(type))
            print("id:       ", str(id))
            print("reqno:    ", str(reqno))
            print("mandatory:", str(mandatory))
            print("optional: ", str(optional))
            print("share:    ", str(share))
            print("mask:     ", str(mask))
            print("app_id:   ", app_id)
            print("class:    ", klass)
            print("mode:     ", str(mode))

            return (9, id, reqno, error_code, error_string)

        except:
            log.exception('')

    @dbus_method_async("iuu", "iuuis", method_name="acquire")
    async def acquire(
        self, type: int, id: int, reqno: int
    ) -> tuple[int, int, int, int, str]:
        try:
            sender = get_current_message().sender

            error_code = 0
            error_string = "OK"

            print("")
            print("ACQUIRE", sender)
            print("type:     ", str(type))
            print("id:       ", str(id))
            print("reqno:    ", str(reqno))

            assert sender is not None

            # find the client
            resource = self.clients[sender][self.path(id)]

            if resource.is_alien_call:
                log.info("switching to call mode")
                #            subprocess.run(['callaudiocli', '-m', '1'])
                await self.route_manager.route_output_for_call(True)

            self.idle_response(self.grant, sender, id, reqno, resource.mandatory)

            return (9, id, reqno, error_code, error_string)

        except:
            log.exception('')

    @dbus_method_async("iuu", "iuuis", method_name="release")
    async def release(
        self, type: int, id: int, reqno: int
    ) -> tuple[int, int, int, int, str]:
        try:
            sender = get_current_message().sender

            error_code = 0
            error_string = "OK"

            print("")
            print("RELEASE", sender)
            print("type:     ", str(type))
            print("id:       ", str(id))
            print("reqno:    ", str(reqno))

            assert sender is not None

            self.idle_response(self.grant, sender, id, reqno, 0)

            return (9, id, reqno, error_code, error_string)

        except:
            log.exception('')

    @dbus_method_async("iuu", "iuuis", method_name="unregister")
    async def unregister(
        self, type: int, id: int, reqno: int
    ) -> tuple[int, int, int, int, str]:
        try:
            sender = get_current_message().sender

            errorCode = 0
            errorString = "OK"

            print("")
            print("UNREGISTER", sender)
            print("type:     ", str(type))
            print("id:       ", str(id))
            print("reqno:    ", str(reqno))

            assert sender is not None

            client = self.clients[sender]
            path = self.path(id)
            resource = client[path]

            if resource.is_alien_call:
                log.info("switching back from call mode")
                #            subprocess.run(['callaudiocli', '-m', '0'])

                await self.route_manager.route_output_for_call(False)

            del client[path]

            if len(client) == 0:
                del self.clients[sender]

            return (9, id, reqno, errorCode, errorString)

        except:
            log.exception('')

    @dbus_method_async("iuusssis", "iuuis", method_name="audio")
    async def audio(
        self,
        type: int,
        id: int,
        reqno: int,
        group: str,
        pid: str,
        streamName: str,
        method: int,
        pattern: str,
    ) -> tuple[int, int, int, int, str]:
        try:
            sender = get_current_message().sender

            error_code = 0
            error_string = "OK"

            print("")
            print("AUDIO", sender)
            print("type:       ", str(type))
            print("id:         ", str(id))
            print("reqno:      ", str(reqno))
            print("group:      ", group)
            print("pid:        ", pid)
            print("stream name:", streamName)
            # typedef enum {
            #    resmsg_method_equals = 0,
            #    resmsg_method_startswith,
            #    resmsg_method_matches
            # } resmsg_match_method_t;
            print("method:     ", str(method))
            print("pattern:    ", pattern)

            assert sender is not None

            resource = self.clients[sender][self.path(id)]

            if streamName == "media.name" and method == 0 and pattern == "voice":
                log.info("detected voice call, will switch devices on acquire()")
                resource.is_alien_call = True

            return (9, id, reqno, error_code, error_string)

        except:
            log.exception('')

    @dbus_method_async("iuuu", "iuuis", method_name="video")
    async def video(
        self, type: int, id: int, reqno: int, pid: int
    ) -> tuple[int, int, int, int, str]:
        try:
            sender = get_current_message().sender

            error_code = 0
            error_string = "OK"

            print("")
            print("VIDEO", sender)
            print("type:       ", str(type))
            print("id:         ", str(id))
            print("reqno:      ", str(reqno))
            print("pid:        ", str(pid))

            return (9, id, reqno, error_code, error_string)

        except:
            log.exception('')


class AudioOutputRoute(IntEnum):
    EARPIECE = 1
    PHONE_SPEAKER = 2
    EXTERNAL_SPEAKER = 3
    HEADPHONE = 4


class OhmRouteType(IntFlag):
    OUTPUT = 1 << 0  # sink
    INPUT = 1 << 1  # source
    BUILTIN = 1 << 2
    WIRED = 1 << 3
    WIRELESS = 1 << 4
    VOICE = 1 << 5
    BLUETOOTH_SCO = 1 << 6
    BLUETOOTH_A2DP = 1 << 7
    HEADSET = 1 << 8
    HEADPHONE = 1 << 9
    USB = 1 << 10
    UNKNOWN = 1 << 11
    AVAILABLE = 1 << 25
    PREFERRED = 1 << 26
    ACTIVE = 1 << 27


class RouteManagerShim(
    DbusInterfaceCommonAsync, interface_name="org.nemomobile.Route.Manager"
):
    def __init__(self, pulse: pulsectl_asyncio.PulseAsync) -> None:
        super().__init__()

        self.pulse = pulse

        self.output_route_before_call: Optional[
            tuple[pulsectl.PulseSinkInfo, pulsectl.PulsePortInfo]
        ] = None
        self.output_route_before_speaker: Optional[
            tuple[pulsectl.PulseSinkInfo, pulsectl.PulsePortInfo]
        ] = None
        self.input_route_before_speaker: Optional[
            tuple[pulsectl.PulseSourceInfo, pulsectl.PulsePortInfo]
        ] = None

        self.background_tasks: set[asyncio.Task[None]] = set()

        async def listen() -> None:
            async for event in self.pulse.subscribe_events('all'):
                if event.t == "change":
                    log.info("pulse event happend")
                    await self.mce.sync_proximity_monitoring()

        task = asyncio.create_task(listen())
        task.add_done_callback(self.background_tasks.discard)
        self.background_tasks.add(task)

        # register signal handlers to cancel listener when program is asked to terminate
        # Alternatively, the PulseAudio event subscription can be ended by breaking/returning from the `async for` loop
#        for sig in (signal.SIGTERM, signal.SIGHUP, signal.SIGINT):
#            loop.add_signal_handler(sig, task.cancel)


    async def _sink_list(self) -> list[pulsectl.PulseSinkInfo]:
        sinks = await self.pulse.sink_list()
        assert isinstance(sinks, list)
        return sinks

    async def _source_list(self) -> list[pulsectl.PulseSourceInfo]:
        sources = await self.pulse.source_list()
        assert isinstance(sources, list)
        return sources

    @dbus_method_async("s", method_name="Enable")
    async def enable(self, route: str) -> None:
        try:
            log.info("route enable: %s", route)

            if route == "speaker":
                await self.use_speaker_in_call(True)

        except:
            log.exception('')

    @dbus_method_async("s", method_name="Disable")
    async def disable(self, route: str) -> None:
        try:
            log.info("route disable: %s", route)

            if route == "speaker":
                await self.use_speaker_in_call(False)

        except:
            log.exception('')

    @dbus_method_async(result_signature="susu", method_name="ActiveRoutes")
    async def active_routes(self) -> tuple[str, int, str, int]:
        try:
            OHM_EXT_ROUTE_TYPE_OUTPUT = 1 << 0  # sink
            OHM_EXT_ROUTE_TYPE_INPUT = 1 << 1  # source
            OHM_EXT_ROUTE_TYPE_BUILTIN = 1 << 2
            OHM_EXT_ROUTE_TYPE_WIRED = 1 << 3
            OHM_EXT_ROUTE_TYPE_WIRELESS = 1 << 4
            OHM_EXT_ROUTE_TYPE_VOICE = 1 << 5
            OHM_EXT_ROUTE_TYPE_BLUETOOTH_SCO = 1 << 6
            OHM_EXT_ROUTE_TYPE_BLUETOOTH_A2DP = 1 << 7
            OHM_EXT_ROUTE_TYPE_HEADSET = 1 << 8
            OHM_EXT_ROUTE_TYPE_HEADPHONE = 1 << 9
            OHM_EXT_ROUTE_TYPE_USB = 1 << 10
            OHM_EXT_ROUTE_TYPE_UNKNOWN = 1 << 11

            OHM_EXT_ROUTE_TYPE_AVAILABLE = 1 << 25
            OHM_EXT_ROUTE_TYPE_PREFERRED = 1 << 26
            OHM_EXT_ROUTE_TYPE_ACTIVE = 1 << 27

            routes = (
                "speaker",
                OHM_EXT_ROUTE_TYPE_OUTPUT
                | OHM_EXT_ROUTE_TYPE_BUILTIN
                | OHM_EXT_ROUTE_TYPE_AVAILABLE
                | OHM_EXT_ROUTE_TYPE_ACTIVE,
                "microphone",
                OHM_EXT_ROUTE_TYPE_INPUT | OHM_EXT_ROUTE_TYPE_BUILTIN,
            )

            log.info("ActiveRoutes: %s", routes)

            return routes

        except:
            log.exception('')

    @dbus_signal_async("su", signal_name="AudioRouteChanged")
    async def audio_route_changed(self) -> None:
        raise NotImplementedError

    def audio_route_changed_emit(self, name: str, type: OhmRouteType) -> None:
        # fairly certain the checker is wrong here
        self.audio_route_changed.emit((name, int(type)))  # type: ignore

    @dbus_signal_async("suu", signal_name="AudioFeatureChanged")
    async def audio_feature_changed(self) -> None:
        raise NotImplementedError

    def audio_feature_changed_emit(self, name: str, a: bool, b: bool) -> None:
        # fairly certain the checker is wrong here
        self.audio_feature_changed.emit((name, int(a), int(b)))  # type: ignore

    async def get_default_sink(self) -> pulsectl.PulseSinkInfo:
        sink_name = (await self.pulse.server_info()).default_sink_name
        for sink in await self._sink_list():
            if sink.name == sink_name:
                return sink

        raise Exception("No default sink found")

    async def get_default_source(self) -> pulsectl.PulseSourceInfo:
        source_name = (await self.pulse.server_info()).default_source_name
        for source in await self._source_list():
            if source.name == source_name:
                return source

        raise Exception("No default source found")

    async def find_private_output_for_call(
        self,
    ) -> tuple[pulsectl.PulseSinkInfo, pulsectl.PulsePortInfo]:
        # first try active sink
        sink = await self.get_default_sink()
        for port in sink.port_list:
            if port.available != "no" and port.type == "headphones":
                return (sink, port)

            if port.available != "no" and port.type == "earpiece":
                return (sink, port)

        # now try all sinks (these don't have priorities)
        for sink in await self._sink_list():
            # ports should be ordered by priority already
            for port in sink.port_list:
                if port.available != "no" and port.type == "headphones":
                    return (sink, port)

                if port.available != "no" and port.type == "earpiece":
                    return (sink, port)

        log.info("no earpiece found")
        raise Exception("No earpiece found")

    async def find_mic_for_output(
        self,
        sink: pulsectl.PulseSinkInfo,
        port: pulsectl.PulsePortInfo,
    ) -> tuple[pulsectl.PulseSourceInfo, pulsectl.PulsePortInfo]:
        # look for mic on same device as the output
        sink_device = sink.name.split('.')[1]
        log.info("looking at output dev: %s", sink_device)

        searching_for_type = "mic"

        for source in await self._source_list():
            device = source.name.split('.')[1]

            if device == sink_device:
                for source_port in source.port_list:
                    if source_port.available != "no" and source_port.type == searching_for_type:
                        return (source, source_port)

        return (None, None)

    async def find_best_output(
        self,
    ) -> tuple[pulsectl.PulseSinkInfo, pulsectl.PulsePortInfo]:
        # first try the highest priority port on the current sink
        sink = await self.get_default_sink()
        for port in sink.port_list:
            if port.available != "no":
                return (sink, port)

        # now try other sinks (these don't have priorities)
        for sink in await self._sink_list():
            for port in sink.port_list:
                if port.available != "no":
                    return (sink, port)

        raise Exception("No best output found")

    async def find_speaker_for_call(
        self,
    ) -> tuple[pulsectl.PulseSinkInfo, pulsectl.PulsePortInfo]:
        # if route before call happened to be speaker, use that
        if self.output_route_before_call is not None:
            (sink, port) = self.output_route_before_call
            if await self.route_still_exists(sink, port):
                if port.available != "no" and port.type == "speaker":
                    return (sink, port)

        # otherwise look for a speaker on current sink
        sink = await self.get_default_sink()
        for port in sink.port_list:
            if port.available != "no" and port.type == "speaker":
                return (sink, port)

        # otherwise try all sinks (these don't have priorities)
        for sink in await self._sink_list():
            # ports should be ordered by priority already
            for port in sink.port_list:
                if port.available != "no" and port.type == "speaker":
                    return (sink, port)

        log.info("no speaker found")
        raise Exception("No speaker found")

    async def route_still_exists(
        self,
        existing_sink: pulsectl.PulseSinkInfo,
        existing_port: pulsectl.PulsePortInfo,
    ) -> bool:
        log.info("checking if route still exists")
        for sink in await self._sink_list():
            if sink.name != existing_sink.name:
                continue

            for port in sink.port_list:
                if port.name == existing_port.name and port.available != "no":
                    print("yupppp")
                    return True

        log.info("that's a no")
        return False

    async def source_route_still_exists(
        self,
        existing_source: pulsectl.PulseSourceInfo,
        existing_port: pulsectl.PulsePortInfo,
    ) -> bool:
        log.info("checking if route still exists")
        for source in await self._source_list():
            if source.name != existing_source.name:
                continue

            for port in source.port_list:
                if port.name == existing_port.name and port.available != "no":
                    print("yupppp")
                    return True

        log.info("that's a no")
        return False

    async def switch_output(
        self,
        sink: pulsectl.PulseSinkInfo,
        port: pulsectl.PulsePortInfo,
    ) -> None:
        default_sink = await self.get_default_sink()

        if sink.name != default_sink.name:
            print("need to switch sink")
            await self.pulse.default_set(sink)
            for sink_input in await self.pulse.sink_input_list():
                print("moving stream over to new sink")
                await self.pulse.sink_input_move(sink_input.index, sink.index)

        if len(sink.port_list) > 1:
            await self.pulse.port_set(sink, port)

    async def switch_input(
        self,
        source: pulsectl.PulseSourceInfo,
        port: pulsectl.PulsePortInfo,
    ) -> None:
        default_source = await self.get_default_source()

        if source.name != default_source.name:
            print("need to switch source")
            await self.pulse.default_set(source)
            for source_output in await self.pulse.source_output_list():
                print("moving stream over to new source")
                await self.pulse.source_output_move(source_output.index, source.index)

        if len(source.port_list) > 1:
            await self.pulse.port_set(source, port)

    async def route_output_for_call(self, in_call: bool) -> None:
        log.info("routeOutputForCall %r", in_call)
        default_sink = await self.get_default_sink()

        log.info("cur route %s", default_sink.port_active.name)

        if in_call:
            # if any speaker (could also be bt) is in use, switch to a more
            # private device (earpiece or headphones), otherwise just leave
            # things as-is
            if default_sink.port_active.type == "speaker":
                self.output_route_before_call = (default_sink, default_sink.port_active)
                (sink, port) = await self.find_private_output_for_call()
                await self.switch_output(sink, port)

            self.audio_route_changed_emit(
                "speaker",
                OhmRouteType.OUTPUT | OhmRouteType.BUILTIN | OhmRouteType.AVAILABLE,
            )
            self.audio_route_changed_emit(
                "earpiece",
                OhmRouteType.OUTPUT
                | OhmRouteType.BUILTIN
                | OhmRouteType.VOICE
                | OhmRouteType.AVAILABLE
                | OhmRouteType.ACTIVE,
            )

            self.audio_feature_changed_emit("speaker", True, False)
        else:
            if default_sink.port_active.type == "earpiece":
                if self.output_route_before_call is not None:
                    log.info("Restoring output route from before call")
                    (sink, prev_port) = self.output_route_before_call
                    if not await self.route_still_exists(sink, prev_port):
                        log.info("Route no longer exists, finding new one")
                        (sink, prev_port) = await self.find_best_output()
                else:
                    (sink, prev_port) = await self.find_best_output()

                await self.switch_output(sink, prev_port)

            if self.input_route_before_speaker is not None:
                log.info("Restoring input route from before call")
                (source, prev_port) = self.input_route_before_speaker
                if await self.source_route_still_exists(source, prev_port):
                    await self.switch_input(source, prev_port)
                else:
                    log.info("Route no longer exists, not restoring")

            self.output_route_before_call = None
            self.output_route_before_speaker = None
            self.input_route_before_speaker = None

        log.info("new route: %s", (await self.get_default_sink()).port_active.name)

        await self.mce.sync_proximity_monitoring()

    async def use_speaker_in_call(self, use_speaker: bool) -> None:
        log.info("useSpeakerInCall %r", use_speaker)
        default_sink = await self.get_default_sink()

        log.info("cur route %s", default_sink.port_active.name)

        if use_speaker:
            if default_sink.port_active.type != "speaker":
                self.output_route_before_speaker = (default_sink, default_sink.port_active)
                (sink, port) = await self.find_speaker_for_call()
                await self.switch_output(sink, port)

                default_source = await self.get_default_source()
                if default_source.port_active.type != "mic":
                    (source, port) = await self.find_mic_for_output(sink, port)
                    if source != None:
                        log.info("Found an internal mic, switching")
                        self.input_route_before_speaker = (default_source, default_source.port_active)
                        await self.switch_input(source, port)

                self.audio_route_changed_emit(
                    "earpiece",
                    OhmRouteType.OUTPUT
                    | OhmRouteType.BUILTIN
                    | OhmRouteType.VOICE
                    | OhmRouteType.AVAILABLE,
                )
                self.audio_route_changed_emit(
                    "speaker",
                    OhmRouteType.OUTPUT
                    | OhmRouteType.BUILTIN
                    | OhmRouteType.AVAILABLE
                    | OhmRouteType.ACTIVE,
                )
                self.audio_feature_changed_emit("speaker", True, True)
            else:
                # we're not listening to sink/port changes from pulse,
                # so if output changed to a speaker underneath our feet, just change the route silently
                self.audio_route_changed_emit(
                    "earpiece",
                    OhmRouteType.OUTPUT
                    | OhmRouteType.BUILTIN
                    | OhmRouteType.VOICE
                    | OhmRouteType.AVAILABLE,
                )
                self.audio_route_changed_emit(
                    "speaker",
                    OhmRouteType.OUTPUT
                    | OhmRouteType.BUILTIN
                    | OhmRouteType.AVAILABLE
                    | OhmRouteType.ACTIVE,
                )
                self.audio_feature_changed_emit("speaker", True, True)
        else:
            if default_sink.port_active.type == "speaker":
                if self.output_route_before_speaker is not None:
                    log.info("Restoring output route from before speaker")
                    (sink, prev_port) = self.output_route_before_speaker
                    if not await self.route_still_exists(sink, prev_port):
                        log.info("Route no longer exists, finding new one")
                        (sink, prev_port) = await self.find_private_output_for_call()
                else:
                    (sink, prev_port) = await self.find_private_output_for_call()

                await self.switch_output(sink, prev_port)

                if self.input_route_before_speaker is not None:
                    log.info("Restoring input route from before speaker")
                    (source, prev_port) = self.input_route_before_speaker
                    if await self.source_route_still_exists(source, prev_port):
                        await self.switch_input(source, prev_port)
                    else:
                        log.info("Route no longer exists, not restoring")

                self.audio_route_changed_emit(
                    "speaker",
                    OhmRouteType.OUTPUT | OhmRouteType.BUILTIN | OhmRouteType.AVAILABLE,
                )
                self.audio_route_changed_emit(
                    "earpiece",
                    OhmRouteType.OUTPUT
                    | OhmRouteType.BUILTIN
                    | OhmRouteType.VOICE
                    | OhmRouteType.AVAILABLE
                    | OhmRouteType.ACTIVE,
                )
                self.audio_feature_changed_emit("speaker", True, False)
            else:
                # we're not listening to sink/port changes from pulse,
                # so if output changed to a speaker underneath our feet, just change the route silently
                self.audio_route_changed_emit(
                    "speaker",
                    OhmRouteType.OUTPUT | OhmRouteType.BUILTIN | OhmRouteType.AVAILABLE,
                )
                self.audio_route_changed_emit(
                    "earpiece",
                    OhmRouteType.OUTPUT
                    | OhmRouteType.BUILTIN
                    | OhmRouteType.VOICE
                    | OhmRouteType.AVAILABLE
                    | OhmRouteType.ACTIVE,
                )
                self.audio_feature_changed_emit("speaker", True, False)

            self.output_route_before_speaker = None
            self.input_route_before_speaker = None

        log.info("new route: %s", (await self.get_default_sink()).port_active.name)

        await self.mce.sync_proximity_monitoring()

    async def is_routing_to_earpiece(self) -> bool:
        default_sink = await self.get_default_sink()
        return default_sink.port_active.type == "earpiece"

    def set_mce(self, mce: MceShim) -> bool:
        self.mce = mce


class SensorDaemon(DbusInterfaceCommonAsync, interface_name="org.gnome.Shell.SensorDaemon"):
    @dbus_method_async(method_name="StartProximityMonitoring")
    async def start_proximity_monitoring() -> None:
        raise NotImplementedError

    @dbus_method_async(method_name="StopProximityMonitoring")
    async def stop_proximity_monitoring() -> None:
        raise NotImplementedError


class MceShim(DbusInterfaceCommonAsync, interface_name="com.nokia.mce.request"):
    def __init__(self, route_manager: RouteManagerShim) -> None:
        super().__init__()

        self.route_manager = route_manager

        self.sensor_daemon = SensorDaemon.new_proxy(
            "org.gnome.Shell.SensorDaemon", "/org/gnome/Shell/SensorDaemon"
        )

        self.in_call = False
        self.monitoring_proximity = False

    async def sync_proximity_monitoring(self) -> None:
        should_monitor = self.in_call and await self.route_manager.is_routing_to_earpiece()
        log.info("sync moni %d", should_monitor)
        if should_monitor != self.monitoring_proximity:
            self.monitoring_proximity = should_monitor

            if should_monitor:
                await self.sensor_daemon.start_proximity_monitoring()
            else:
                await self.sensor_daemon.stop_proximity_monitoring()

    @dbus_method_async("ss", "b", method_name="req_call_state_change")
    async def req_call_state_change(self, state: str, desc: str) -> bool:
        try:
            log.info("req_call_state_change: %s", state)

            if state == "active":
                self.in_call = True
            elif state == "none":
                self.in_call = False

            await self.sync_proximity_monitoring()

            return True

        except:
            log.exception('')


class Feedbackd(DbusInterfaceCommonAsync, interface_name="org.sigxcpu.Feedback"):
    @dbus_method_async("ssa{sv}i", "u", method_name="TriggerFeedback")
    async def trigger_feedback(
        self, app_id: str, event: str, hints: dict[str, Any], timeout: int
    ) -> int:
        raise NotImplementedError

    @dbus_method_async("u", method_name="EndFeedback")
    async def end_feedback(self, id: int) -> None:
        raise NotImplementedError


class NonGraphicFeedbackShim(
    DbusInterfaceCommonAsync, interface_name="com.nokia.NonGraphicFeedback1"
):
    def __init__(self) -> None:
        super().__init__()

        self.feedbackd = Feedbackd.new_proxy(
            "org.sigxcpu.Feedback", "/org/sigxcpu/Feedback"
        )
        self.current_id = -1

    def translate_feedback(
        self, name: str, params: dict[str, Any]
    ) -> tuple[str, dict[str, Any]]:
        hints = {}
        feedbackName = None


        if name == "feedback_press":
            # generic touch feedback
            feedbackName = "button-pressed"
        elif name == "pulldown_highlight":
            # another generic touch feedback
            feedbackName = "button-pressed"
        elif name == "vibra":
            # only vibration feedback, no sound
            feedbackName = "bell-terminal"
            hints["profile"] = ("s", "quiet")

        if 'media.vibra' in params:
            should_vibrate = params['media.vibra'][1]
            if not should_vibrate:
                # unfortunately we can only disable vibration by also disabling sound
                hints["profile"] = ("s", "silent")

        if 'haptic.sequence' in params:
            haptic_sequence = params['haptic.sequence'][1]
            # in format "on=timeMs"

        if 'haptic.type' in params:
            haptic_type = params['haptic.type'][1]
            # can be "touch"

        return (feedbackName, hints)

    @dbus_method_async("sa{sv}", "u", method_name="Play")
    async def play(self, name: str, params: dict[str, Any]) -> int:
        try:
            log.info("NonGraphicFeedbackShim Play: %s", name)
            print(params)

            (feedback_name, hints) = self.translate_feedback(name, params)

            if feedback_name == None:
                return 0

            print("NonGraphicFeedbackShim actually playing:", feedback_name, hints)
            self.current_id = await self.feedbackd.trigger_feedback(
                "aliendalvik", feedback_name, hints, -1
            )

            return self.current_id

        except:
            log.exception('')

    @dbus_method_async("u", "u", method_name="Stop")
    async def stop(self, feedback_id: int) -> int:
        try:
            log.info("NonGraphicFeedbackShim Stop: %i", feedback_id)
            await self.feedbackd.end_feedback(feedback_id)
            self.current_id = -1
            return 0

        except:
            log.exception('')


async def main() -> None:
    system = sd_bus_open_system()
    await asyncio.gather(
        *(
            system.request_name_async(name, 0)
            for name in [
                POLICY_DBUS_SERVICE,
                "com.nokia.NonGraphicFeedback1.Backend",
                "org.nemomobile.Route.Manager",
                "com.nokia.mce",
            ]
        )
    )

    async with pulsectl_asyncio.PulseAsync("my-client-name") as pulse:
        log.info("sinks: %r", await pulse.sink_list())

        route_manager = RouteManagerShim(pulse)
        route_manager.export_to_dbus("/org/nemomobile/Route/Manager", system)
        call_daemon = DummyPolicyDaemon(route_manager)
        call_daemon.export_to_dbus(POLICY_DBUS_PATH, system)
        mce = MceShim(route_manager)
        route_manager.set_mce(mce)
        mce.export_to_dbus("/com/nokia/mce/request", system)

        ngf = NonGraphicFeedbackShim()
        ngf.export_to_dbus("/com/nokia/NonGraphicFeedback1", system)

        while True:
            await asyncio.sleep(10)


if __name__ == "__main__":
    asyncio.run(main())