fp6-img/aports/device/callaudioshim/callaudioshim.py

#!/usr/bin/env python3
"""callaudioshim.py — drop-in replacement for callaudiod on the FP6.

Stock callaudiod 0.1.99 is broken on this hardware: it never finds a
"voice-capable" PulseAudio card (our UCM exposes HiFi with Speaker+Mic, no
Earpiece port), so
  - its MicState property stays at the D-Bus default 0 = MIC_OFF and
    plasma-dialer's in-call page permanently shows the mic as MUTED;
  - MuteMic() bails with "card has no usable source" — the mute button does
    nothing;
  - SelectMode() never replies, blocking kde-telephony-daemon's main loop for
    the full 25 s GDBus timeout on every call (journal/ims.md s41).

A second, independent upstream defect (plasma-dialer 6.7.3) makes the UI state
wrong even with a working callaudiod: the mute button binds to
kde-telephony-daemon's DialerUtils `m_mute` — an UNINITIALIZED bool that
nothing ever syncs with reality (it only changes on button presses, and the
daemon never re-reads MicState). So every call page opened showing "muted"
while the mic was live, and the first press was wasted re-syncing. This shim
papers over that too: whenever the REAL mic state changes (and whenever
kde-telephony-daemon [re]appears on the bus), it pushes the truth into
DialerUtils.setMute(b), which updates m_mute and fires muteChanged to the UI.
The loop is self-terminating (the daemon only re-emits on value change, and
the bounce-back MuteMic is a no-op when the state already matches).

This shim owns org.mobian_project.CallAudio instead and implements the same
interface (verified against callaudiod's org.mobian_project.CallAudio.xml):
  - MuteMic(b) mutes/unmutes @DEFAULT_AUDIO_SOURCE@ via wpctl. The default
    source is ec_source (the AEC'd mic, s43), which every capture stream —
    including the call's pw-record — consumes, so mute = real uplink silence
    (verified: tone capture RMS 94.5 -> 0.0 on mute).
  - MicState mirrors the ACTUAL source mute (polled every 2 s too, so
    out-of-band wpctl/pactl changes stay reflected), with PropertiesChanged
    so kde-telephony-daemon's cached GDBusProxy tracks it.
  - SelectMode / EnableSpeaker reply immediately (bookkeeping only: the
    loudspeaker is the only output today, and UCM verb switching is not
    dynamic in this setup — journal/audio.md 2026-07-18). SpeakerState
    defaults to ON. When an Earpiece UCM device exists these become real.

Stock callaudiod must not own the name: pkill it and mask its D-Bus
activation (see journal/ims.md s44 deploy notes). Runs as the session user
(autostart .desktop), no root needed.
"""
import subprocess

import dbus
import dbus.service
import dbus.mainloop.glib
from gi.repository import GLib

BUS_NAME = "org.mobian_project.CallAudio"
OBJ_PATH = "/org/mobian_project/CallAudio"
IFACE = "org.mobian_project.CallAudio"
PROPS_IFACE = "org.freedesktop.DBus.Properties"
SOURCE = "@DEFAULT_AUDIO_SOURCE@"

# kde-telephony-daemon's DialerUtils — the mute-button state lives there
DIALER_BUS = "org.kde.telephony.DialerUtils"
DIALER_PATH = "/org/kde/telephony/DialerUtils/tel/mm"
DIALER_IFACE = "org.kde.telephony.DialerUtils"

MODE_DEFAULT, MODE_CALL = 0, 1
OFF, ON, UNKNOWN = 0, 1, 255


def log(msg):
    print(msg, flush=True)


def source_muted():
    """True/False from wpctl ('Volume: 1.00 [MUTED]'), None if unreadable."""
    try:
        out = subprocess.run(["wpctl", "get-volume", SOURCE],
                             capture_output=True, text=True, timeout=5).stdout
    except (OSError, subprocess.TimeoutExpired):
        return None
    if "Volume:" not in out:
        return None
    return "[MUTED]" in out


def set_source_muted(mute):
    return subprocess.run(
        ["wpctl", "set-mute", SOURCE, "1" if mute else "0"],
        capture_output=True, timeout=5).returncode == 0


