app/adapters/audio/decode.py

"""Getting a phone recording into a NumPy array, and refusing when it cannot.

Directive §26 asks the farmer to "stand quietly in the house and record 30
seconds". Whatever the phone hands over — `.m4a`, `.opus`, `.ogg`, `.3gp` — has
to become mono float samples at one known rate before any of the arithmetic in
`features` or `events` means anything.

**The service has no audio library.** There is no `soundfile`, no `librosa`, no
`av`; `cv2` bundles FFmpeg but exposes no audio path, and `wave` in the standard
library reads WAV and nothing else. So this module shells out to an `ffmpeg`
binary, which is a real external dependency and is treated as one: `available()`
looks for it, and the adapter that wraps this reports unavailable rather than
raising from the middle of a request.

**Two things about that binary are recorded rather than assumed.**

*It writes a file, not a pipe.* The obvious form is `-f s16le -` into
`subprocess`. That fails on any FFmpeg configured with a muxer whitelist, and
the build on this machine is one — `--disable-muxers --enable-muxer='webm,opus,
mp4,wav,...'` has no `s16le` in it, so the pipe form exits 234 with *"Requested
output format 's16le' is not known"*. Writing a temporary `.wav` and reading it
back with `wave` costs one file and works against every build, including the
minimal ones.

*Its licence is not this repository's to assume.* See
`adapters/licences.py:ffmpeg-cli`. The build here is `--enable-gpl
--enable-nonfree`, which is the one configuration FFmpeg may not be
redistributed under at all. That does not reach Animap's own code — calling a
separate program over a pipe is not linking, and `app/adapters/licences.py`
records the reasoning — but it does mean **the binary on a developer's laptop is
not the binary a deployment may ship**, and that is a deployment finding rather
than a footnote.
"""

from __future__ import annotations

import shutil
import subprocess
import tempfile
import wave
from dataclasses import dataclass
from pathlib import Path

import numpy as np

#: What everything downstream assumes, in hertz.
#:
#: 16 kHz resolves to 8 kHz, and a chicken snick's energy is in the low
#: kilohertz — Mahdavian et al. and the broader poultry-audio literature place
#: sneeze and rale energy under 6 kHz. 22.05 or 44.1 kHz would carry more of the
#: transient's top edge and would quadruple the STFT cost for a band nothing
#: here measures.
#:
#: **It is fixed rather than passed through** because every threshold in
#: `events.py` is a property of the analysis band, and the periodicity module
#: next door is a standing lesson in what happens when a constant derived at one
#: band is applied at another.
SAMPLE_RATE_HZ = 16_000

#: Longest clip this will decode, in seconds. §26 asks for 30 and prefers 60;
#: §27 is explicit that continuous monitoring is a different capability with a
#: fixed microphone. Ten minutes is far past a spot check and is here to stop a
#: mis-sent file eating the container's memory, not to express a product limit.
MAX_SECONDS = 600.0

#: How long to let FFmpeg run. A 60-second clip transcodes in well under a
#: second; anything near this is a malformed file FFmpeg is chewing on.
DECODE_TIMEOUT_SECONDS = 120


class AudioUnreadable(RuntimeError):
    """The recording could not be decoded, so there is nothing to measure."""


class DecoderMissing(RuntimeError):
    """No `ffmpeg` on PATH. A missing tool, not a broken recording."""


@dataclass(frozen=True)
class Recording:
    """Mono float samples in [-1, 1], plus what it took to get them."""

    samples: np.ndarray
    sample_rate_hz: int
    #: The file it came from, for a result that has to be traced back.
    source: str
    #: Before resampling, so a clip recorded at 8 kHz is diagnosable later — it
    #: has no content above 4 kHz however it is resampled, and half the band
    #: this module analyses is empty for it.
    source_sample_rate_hz: int
    source_channels: int

    @property
    def duration_seconds(self) -> float:
        return len(self.samples) / self.sample_rate_hz

    @property
    def is_silent(self) -> bool:
        """No signal at all, as distinct from no events.

        A muted microphone and a quiet house are different findings and only
        one of them is about the birds.
        """
        return float(np.max(np.abs(self.samples), initial=0.0)) < 1e-6


def ffmpeg_path() -> str | None:
    return shutil.which("ffmpeg")


def ffprobe_path() -> str | None:
    return shutil.which("ffprobe")


def available() -> bool:
    return ffmpeg_path() is not None


