"""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=" 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, )