class CallAudio(dbus.service.Object):
    def __init__(self, bus):
        super().__init__(bus, OBJ_PATH)
        self.bus = bus
        self.audio_mode = MODE_DEFAULT
        self.speaker_state = ON      # loudspeaker is the only output today
        m = source_muted()
        self.mic_state = UNKNOWN if m is None else (OFF if m else ON)
        log(f"callaudioshim: up, MicState={self.mic_state}")
        GLib.timeout_add_seconds(2, self._poll)
        self._push_dialer()
        # re-push whenever kde-telephony-daemon (re)appears: its m_mute is an
        # uninitialized bool until someone calls setMute
        bus.watch_name_owner(DIALER_BUS, self._dialer_owner_changed)

    # -------------------------------------------- DialerUtils state push
    def _dialer_owner_changed(self, owner):
        if owner:
            log(f"callaudioshim: DialerUtils owner {owner}; pushing mute state")
            self._push_dialer()

    def _push_dialer(self):
        if self.mic_state == UNKNOWN:
            return
        try:
            self.bus.call_async(DIALER_BUS, DIALER_PATH, DIALER_IFACE,
                                "setMute", "b", (self.mic_state == OFF,),
                                lambda *a: None, lambda *a: None)
        except dbus.DBusException:
            pass

    # ------------------------------------------------ property plumbing
    def _props(self):
        return {
            "AudioMode": dbus.UInt32(self.audio_mode),
            "SpeakerState": dbus.UInt32(self.speaker_state),
            "MicState": dbus.UInt32(self.mic_state),
        }

    def _set(self, attr, prop, value):
        if getattr(self, attr) == value:
            return
        setattr(self, attr, value)
        self.PropertiesChanged(IFACE, {prop: dbus.UInt32(value)}, [])
        if prop == "MicState":
            self._push_dialer()

    def _poll(self):
        # catch out-of-band mute changes (wpctl, pactl, other clients)
        m = source_muted()
        if m is not None:
            self._set("mic_state", "MicState", OFF if m else ON)
        return True

    @dbus.service.method(PROPS_IFACE, in_signature="ss", out_signature="v")
    def Get(self, interface, prop):
        return self._props()[prop]

    @dbus.service.method(PROPS_IFACE, in_signature="s", out_signature="a{sv}")
    def GetAll(self, interface):
        return self._props()

    @dbus.service.signal(PROPS_IFACE, signature="sa{sv}as")
    def PropertiesChanged(self, interface, changed, invalidated):
        pass

    # ------------------------------------------------ org.mobian_project.CallAudio
    @dbus.service.method(IFACE, in_signature="u", out_signature="b")
    def SelectMode(self, mode):
        if mode not in (MODE_DEFAULT, MODE_CALL):
            raise dbus.exceptions.DBusException(
                "invalid mode", name="org.freedesktop.DBus.Error.InvalidArgs")
        log(f"callaudioshim: SelectMode({int(mode)})")
        self._set("audio_mode", "AudioMode", int(mode))
        if mode == MODE_DEFAULT:
            # leaving a call: never leave the mic muted behind
            if source_muted():
                set_source_muted(False)
                self._set("mic_state", "MicState", ON)
        return True

    @dbus.service.method(IFACE, in_signature="b", out_signature="b")
    def EnableSpeaker(self, enable):
        log(f"callaudioshim: EnableSpeaker({bool(enable)})")
        self._set("speaker_state", "SpeakerState", ON if enable else OFF)
        return True

    @dbus.service.method(IFACE, in_signature="b", out_signature="b")
    def MuteMic(self, mute):
        ok = set_source_muted(bool(mute))
        log(f"callaudioshim: MuteMic({bool(mute)}) -> {ok}")
        if not ok:
            return False
        self._set("mic_state", "MicState", OFF if mute else ON)
        return True


def main():
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    session = dbus.SessionBus()
    name = dbus.service.BusName(BUS_NAME, session, do_not_queue=True)  # noqa: F841
    CallAudio(session)
    log(f"callaudioshim: owning {BUS_NAME}")
    GLib.MainLoop().run()


if __name__ == "__main__":
    main()