def _probe(path: Path) -> tuple[int, int]:
    """The source rate and channel count, or `(0, 0)` when ffprobe is absent.

    Recorded rather than required. The decode does not need it — FFmpeg
    resamples whatever it finds — but a rate of 8,000 explains an empty upper
    band better than any later measurement can, and losing that to a missing
    optional tool would be worse than reporting it as unknown.
    """
    probe = ffprobe_path()
    if probe is None:
        return 0, 0
    try:
        result = subprocess.run(
            [probe, "-v", "error", "-select_streams", "a:0", "-show_entries",
             "stream=sample_rate,channels", "-of", "csv=p=0", str(path)],
            capture_output=True, text=True, timeout=30, check=True,
        )
    except (subprocess.SubprocessError, OSError):
        return 0, 0
    parts = result.stdout.strip().split(",")
    try:
        return int(parts[0]), int(parts[1])
    except (IndexError, ValueError):
        return 0, 0


def read_audio(
    path: Path | str,
    *,
    sample_rate_hz: int = SAMPLE_RATE_HZ,
    max_seconds: float = MAX_SECONDS,
) -> Recording:
    """Decode to mono float32 at `sample_rate_hz`.

    Raises `DecoderMissing` when there is no FFmpeg and `AudioUnreadable` when
    there is one and the file defeats it. The two are separate exceptions
    because they need different answers: install a tool, or ask for a different
    recording.
    """
    path = Path(path)
    binary = ffmpeg_path()
    if binary is None:
        raise DecoderMissing(
            "No `ffmpeg` on PATH. This service has no audio decoding library — "
            "no soundfile, no librosa, no av — so a phone recording cannot be "
            "read at all without it. Install FFmpeg, and read "
            "adapters/licences.py:ffmpeg-cli before choosing a build."
        )
    if not path.is_file():
        raise AudioUnreadable(f"{path} is not a file.")

    source_rate, channels = _probe(path)

    with tempfile.TemporaryDirectory(prefix="animap-audio-") as workspace:
        decoded = Path(workspace) / "mono.wav"
        command = [
            binary, "-v", "error", "-nostdin", "-y",
            "-i", str(path),
            # `-t` before the output rather than `-ss`: the cap is on how much
            # is decoded, and a spot check has no reason to start late.
            "-t", f"{max_seconds:.3f}",
            "-map", "a:0?",
            "-ac", "1",
            "-ar", str(sample_rate_hz),
            "-acodec", "pcm_s16le",
            "-f", "wav",
            str(decoded),
        ]
        try:
            result = subprocess.run(
                command, capture_output=True, text=True,
                timeout=DECODE_TIMEOUT_SECONDS,
            )
        except subprocess.TimeoutExpired as expired:
            raise AudioUnreadable(
                f"FFmpeg did not finish decoding {path.name} within "
                f"{DECODE_TIMEOUT_SECONDS} s."
            ) from expired
        except OSError as failure:
            raise AudioUnreadable(f"Could not run ffmpeg: {failure}") from failure

        if result.returncode != 0 or not decoded.is_file():
            raise AudioUnreadable(
                f"FFmpeg could not decode {path.name} "
                f"(exit {result.returncode}): {result.stderr.strip()[:300]}"
            )

        with wave.open(str(decoded)) as handle:
            frames = handle.getnframes()
            width = handle.getsampwidth()
            raw = handle.readframes(frames)

    if width != 2:
        # Only reachable if a future edit changes `-acodec`; asserted rather
        # than assumed because reading 16-bit as 32-bit is silent and produces
        # a plausible-looking waveform of noise.
        raise AudioUnreadable(
            f"Expected 16-bit samples from the decode step, got {width * 8}-bit."
        )

    samples = np.frombuffer(raw, dtype="<i2").astype(np.float32) / 32768.0
    if samples.size < 2:
        raise AudioUnreadable(
            f"{path.name} yielded {samples.size} samples. There is no audio in it."
        )

    return Recording(
        samples=samples,
        sample_rate_hz=sample_rate_hz,
        source=str(path),
        source_sample_rate_hz=source_rate,
        source_channels=channels,
    )


def from_samples(
    samples: np.ndarray, sample_rate_hz: int = SAMPLE_RATE_HZ, *, source: str = "memory",
) -> Recording:
    """A `Recording` over samples that are already in hand.

    The mixer in the respiratory experiment builds its composites in memory, and
    routing them through a temporary file to get a `Recording` would mean the
    measured pipeline and the tested pipeline differed by an encode.
    """
    samples = np.asarray(samples, dtype=np.float32).ravel()
    return Recording(
        samples=samples,
        sample_rate_hz=int(sample_rate_hz),
        source=source,
        source_sample_rate_hz=int(sample_rate_hz),
        source_channels=1,
    )