124 lines
4.4 KiB
Python
124 lines
4.4 KiB
Python
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Tuple
|
|
|
|
import numpy as np
|
|
|
|
|
|
SAMPLE_RATE = 16_000
|
|
FRAME_SAMPLES = 480
|
|
|
|
|
|
class AudioAnalysisError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AudioMetrics:
|
|
recording_duration_ms: int
|
|
speech_duration_ms: int
|
|
internal_silence_ms: int
|
|
internal_pause_ratio: float
|
|
|
|
|
|
def decode_audio_mono(path: Path) -> Tuple[np.ndarray, int]:
|
|
try:
|
|
import av
|
|
except ImportError as exc:
|
|
raise AudioAnalysisError("PyAV is required for audio analysis.") from exc
|
|
|
|
container = av.open(str(path))
|
|
chunks = []
|
|
try:
|
|
audio_stream = next(
|
|
(stream for stream in container.streams if stream.type == "audio"),
|
|
None,
|
|
)
|
|
if audio_stream is None:
|
|
raise AudioAnalysisError("The uploaded file does not contain an audio stream.")
|
|
resampler = av.AudioResampler(format="fltp", layout="mono", rate=SAMPLE_RATE)
|
|
for packet in container.demux(audio_stream):
|
|
for frame in packet.decode():
|
|
for output in resampler.resample(frame):
|
|
chunks.append(output.to_ndarray()[0].astype(np.float32, copy=False))
|
|
for output in resampler.resample(None):
|
|
chunks.append(output.to_ndarray()[0].astype(np.float32, copy=False))
|
|
finally:
|
|
container.close()
|
|
|
|
if not chunks:
|
|
raise AudioAnalysisError("The uploaded audio is empty.")
|
|
return np.concatenate(chunks), SAMPLE_RATE
|
|
|
|
|
|
def analyze_audio(path: Path) -> AudioMetrics:
|
|
samples, sample_rate = decode_audio_mono(path)
|
|
return analyze_samples(samples, sample_rate)
|
|
|
|
|
|
def analyze_samples(samples: np.ndarray, sample_rate: int = SAMPLE_RATE) -> AudioMetrics:
|
|
if samples.ndim != 1:
|
|
samples = samples.reshape(-1)
|
|
if samples.size == 0 or sample_rate <= 0:
|
|
raise AudioAnalysisError("The uploaded audio is empty.")
|
|
|
|
recording_duration_ms = max(1, int(round(samples.size / sample_rate * 1000)))
|
|
window = max(1, int(round(sample_rate * 0.03)))
|
|
complete_frames = int(np.ceil(samples.size / window))
|
|
padded = np.pad(samples, (0, complete_frames * window - samples.size))
|
|
frames = padded.reshape(complete_frames, window).astype(np.float64, copy=False)
|
|
rms = np.sqrt(np.mean(frames * frames, axis=1))
|
|
|
|
signal_level = float(np.percentile(rms, 95))
|
|
if signal_level < 0.002:
|
|
raise AudioAnalysisError("No usable speech was detected in the recording.")
|
|
noise_floor = float(np.percentile(rms, 10))
|
|
threshold = min(signal_level * 0.45, max(0.006, noise_floor * 2.2))
|
|
speech = rms >= threshold
|
|
|
|
# Treat very short gaps inside a word as speech, then reject short clicks.
|
|
_bridge_false_runs(speech, max_frames=5)
|
|
_remove_true_runs(speech, max_frames=2)
|
|
speech_indexes = np.flatnonzero(speech)
|
|
if speech_indexes.size == 0:
|
|
raise AudioAnalysisError("No usable speech was detected in the recording.")
|
|
|
|
frame_ms = window / sample_rate * 1000
|
|
speech_duration_ms = max(1, int(round(speech.sum() * frame_ms)))
|
|
first = int(speech_indexes[0])
|
|
last = int(speech_indexes[-1])
|
|
internal_frames = max(1, last - first + 1)
|
|
internal_silence_frames = int((~speech[first : last + 1]).sum())
|
|
internal_silence_ms = int(round(internal_silence_frames * frame_ms))
|
|
internal_pause_ratio = internal_silence_frames / internal_frames
|
|
return AudioMetrics(
|
|
recording_duration_ms=recording_duration_ms,
|
|
speech_duration_ms=speech_duration_ms,
|
|
internal_silence_ms=internal_silence_ms,
|
|
internal_pause_ratio=round(internal_pause_ratio, 4),
|
|
)
|
|
|
|
|
|
def _bridge_false_runs(values: np.ndarray, max_frames: int) -> None:
|
|
start = None
|
|
for index, value in enumerate(values):
|
|
if not value and start is None:
|
|
start = index
|
|
elif value and start is not None:
|
|
if start > 0 and index - start <= max_frames:
|
|
values[start:index] = True
|
|
start = None
|
|
|
|
|
|
def _remove_true_runs(values: np.ndarray, max_frames: int) -> None:
|
|
start = None
|
|
for index, value in enumerate(values):
|
|
if value and start is None:
|
|
start = index
|
|
elif not value and start is not None:
|
|
if index - start <= max_frames:
|
|
values[start:index] = False
|
|
start = None
|
|
if start is not None and len(values) - start <= max_frames:
|
|
values[start:] = False
|