AudioControl

def __init__(self, client_name: str, connect_timeout: float = 2.0,
             op_timeout: float = 5.0) -> None:
    self.client_name = client_name
    self._pulse = (_PulsectlBackend(client_name, connect_timeout, op_timeout)
                   if PULSE_AVAILABLE else None)
    self._pactl = _PactlBackend(op_timeout)
    self._backend: Any = None
    self._fallback_logged = False

async def __aenter__(self) -> "AudioControl":
    await self.open()
    return self

async def __aexit__(self, *exc: Any) -> None:
    await self.aclose()

def _announce_fallback(self, reason: str) -> None:
    global _fallback_announced
    if self._fallback_logged:
        return
    self._fallback_logged = True
    level = logging.DEBUG if _fallback_announced else logging.WARNING
    _fallback_announced = True
    logger.log(level, f"Sound server control: {reason}; using pactl subprocesses.")

async def open(self) -> bool:
    """Settle the backend: connect the bindings, else engage the pactl fallback.

    Returns:
        True when a backend is ready; False when neither the bindings nor
        pactl can reach a server. The fallback, once engaged, stays until
        `aclose`.
    """
    if self._backend is self._pactl:
        return True
    if self._backend is self._pulse and self._pulse is not None and self._pulse.connected:
        return True
    if self._pulse is not None:
        try:
            await self._pulse.connect()
            self._backend = self._pulse
            return True
        except Exception as e:
            self._announce_fallback(f"connection failed ({e})")
    else:
        self._announce_fallback("pulsectl_asyncio is unavailable")
    self._backend = self._pactl
    try:
        await self._pactl.server_defaults()
    except AudioControlError as e:
        logger.warning(f"Sound server control unavailable: {e}")
        return False
    return True

async def aclose(self) -> None:
    """Release the connection; the next operation reopens it."""
    self._backend = None
    if self._pulse is not None:
        await self._pulse.aclose()

async def _op(self, what: str, fn: Callable[[Any], Awaitable[T]], default: T) -> T:
    """Run `fn(backend)`, reconnecting once after a lost connection.

    Returns `default` (after one warning) when the operation fails, times
    out, or no backend is usable, so callers degrade rather than raise.
    """
    for attempt in (0, 1):
        if self._backend is None or (
                self._backend is self._pulse and not self._pulse.connected):
            if not await self.open():
                return default
        try:
            return await fn(self._backend)
        except asyncio.TimeoutError:
            logger.warning(f"Sound server did not answer ({what}); connection abandoned.")
            return default
        except AudioControlError as e:
            lost = (self._backend is self._pulse and self._pulse is not None
                    and not self._pulse.connected)
            if lost and attempt == 0:
                logger.info(f"Sound server connection lost during {what}; reconnecting.")
                continue
            logger.warning(f"Sound server control failed ({what}): {e}")
            return default
    return default

async def sinks(self) -> List[PulseNode]:
    return await self._op("sink list", lambda b: b.sink_list(), [])

async def sources(self) -> List[PulseNode]:
    return await self._op("source list", lambda b: b.source_list(), [])

async def default_devices(self) -> Tuple[Optional[str], Optional[str]]:
    """``(default sink name, default source name)``; None when unknown."""
    return await self._op("server info", lambda b: b.server_defaults(), (None, None))

async def load_module(self, name: str, args: str) -> Optional[int]:
    """Load a server module; its index, or None when the load failed."""
    return await self._op(f"load {name}", lambda b: b.module_load(name, args), None)

async def unload_module(self, index: int) -> bool:
    async def run(b: Any) -> bool:
        await b.module_unload(index)
        return True
    return await self._op(f"unload module {index}", run, False)

async def set_default_sink(self, name: str) -> bool:
    async def run(b: Any) -> bool:
        await b.sink_default_set(name)
        return True
    return await self._op(f"default sink {name}", run, False)

async def set_default_source(self, name: str) -> bool:
    async def run(b: Any) -> bool:
        await b.source_default_set(name)
        return True
    return await self._op(f"default source {name}", run, False)

