google_meet

1"""Virtual audio bridge for feeding generated speech into Chrome's mic.2 3Linux: pactl creates a null-sink plus a virtual source on the sink's monitor; callers set4``PULSE_SOURCE=<source_name>`` in Chrome's env. macOS: only verifies BlackHole 2ch is installed5(the default-input switch is left to the user). Windows: unsupported.6"""7 8from __future__ import annotations9 10import contextlib11import platform12import subprocess13from typing import Optional14 15 16_BLACKHOLE_DEVICE = "BlackHole 2ch"17 18 19def _pactl(*args: str, check: bool) -> subprocess.CompletedProcess:20    return subprocess.run(["pactl", *args], check=check, capture_output=True, text=True, encoding='utf-8',21                          errors='replace', stdin=subprocess.DEVNULL)22 23 24class AudioBridge:25    """Virtual audio device for Chrome fake-mic input: ``setup()`` before launch, ``teardown()`` after."""26 27    def __init__(self, name_prefix: str = "hermes_meet") -> None:28        self._name_prefix = name_prefix29        self._platform: Optional[str] = None30        self._device_name: Optional[str] = None31        self._write_target: Optional[str] = None32        self._module_ids: list[int] = []33 34    def _ready(self, value: Optional[str]) -> str:35        if not value:36            raise RuntimeError("AudioBridge not set up yet")37        return value38 39    device_name = property(lambda self: self._ready(self._device_name))40    write_target = property(lambda self: self._ready(self._write_target))41 42    def setup(self) -> dict:43        """Provision the device; raises RuntimeError on unsupported platforms or missing tools."""44        system = platform.system()45        impl = {"Linux": self._setup_linux, "Darwin": self._setup_darwin}.get(system)46        if impl is None:47            raise RuntimeError("windows not supported in v2" if system == "Windows"48                               else f"unsupported platform: {system}")49        return impl()50 51    def teardown(self) -> None:52        """Release the virtual audio device. Idempotent; never raises."""53        for mod_id in reversed(self._module_ids):  # linux only; virtual-source before null-sink54            with contextlib.suppress(Exception):55                _pactl("unload-module", str(mod_id), check=False)56        self._module_ids = []57 58    def _finish(self, tag: str, device: str, write_target: str, module_ids: list[int]) -> dict:59        self._platform = tag60        self._device_name = device61        self._write_target = write_target62        self._module_ids = module_ids63        return {"platform": tag, "device_name": device, "sample_rate": 48000, "channels": 2,64                "module_ids": list(module_ids), "write_target": write_target}65 66    def _setup_linux(self) -> dict:67        sink_name = f"{self._name_prefix}_sink"68        src_name = f"{self._name_prefix}_src"69        try:70            sink_out = _pactl(71                "load-module", "module-null-sink", f"sink_name={sink_name}",72                "sink_properties=device.description=HermesMeetSink", check=True)73        except FileNotFoundError as exc:74            raise RuntimeError("pactl not found — install PulseAudio/pipewire-pulse") from exc75        except subprocess.CalledProcessError as exc:76            raise RuntimeError(f"pactl load-module null-sink failed: {exc.stderr or exc}") from exc77        sink_mod_id = self._parse_module_id(sink_out.stdout)78        try:79            src_out = _pactl(80                "load-module", "module-virtual-source", f"source_name={src_name}",81                f"master={sink_name}.monitor", check=True)82        except subprocess.CalledProcessError as exc:83            # Roll back the null-sink we just created so we don't leak it.84            _pactl("unload-module", str(sink_mod_id), check=False)85            raise RuntimeError(f"pactl load-module virtual-source failed: {exc.stderr or exc}") from exc86        return self._finish("linux", src_name, sink_name, [sink_mod_id, self._parse_module_id(src_out.stdout)])87 88    def _setup_darwin(self) -> dict:89        try:90            out = subprocess.check_output(["system_profiler", "SPAudioDataType"], text=True, encoding='utf-8',91                                          errors='replace', stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL)92        except FileNotFoundError as exc:93            raise RuntimeError("system_profiler not found (macOS-only command)") from exc94        except subprocess.CalledProcessError as exc:95            raise RuntimeError(f"system_profiler failed: {exc.output}") from exc96        if "BlackHole" not in out:97            raise RuntimeError("BlackHole virtual audio device not installed. "98                               "Install via: brew install blackhole-2ch")99        return self._finish("darwin", _BLACKHOLE_DEVICE, _BLACKHOLE_DEVICE, [])100 101    @staticmethod102    def _parse_module_id(stdout: str) -> int:103        """pactl load-module prints the new module ID as the last token of its first line."""104        text = (stdout or "").strip()105        if not text:106            raise RuntimeError("pactl load-module returned empty stdout")107        token = text.splitlines()[0].strip().split()[-1]108        try:109            return int(token)110        except ValueError as exc:111            raise RuntimeError(f"could not parse pactl module id from: {stdout!r}") from exc112 113 114# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----115# Names external plugins imported from this module before the Sep 2026 decomposition.116# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).117# The whole block is removed by reverting the commit that added it.118 119def chrome_fake_audio_flags(bridge_info: dict) -> list[str]:120    """Return Chrome flags for using the fake audio input.121 122    The PulseAudio source is selected via the ``PULSE_SOURCE`` env var,123    which callers must set in Chrome's environment before launch:124 125        env["PULSE_SOURCE"] = bridge_info["device_name"]126 127    On macOS the caller must ensure the system default audio input is128    set to the returned BlackHole device (we do not flip that switch).129    """130    system = platform.system()131    if system == "Linux":132        # Chromium on Linux picks up the PulseAudio source selected via133        # PULSE_SOURCE env var; the fake-ui flag skips the permission134        # prompt so the bot can pick "use my mic" without user input.135        return ["--use-fake-ui-for-media-stream"]136    if system == "Darwin":137        return ["--use-fake-ui-for-media-stream"]138    if system == "Windows":139        raise RuntimeError("windows not supported in v2")140    raise RuntimeError(f"unsupported platform: {system}")141# ---- END PLUGIN-COMPAT ----142