idriz/idriz/audio.py

"""Mikrofon i zvučnik preko PipeWire-a (pacat), PCM16 mono 24 kHz kako traži Realtime API."""
import array
import asyncio
import logging
import math
import time

log = logging.getLogger("idriz.audio")

RATE = 24000
BYTES_PER_SEC = RATE * 2  # s16le mono
CHUNK = 2400  # 50 ms


def _pacat_args(mode: str, device: str) -> list[str]:
    args = ["pacat", mode, "--raw", "--format=s16le", f"--rate={RATE}", "--channels=1", "--latency-msec=60"]
    if device:
        args.append(f"--device={device}")
    return args


def apply_gain(pcm: bytes, gain: float) -> bytes:
    if gain == 1.0:
        return pcm
    samples = array.array("h", pcm)
    for i, v in enumerate(samples):
        x = int(v * gain)
        samples[i] = 32767 if x > 32767 else (-32768 if x < -32768 else x)
    return samples.tobytes()


def tone(freq: float = 880.0, ms: int = 150, volume: float = 0.3) -> bytes:
    """Kratak signalni ton (PCM16 mono)."""
    n = RATE * ms // 1000
    samples = array.array("h", (int(volume * 32767 * math.sin(2 * math.pi * freq * i / RATE) * min(1.0, (n - i) / (n * 0.3))) for i in range(n)))
    return samples.tobytes()


def tone_on() -> bytes:
    """Mikrofon uključen: dva tona naviše."""
    return tone(660, 120) + tone(990, 160)


def tone_off() -> bytes:
    """Mikrofon isključen / prekid govora: dva tona naniže."""
    return tone(990, 120) + tone(440, 220)


async def source_present(name: str) -> bool:
    """Da li PipeWire trenutno ima izvor tog imena (USB mikrofon nestane kad se isključi prekidač)."""
    if not name:
        return True
    proc = await asyncio.create_subprocess_exec("pactl", "list", "sources", "short",
                                                stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL)
    out, _ = await proc.communicate()
    return any(name in line for line in out.decode(errors="replace").splitlines())


class Mic:
    def __init__(self, device: str = "", gain: float = 1.0) -> None:
        self.device = device
        self.gain = gain
        self.proc: asyncio.subprocess.Process | None = None

    @property
    def running(self) -> bool:
        return self.proc is not None and self.proc.returncode is None

    async def start(self) -> None:
        self.proc = await asyncio.create_subprocess_exec(
            *_pacat_args("--record", self.device),
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
        )
        log.info("mikrofon pokrenut (pid %s)", self.proc.pid)

    async def read(self) -> bytes:
        assert self.proc and self.proc.stdout
        return apply_gain(await self.proc.stdout.read(CHUNK), self.gain)

    async def stop(self) -> None:
        if self.proc and self.proc.returncode is None:
            self.proc.kill()
            await self.proc.wait()
        self.proc = None


class Speaker:
    """Reproducira PCM; pamti do kada traje reprodukcija da bi se mikrofon utišao (half-duplex)."""

    def __init__(self, device: str = "") -> None:
        self.device = device
        self.proc: asyncio.subprocess.Process | None = None
        self.playing_until = 0.0
        self._lock = asyncio.Lock()

    async def start(self) -> None:
        self.proc = await asyncio.create_subprocess_exec(
            *_pacat_args("--playback", self.device),
            stdin=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,
        )
        log.info("zvučnik pokrenut (pid %s)", self.proc.pid)

    async def write(self, pcm: bytes) -> None:
        async with self._lock:
            if not self.proc or self.proc.returncode is not None:
                await self.start()
            assert self.proc and self.proc.stdin
            now = time.monotonic()
            base = max(now, self.playing_until)
            self.playing_until = base + len(pcm) / BYTES_PER_SEC
            try:
                self.proc.stdin.write(pcm)
                await self.proc.stdin.drain()
            except (BrokenPipeError, ConnectionResetError):
                log.warning("zvučnik prekinut, restart")
                await self.start()

    def is_playing(self, tail: float = 0.35) -> bool:
        return time.monotonic() < self.playing_until + tail

    async def clear(self) -> None:
        """Odbaci sve što još nije pušteno (prekid govora)."""
        async with self._lock:
            if self.proc and self.proc.returncode is None:
                self.proc.kill()
                await self.proc.wait()
            self.playing_until = 0.0
            await self.start()

    async def stop(self) -> None:
        if self.proc and self.proc.returncode is None:
            self.proc.kill()
            await self.proc.wait()