3cx-telefonie-middleware/app/audio_processor.py

from __future__ import annotations

import asyncio
import json
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class AudioInfo:
    path: Path
    codec: str | None
    sample_rate: int | None
    channels: int
    channel_layout: str | None
    duration: float | None


class AudioProcessor:

    async def inspect(self, path: str | Path) -> AudioInfo:
        path = Path(path)

        if not path.is_file():
            raise FileNotFoundError(path)

        return await asyncio.to_thread(
            self._inspect_sync,
            path,
        )

    @staticmethod
    def _inspect_sync(path: Path) -> AudioInfo:
        result = subprocess.run(
            [
                "ffprobe",
                "-v", "error",
                "-select_streams", "a:0",
                "-show_entries",
                "stream=codec_name,sample_rate,channels,channel_layout",
                "-show_entries",
                "format=duration",
                "-of", "json",
                str(path),
            ],
            capture_output=True,
            text=True,
            check=True,
        )

        data = json.loads(result.stdout)
        stream = (data.get("streams") or [{}])[0]
        fmt = data.get("format") or {}

        return AudioInfo(
            path=path,
            codec=stream.get("codec_name"),
            sample_rate=(
                int(stream["sample_rate"])
                if stream.get("sample_rate")
                else None
            ),
            channels=int(stream.get("channels") or 1),
            channel_layout=stream.get("channel_layout"),
            duration=(
                float(fmt["duration"])
                if fmt.get("duration")
                else None
            ),
        )

    async def prepare_for_transcription(
        self,
        path: str | Path,
    ) -> list[Path]:

        info = await self.inspect(path)

        if info.channels <= 1:
            return [Path(path)]

        return await asyncio.to_thread(
            self._split_stereo_sync,
            info.path,
        )

    @staticmethod
    def _split_stereo_sync(path: Path) -> list[Path]:
        tmp = Path(
            tempfile.mkdtemp(prefix="3cx-audio-")
        )

        outputs = []

        for channel in (0, 1):
            output = tmp / f"channel-{channel}.wav"

            subprocess.run(
                [
                    "ffmpeg",
                    "-hide_banner",
                    "-loglevel", "error",
                    "-i", str(path),
                    "-map_channel", f"0.0.{channel}",
                    "-ar", "16000",
                    "-ac", "1",
                    "-c:a", "pcm_s16le",
                    str(output),
                ],
                check=True,
            )

            outputs.append(output)

        return outputs