async def _wait_for(self, list_fn: Callable[[], Awaitable[List[PulseNode]]],
                    names: Sequence[str], timeout: float = 2.0) -> Optional[PulseNode]:
    """Poll a listing until an object named in `names` appears.

    PipeWire creates sinks and sources asynchronously, so a freshly loaded
    module's object may not be listed immediately.
    """
    deadline = asyncio.get_running_loop().time() + timeout
    while True:
        for node in await list_fn():
            if node.name in names:
                return node
        if asyncio.get_running_loop().time() >= deadline:
            return None
        await asyncio.sleep(0.1)

async def ensure_null_sink(self, name: str) -> bool:
    """Make sure a null sink called `name` exists, loading it if needed."""
    if any(s.name == name for s in await self.sinks()):
        return True
    logger.info(f"Sink '{name}' not found. Creating it...")
    if await self.load_module("module-null-sink", f"sink_name={name}") is None:
        return False
    if await self._wait_for(self.sinks, [name]) is not None:
        logger.info(f"Created sink '{name}'.")
        return True
    logger.error(f"Loaded module-null-sink for '{name}' but it never appeared.")
    return False

async def ensure_capture_sink(self, audio_device_name: Optional[str]) -> bool:
    """Make sure the sink whose monitor the server captures exists.

    A container's PipeWire or PulseAudio comes up with no sink at all when
    the host exposes no sound card, so the monitor source named by the
    configured audio device is missing and pcmflux gives up after its retry
    budget. The microphone control plane creates the same sink, but only
    once a client sends mic data, which server-to-client audio must not
    wait for.

    Args:
        audio_device_name: The configured capture device (a ``.monitor``
            suffix is stripped to derive the sink name); ``output`` when
            unset.

    Returns:
        True when the sink is present afterwards. Best effort: False only
        means the capture will fail for the usual reasons, so callers
        proceed and let pcmflux report.
    """
    return await self.ensure_null_sink(capture_sink_name(audio_device_name))

async def resolve_capture_source(self, audio_device_name: Optional[str]) -> Optional[str]:
    """The source to capture: the configured one if it exists, else a monitor.

    Falls back to the default sink's monitor, then to PipeWire's
    ``auto_null.monitor``.

    Returns:
        The source name to capture from, or None when no usable source
        exists (or the server could not be queried).
    """
    default_sink, _ = await self.default_devices()
    default_monitor = f"{default_sink}.monitor" if default_sink else None
    if default_sink:
        logger.info(f"Default sink: '{default_sink}'")
    else:
        logger.warning("Could not determine the default sink.")
    available = {s.name for s in await self.sources()}
    if not available:
        logger.error("Failed to enumerate audio sources.")
        return None
    if audio_device_name and audio_device_name in available:
        logger.info(f"Configured audio device '{audio_device_name}' is valid.")
        return audio_device_name
    if audio_device_name:
        logger.warning(
            f"Configured audio device '{audio_device_name}' not found in available sources.")
    if default_monitor and default_monitor in available:
        logger.info(f"Falling back to the default sink's monitor: '{default_monitor}'")
        return default_monitor
    if PIPEWIRE_NULL_MONITOR in available:
        logger.info(
            f"Default sink monitor not available; falling back to '{PIPEWIRE_NULL_MONITOR}'")
        return PIPEWIRE_NULL_MONITOR
    logger.error(
        "No valid audio source found. Audio capture will likely fail. "
        f"Available sources: {sorted(available)}")
    return None

