src/pipeline.py

import numpy as np

from . import features, fusion, models
from .config import SR
from .schema import Result


def analyze_one(path: str) -> Result:
    y = features.load_audio(path)
    feat = features.extract(y)

    speech_segs = models.speech_segments(y)
    snr_gap_db, longest_gap_s = _gap_stats(y, speech_segs, feat.duration_s)

    ser_categorical = models.categorical_emotion(y)
    dim_emotion = models.dimensional_emotion(y)

    tags = models.noise_tags(y)
    noise_type = models.noise_type_from_tags(tags) if tags else ""
    if not noise_type and snr_gap_db is not None and snr_gap_db < 18:
        noise_type = "background noise (unspecified)"

    overlap_s = models.overlap_seconds(path)

    return fusion.fuse(
        feat=feat,
        snr_gap_db=snr_gap_db,
        ser_categorical=ser_categorical,
        dimensional_emotion=dim_emotion,
        noise_type=noise_type,
        overlap_s=overlap_s,
        longest_gap_s=longest_gap_s,
    )


def _gap_stats(
    y: np.ndarray, speech_segs: list[tuple[float, float]] | None, duration_s: float
) -> tuple[float | None, float]:
    if not speech_segs:
        return None, 0.0

    speech_mask = np.zeros(len(y), dtype=bool)
    for start_s, end_s in speech_segs:
        speech_mask[int(start_s * SR) : int(end_s * SR)] = True

    gap = y[~speech_mask]
    speech = y[speech_mask]
    gap_rms = float(np.sqrt(np.mean(gap**2))) if len(gap) else 0.0
    speech_rms = float(np.sqrt(np.mean(speech**2))) if len(speech) else 0.0
    snr_gap_db = 20 * np.log10((speech_rms + 1e-8) / (gap_rms + 1e-8))

    longest_gap_s = 0.0
    prev_end = 0.0
    for start_s, end_s in speech_segs:
        longest_gap_s = max(longest_gap_s, start_s - prev_end)
        prev_end = end_s
    longest_gap_s = max(longest_gap_s, duration_s - prev_end)

    return snr_gap_db, longest_gap_s