_PactlBackend

def __init__(self, timeout: float) -> None:
    self._timeout = timeout

async def run(self, *args: str) -> str:
    """Run one pactl command and return its stdout.

    Raises:
        AudioControlError: pactl is missing, failed, or timed out.
    """
    logger.debug("pactl fallback: pactl %s", " ".join(args))
    return await self._exec(*args)

async def _exec(self, *args: str) -> str:
    env = dict(os.environ, LC_ALL="C", LANGUAGE="C")
    try:
        proc = await asyncio.create_subprocess_exec(
            "pactl", *args, env=env,
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    except Exception as e:
        raise AudioControlError(f"pactl unavailable: {e}") from e
    try:
        out, err = await asyncio.wait_for(proc.communicate(), timeout=self._timeout)
    except asyncio.TimeoutError:
        # Reap it rather than leaking one blocked pactl per call on a
        # stuck server.
        with contextlib.suppress(ProcessLookupError):
            proc.kill()
        await proc.wait()
        raise AudioControlError(f"pactl {' '.join(args)} timed out") from None
    if proc.returncode != 0:
        raise AudioControlError(
            f"pactl {' '.join(args)} failed: {err.decode(errors='replace').strip()}")
    return out.decode(errors="replace")

async def sink_list(self) -> List[PulseNode]:
    return _parse_pactl_list(await self.run("list", "sinks"))

async def source_list(self) -> List[PulseNode]:
    return _parse_pactl_list(await self.run("list", "sources"))

async def source_output_list(self) -> List[PulseNode]:
    return _parse_pactl_list(await self.run("list", "source-outputs"))

async def server_defaults(self) -> Tuple[Optional[str], Optional[str]]:
    sink = source = None
    for line in (await self.run("info")).splitlines():
        key, _, value = line.partition(":")
        if key.strip() == "Default Sink":
            sink = value.strip() or None
        elif key.strip() == "Default Source":
            source = value.strip() or None
    return sink, source

async def module_load(self, name: str, args: str) -> int:
    out = (await self.run("load-module", name, *args.split())).strip()
    if not out.isdigit():
        raise AudioControlError(f"load-module {name} returned no index: {out!r}")
    return int(out)

async def module_unload(self, index: int) -> None:
    await self.run("unload-module", str(index))

async def sink_default_set(self, name: str) -> None:
    await self.run("set-default-sink", name)

async def source_default_set(self, name: str) -> None:
    await self.run("set-default-source", name)

async def source_output_move(self, output_index: int, source_index: int) -> None:
    await self.run("move-source-output", str(output_index), str(source_index))