async def route_pcmflux(self, targets: Sequence[str]) -> Optional[str]:
    """Put the pcmflux record stream on one of `targets` if it strayed.

    PipeWire often ignores a recording app's requested device and attaches
    it to the default source, particularly across streaming-mode switches.

    Returns:
        The name of the source pcmflux records from afterwards, or None
        when no pcmflux stream (or no target source) exists.
    """
    wanted = [t for t in targets if t]

    async def run(b: Any) -> Optional[str]:
        sources = await b.source_list()
        outputs = await b.source_output_list()
        stream = next((o for o in outputs
                       if o.proplist.get("application.name") == PCMFLUX_APP_NAME), None)
        if stream is None:
            logger.debug("pcmflux has no record stream to route.")
            return None
        by_index = {s.index: s for s in sources}
        current = by_index.get(stream.source)
        if current is not None and current.name in wanted:
            logger.info(f"pcmflux correctly connected to '{current.name}'")
            return current.name
        target = next((s for s in sources if s.name in wanted), None)
        if target is None:
            logger.warning(f"Routing enforcement: no target source among {wanted} exists.")
            return current.name if current else None
        logger.warning(
            f"pcmflux connected to '{current.name if current else stream.source}', "
            f"moving it to '{target.name}'")
        await b.source_output_move(stream.index, target.index)
        return target.name

    return await self._op("pcmflux routing", run, None)

async def ensure_virtual_microphone(
    self, audio_device_name: Optional[str], is_pcmflux_capturing: bool,
) -> Tuple[Optional[int], bool]:
    """Provision the SelkiesVirtualMic control plane shared by both transports.

    Creates the ``input`` and capture null sinks, loads module-virtual-source
    bridging ``input.monitor`` to a recordable source, and makes them the
    system default sink/source so an app recording the default source hears
    the client's forwarded mic. The PCM data plane (pcmflux AudioPlayback
    into the ``input`` sink) belongs to the caller.

    Idempotent: an existing SelkiesVirtualMic is reused, so the websockets
    mic path and the WebRTC mic playback never double-load the module when
    both are live.

    Args:
        audio_device_name: The capture device name whose sink half becomes
            the default output sink.
        is_pcmflux_capturing: When True, verify pcmflux's record stream is
            attached to a valid capture target and move it if not.

    Returns:
        ``(module_index, owns_module)``. `owns_module` is True only when
        THIS call loaded the module, so a caller that merely reused an
        existing source never unloads it out from under the other
        transport on teardown. ``(None, False)`` when the module load could
        not be verified.
    """
    output_sink = capture_sink_name(audio_device_name)
    for sink_name in (VIRTUAL_MIC_SINK, output_sink):
        await self.ensure_null_sink(sink_name)
    if await self.set_default_sink(output_sink):
        logger.info(f"Set system default sink to '{output_sink}'.")

    existing = next((s for s in await self.sources() if s.name in VIRTUAL_MIC_SOURCE_NAMES), None)
    if existing is not None:
        logger.info(f"Virtual source '{existing.name}' (index {existing.index}) already exists.")
        master = existing.proplist.get("device.master_device")
        if master is not None and master != VIRTUAL_MIC_MASTER:
            logger.warning(
                f"Existing source '{existing.name}' is linked to '{master}', "
                f"not '{VIRTUAL_MIC_MASTER}'.")
        module_index: Optional[int] = existing.owner_module
        owns_module = False
        await self.set_default_source(existing.name)
    else:
        logger.info(f"Virtual source '{VIRTUAL_MIC_SOURCE}' not found. Loading module...")
        module_index = await self.load_module(
            "module-virtual-source",
            f"source_name={VIRTUAL_MIC_SOURCE} master={VIRTUAL_MIC_MASTER}")
        if module_index is None:
            return None, False
        owns_module = True
        created = await self._wait_for(self.sources, VIRTUAL_MIC_SOURCE_NAMES)
        if created is None:
            logger.error(
                f"Loaded module {module_index} but source '{VIRTUAL_MIC_SOURCE}' never appeared.")
            await self.unload_module(module_index)
            return None, False
        logger.info(f"Created source '{created.name}' (index {created.index}).")
        if await self.set_default_source(created.name):
            logger.info(f"Set system default source to '{created.name}'.")

    if is_pcmflux_capturing:
        targets = [audio_device_name or "", PIPEWIRE_NULL_MONITOR]
        await self.route_pcmflux(targets)

    logger.info(f"Virtual microphone '{VIRTUAL_MIC_SOURCE}' is ready for microphone forwarding.")
    return module_index, owns_module