SmartestHome/mic-follow/audio_sources.py

#!/usr/bin/env python3
"""Finding, selecting and moving audio inputs on a PipeWire desktop.

Everything here goes through `pactl` (pipewire-pulse's compatibility layer) rather
than a mix of `wpctl`, `pw-dump` and `pw-metadata`. One tool covers all four things
this needs — list sources, list capture streams, set the default, move a live stream —
and it is the only one of them with a documented JSON output mode, which is the
difference between parsing a stable structure and scraping a table that changes
between releases.

WHY MOVING STREAMS MATTERS, and is not the same as setting the default: changing the
default source only affects applications that asked for "default". Discord, once you
have picked a specific microphone in its settings, holds that device — and the whole
point of this component is that somebody who has set up a studio mic has certainly
picked it explicitly. So the switch does both: it sets the default (for anything that
follows it) and moves the already-running capture streams of the configured
applications (for anything that does not).

Nothing in this file has run against a real PipeWire — see mic-follow/README.md. The
parsing and selection logic is tested against captured pactl output in
test_selection.py; what is unverified is the exact shape of that output on the user's
own version, and every command that changes state.
"""

from __future__ import annotations

import json
import logging
import subprocess
from dataclasses import dataclass

log = logging.getLogger(__name__)


@dataclass(frozen=True)
class Source:
    index: int
    name: str
    description: str

    def matches(self, pattern: str) -> bool:
        pattern = pattern.strip().lower()
        return (pattern == self.name.lower()
                or pattern in self.name.lower()
                or pattern in self.description.lower())


@dataclass(frozen=True)
class CaptureStream:
    index: int
    source: int
    application: str


def _pactl_json(*args: str) -> list | dict | None:
    try:
        result = subprocess.run(["pactl", "-f", "json", *args],
                                capture_output=True, text=True, timeout=10, check=False)
    except (OSError, subprocess.SubprocessError) as exc:
        log.warning("pactl %s: %s", " ".join(args), exc)
        return None
    if result.returncode != 0:
        log.warning("pactl %s failed: %s", " ".join(args), (result.stderr or "").strip()[:200])
        return None
    try:
        return json.loads(result.stdout or "null")
    except json.JSONDecodeError as exc:
        log.warning("pactl %s returned unparseable JSON: %s", " ".join(args), exc)
        return None


def parse_sources(payload) -> list[Source]:
    """pactl's `list sources` JSON -> Source objects, monitors dropped.

    Monitor sources (the loopback of an output) are excluded deliberately: they match
    name patterns surprisingly often, and selecting one means transmitting whatever the
    desktop is playing instead of what the person is saying — the single worst outcome
    this component could produce.
    """
    sources: list[Source] = []
    for entry in payload or []:
        if not isinstance(entry, dict):
            continue
        name = str(entry.get("name") or "")
        if not name or name.endswith(".monitor"):
            continue
        properties = entry.get("properties") or {}
        if str(properties.get("device.class", "")).lower() == "monitor":
            continue
        sources.append(Source(
            index=int(entry.get("index", -1)),
            name=name,
            description=str(entry.get("description") or properties.get("device.description") or name),
        ))
    return sources


def parse_capture_streams(payload) -> list[CaptureStream]:
    """pactl's `list source-outputs` JSON -> the live recording streams."""
    streams: list[CaptureStream] = []
    for entry in payload or []:
        if not isinstance(entry, dict):
            continue
        properties = entry.get("properties") or {}
        application = str(
            properties.get("application.name")
            or properties.get("application.process.binary")
            or ""
        )
        source = entry.get("source")
        streams.append(CaptureStream(
            index=int(entry.get("index", -1)),
            source=int(source) if isinstance(source, int) else -1,
            application=application,
        ))
    return streams


def select_source(sources: list[Source], pattern: str) -> Source | None:
    """The configured pattern -> one source, preferring the least surprising match.

    Exact `node.name` first, then a substring of the name, then a substring of the
    human description. Ties inside a tier are resolved by lowest index (the order
    pactl reports, which is stable within a boot) and logged, because a pattern that
    matches two microphones is a configuration mistake the user should hear about
    rather than a coin flip that lands differently after a reboot.
    """
    pattern = (pattern or "").strip().lower()
    if not pattern:
        return None
    exact = [s for s in sources if s.name.lower() == pattern]
    by_name = [s for s in sources if pattern in s.name.lower()]
    by_description = [s for s in sources if pattern in s.description.lower()]
    for tier, label in ((exact, "exact name"), (by_name, "name"), (by_description, "description")):
        if not tier:
            continue
        chosen = sorted(tier, key=lambda s: s.index)[0]
        if len(tier) > 1:
            log.warning("%r matches %d sources by %s (%s) — using %r",
                        pattern, len(tier), label,
                        ", ".join(s.name for s in tier), chosen.name)
        return chosen
    return None


def streams_to_move(streams: list[CaptureStream], applications: list[str],
                    target: Source) -> list[CaptureStream]:
    """Which live capture streams belong to the configured apps and are on the wrong
    source already. Streams already on the target are left alone: moving a stream that
    is where it should be is a needless glitch in somebody's live audio."""
    wanted = [a.strip().lower() for a in applications if a.strip()]
    if not wanted:
        return []
    return [
        stream for stream in streams
        if stream.source != target.index
        and any(pattern in stream.application.lower() for pattern in wanted)
    ]


# --- The four live commands. Everything above is pure and tested; these are not. -----

def list_sources() -> list[Source]:
    return parse_sources(_pactl_json("list", "sources"))


def list_capture_streams() -> list[CaptureStream]:
    return parse_capture_streams(_pactl_json("list", "source-outputs"))


def _run(*args: str) -> bool:
    try:
        result = subprocess.run(["pactl", *args], capture_output=True, text=True,
                                timeout=10, check=False)
    except (OSError, subprocess.SubprocessError) as exc:
        log.warning("pactl %s: %s", " ".join(args), exc)
        return False
    if result.returncode != 0:
        log.warning("pactl %s failed: %s", " ".join(args), (result.stderr or "").strip()[:200])
        return False
    return True


def set_default_source(source: Source) -> bool:
    return _run("set-default-source", source.name)


def move_stream(stream: CaptureStream, target: Source) -> bool:
    return _run("move-source-output", str(stream.index), target.name)


def current_default_source_name() -> str:
    payload = _pactl_json("info")
    if isinstance(payload, dict):
        return str(payload.get("default_source_name") or "")
    return ""


def main() -> int:
    """`audio_sources.py` on its own prints what this desktop has, which is how you
    fill in the source patterns in CoreSystemConfig.json."""
    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
    sources = list_sources()
    if not sources:
        print("No sources found — is pactl installed and a PipeWire session running?")
        return 1
    default = current_default_source_name()
    print("Audio inputs on this machine (any part of a name or description works as a")
    print("`source` pattern in CoreSystemConfig.json):\n")
    for source in sources:
        marker = "*" if source.name == default else " "
        print(f" {marker} {source.description}")
        print(f"     {source.name}")
    print("\n* = current default")
    streams = list_capture_streams()
    if streams:
        print("\nApplications recording right now:")
        for stream in streams:
            print(f"   {stream.application or '(unnamed)'} (stream {stream.index}, source {stream.source})")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())