# Copyright 2026 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # https://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Numeric voice-consistency metrics over per-turn agent recordings. The Gemini analyses in `audio_analysis.py` answer perceptual questions (same speaker? long pauses?) with PASS/FAIL and prose. This module is the other half: signal measurements that do not know what produced the audio and cannot be talked into anything. Per agent turn it measures * integrated loudness (ITU-R BS.1770, LUFS) and within-turn loudness range (95th minus 10th percentile of the 3 s short-term track), * articulation rate in syllables/second, from energy-peak syllable nuclei (de Jong & Wempe 2009 style: intensity peaks with >= 2 dB prominence, above a floor 25 dB under the 99th percentile, voiced at the peak), * median fundamental frequency, in Hz and converted to semitones for cross-turn spread, * a coarse timbre vector (mean MFCC 1..12 over phonated frames), used only for cosine-drift scores across turns, * within-turn voice seams: the median-pitch shift across each pause, from a two-pass speaker-adaptive pitch track (see `intra_turn_seams`). A jump of 11 semitones or more reads as a different speaker; 7 to 11 is audible drift worth a listen. Per call it aggregates turn-to-turn consistency: the loudness standard deviation and the largest adjacent-turn jump, the articulation-rate coefficient of variation, the largest within-turn loudness range, and the spread of median pitch. `DEFAULT_THRESHOLDS` flags the values that listeners reported as "the voice changed" in live reviews. Caveat: pitch on whispered or otherwise aperiodic audio is unreliable — with no glottal pulse the estimator latches onto formant structure and reads roughly double the true value. When a turn's loudness collapses, trust the loudness numbers and ignore its pitch. Requires the optional audio-metrics dependencies:: pip install "cxas-scrapi[audio-metrics]" """ from __future__ import annotations import math import os import re import statistics import wave from itertools import pairwise from typing import Any import numpy as np from scipy.signal import find_peaks try: # optional heavy deps; everything else in the package works without them import parselmouth import pyloudnorm except ImportError: # pragma: no cover - exercised only without the extra parselmouth = None pyloudnorm = None _MISSING_DEPS_MSG = ( "audio metrics need the optional dependencies; install them with " 'pip install "cxas-scrapi[audio-metrics]"' ) #: Ignore fragments shorter than this (clicks / aborted turns). MIN_TURN_S = 0.6 #: Turns shorter than this still get per-turn rows but are excluded from #: call-level consistency scores, so truncated fragments don't fake a jump. MIN_SCORED_TURN_S = 1.0 #: Call-level values above these read as "the voice changed" to listeners. DEFAULT_THRESHOLDS: dict[str, float] = { "max_adj_dlufs": 3.0, "lufs_sd": 1.5, "rate_cov_pct": 12.0, "max_lra_within_turn": 6.0, "f0_median_sd_st": 2.0, } #: Within-turn voice seams, calibrated 2026-10-13 against human labels on #: 4 reviewed calls (26 agent turns; a small sample, re-check as labels #: accumulate). The shift is the median F0 over `SEAM_WINDOW_S` of speech #: on each side of a pause, from a two-pass pitch track: #: >= SEAM_SPEAKER_CHANGE_ST apparent speaker change (caught all 4 #: labelled male->female greeting splices at #: +12.7 st; 0 of 18 unlabelled turns), #: >= SEAM_DRIFT_ST audible voice drift (adds the labelled #: "voice changes between sentences" turns at #: +7.5 st); listen to confirm. #: Below 7 st is ordinary intonation (5 st flagged 11 of 18 clean turns). SEAM_SPEAKER_CHANGE_ST = 11.0 SEAM_DRIFT_ST = 7.0 #: Speech compared on each side of a candidate seam. SEAM_WINDOW_S = 1.5 #: Voiced 10 ms frames needed per side for a usable median. SEAM_MIN_VOICED_FRAMES = 10 #: Energy segments shorter than this have no stable pitch median. SEAM_MIN_SEGMENT_S = 0.35 #: A gap longer than this is a pause between thoughts, not a splice. SEAM_MAX_GAP_S = 1.5 _TURN_NUM_RE = re.compile(r"agent-turn-(\d+)") def hz_to_semitones(f: float, ref: float = 100.0) -> float: return 12.0 * math.log2(f / ref) def _cosine_dist(a: Any, b: Any) -> float: a = np.asarray(a) b = np.asarray(b) if not np.isfinite(a).all() or not np.isfinite(b).all(): return float("nan") denom = np.linalg.norm(a) * np.linalg.norm(b) + 1e-9 return float(1.0 - np.dot(a, b) / denom) def _read_wav(path: str) -> tuple[np.ndarray, int]: """Reads a PCM WAV (the shape CES writes) to mono float32 in [-1, 1].""" with wave.open(path, "rb") as fh: sr = fh.getframerate() width = fh.getsampwidth() n_ch = fh.getnchannels() raw = fh.readframes(fh.getnframes()) if width == 2: y = np.frombuffer(raw, dtype=" 1: y = y.reshape(-1, n_ch).mean(axis=1) return y, sr def _short_term_loudness( meter: Any, y: np.ndarray, sr: int, win: float = 3.0, hop: float = 1.0, ) -> list[float]: """ITU-R BS.1770 short-term loudness track (3 s windows, 1 s hop).""" n = int(win * sr) h = int(hop * sr) vals = [] for s in range(0, max(1, len(y) - n + 1), h): seg = y[s : s + n] if len(seg) < n: break loud = meter.integrated_loudness(seg) if np.isfinite(loud) and loud > -70: vals.append(loud) return vals def _syllable_nuclei(intensity: Any, pitch: Any) -> tuple[int, float, float]: """Energy-peak syllable nuclei (de Jong & Wempe 2009 style). Peaks in the intensity contour with >= 2 dB prominence, above a threshold 25 dB below the 99th percentile, and voiced (F0 defined) at the peak. Returns (n_syllables, phonation_s, voiced_s). """ vals = intensity.values[0] ts = intensity.xs() if len(vals) < 5: return 0, 0.0, 0.0 p99 = np.percentile(vals, 99) thr = max(p99 - 25.0, np.min(vals) + 2.0) peaks, _ = find_peaks(vals, height=thr, prominence=2.0) n = 0 for p in peaks: f0 = pitch.get_value_at_time(ts[p]) if f0 is not None and not math.isnan(f0) and f0 > 0: n += 1 dt = ts[1] - ts[0] if len(ts) > 1 else 0.01 phonation = float(np.sum(vals > thr) * dt) f0s = pitch.selected_array["frequency"] voiced = float(np.sum(f0s > 0) * pitch.dt) return n, phonation, voiced def _speech_segments(y: np.ndarray, sr: int) -> list[tuple[float, float]]: """Energy-gated speech segments [(start_s, end_s)], short gaps merged.""" hop, win = int(0.010 * sr), int(0.030 * sr) if len(y) < win: return [] n = max(1, (len(y) - win) // hop) frames = np.lib.stride_tricks.sliding_window_view(y, win)[::hop][:n] rms = np.sqrt((frames.astype(np.float64) ** 2).mean(axis=1) + 1e-12) on = rms > max(float(rms.max()) * 0.06, 1e-5) raw: list[tuple[float, float]] = [] start = None for i, voiced in enumerate(on): t = i * hop / sr if voiced and start is None: start = t elif not voiced and start is not None: if t - start >= SEAM_MIN_SEGMENT_S: raw.append((start, t)) start = None end = n * hop / sr if start is not None and end - start >= SEAM_MIN_SEGMENT_S: raw.append((start, end)) merged: list[tuple[float, float]] = [] for s, e in raw: if merged and s - merged[-1][1] < 0.15: merged[-1] = (merged[-1][0], e) else: merged.append((s, e)) return merged def _two_pass_pitch(snd: Any) -> tuple[np.ndarray, np.ndarray] | None: """Speaker-adaptive F0 contour (De Looze & Hirst two-pass). A wide first pass estimates the speaker's range; the second re-tracks inside [0.75 * q15, 1.5 * q85]. Single-pass medians over short fragments produce octave errors that read as speaker changes (24 of 25 single-pass hits in the 2026-10-13 review were artifacts; the one human-confirmed switch survives this method at +12.7 st). """ wide = snd.to_pitch(time_step=0.01, pitch_floor=60, pitch_ceiling=600) v = wide.selected_array["frequency"] v = v[v > 0] if len(v) < 3 * SEAM_MIN_VOICED_FRAMES: return None q15, q85 = np.percentile(v, 15), np.percentile(v, 85) tight = snd.to_pitch( time_step=0.01, pitch_floor=max(50.0, 0.75 * q15), pitch_ceiling=min(700.0, 1.5 * q85), ) return np.asarray(tight.xs()), tight.selected_array["frequency"] def _seam_shift_st(xs: np.ndarray, f0: np.ndarray, t: float) -> float | None: """Semitone shift of median F0 across time `t`, or None if too sparse.""" before = f0[(xs >= t - SEAM_WINDOW_S) & (xs < t) & (f0 > 0)] after = f0[(xs >= t) & (xs < t + SEAM_WINDOW_S) & (f0 > 0)] if ( len(before) < SEAM_MIN_VOICED_FRAMES or len(after) < SEAM_MIN_VOICED_FRAMES ): return None return 12.0 * math.log2(float(np.median(after)) / float(np.median(before))) def _rms_db(y: np.ndarray, sr: int, a: float, b: float) -> float: seg = y[int(a * sr) : int(b * sr)] return 20 * math.log10( float(np.sqrt(np.mean(seg.astype(np.float64) ** 2))) + 1e-9 ) def intra_turn_seams( y: np.ndarray, sr: int, snd: Any, mfcc: Any = None ) -> list[dict]: """Voice seams inside one agent turn. Every pause shorter than `SEAM_MAX_GAP_S` between two speech segments is a candidate. Its pitch shift is compared on the two-pass contour over `SEAM_WINDOW_S` of speech on each side, and shifts of at least `SEAM_DRIFT_ST` are reported with a `tier`: `speaker_change` (>= `SEAM_SPEAKER_CHANGE_ST`) or `drift`. Flags closer together than one window are merged into the first. Each seam also carries the level step between the two segments and, when `mfcc` is given, the cosine distance between the mean MFCC 1..12 on each side. A true speaker change moves timbre as well as pitch, while a tracker octave error does not; both numbers are reported for review and are not part of the calibrated tiers. """ segs = _speech_segments(y, sr) contour = _two_pass_pitch(snd) if contour is None or len(segs) < 2: return [] xs, f0 = contour m_xs = m_arr = None if mfcc is not None: m_xs, m_arr = np.asarray(mfcc.xs()), mfcc.to_array()[1:] seams: list[dict] = [] last = -math.inf for (a1, b1), (a2, b2) in pairwise(segs): if a2 - b1 > SEAM_MAX_GAP_S or b1 - last <= SEAM_WINDOW_S: continue st = _seam_shift_st(xs, f0, b1) if st is None or abs(st) < SEAM_DRIFT_ST: continue seam = { "t_s": round(b1, 2), "shift_st": round(st, 1), "tier": "speaker_change" if abs(st) >= SEAM_SPEAKER_CHANGE_ST else "drift", "level_db": round( _rms_db(y, sr, a2, b2) - _rms_db(y, sr, a1, b1), 1 ), } if m_arr is not None: left = m_arr[:, (m_xs >= b1 - SEAM_WINDOW_S) & (m_xs < b1)] right = m_arr[:, (m_xs >= a2) & (m_xs < a2 + SEAM_WINDOW_S)] if left.shape[1] > 10 and right.shape[1] > 10: seam["timbre_dist"] = round( _cosine_dist(left.mean(axis=1), right.mean(axis=1)), 4 ) seams.append(seam) last = b1 return seams #: Voice jumps at render joins (the JUMP flag in `audio_drift`). A turn sent #: as several response parts plays each as a fresh TTS render, which starts #: louder and higher than the previous one ended. Calibrated 2026-10-28 on a #: production collections agent (composite-v2, one official voice): #: * a listener's "the voice changed" answers on 50 clips were predicted #: best by pitch AND loudness rising together (AUC 0.88; loudness alone #: 0.83, pitch alone 0.72); a pitch rise with flat or falling loudness #: was never heard (0 of 9), nor was a ~2 dB within-turn fade; #: * at render joins the step was a median +4.9 st / +5.9 dB, against #: +2.5 st / +2.2 dB at the other pauses of the same turns (670 turns). #: Whole-turn pause scans flag ordinary emphasis too, so only joins are #: measured. One listener and 50 clips: a first calibration. JUMP_ST = 3.0 JUMP_DB = 5.0 #: Speech compared on each side of a join. JUMP_WINDOW_S = 1.0 #: An estimated join is matched to a pause at most this far away. JUMP_SNAP_S = 1.0 #: Voiced / speech 10 ms frames needed per side. JUMP_MIN_VOICED_FRAMES = 25 JUMP_MIN_SPEECH_FRAMES = 40 #: Loudness counts frames within this of the turn's loudest frame. JUMP_SPEECH_RANGE_DB = 35.0 def _join_segments(y: np.ndarray, sr: int) -> list[tuple[float, float]]: """Speech segments for join finding: keeps the short pauses that `_speech_segments` merges away (a render join is often under 150 ms). RMS over 20 ms frames every 10 ms against a per-turn threshold of max(floor, min(3 x 10th-percentile frame, 0.5 x 90th-percentile frame)); dropouts under 120 ms are bridged. The same speech definition the calibration above was measured with. """ hop, win = int(0.010 * sr), int(0.020 * sr) if len(y) < win: return [] frames = np.lib.stride_tricks.sliding_window_view(y, win)[::hop] rms = np.sqrt((frames.astype(np.float64) ** 2).mean(axis=1)) quiet, loud = np.percentile(rms, 10), np.percentile(rms, 90) thr = max(120.0 / 32768.0, min(3.0 * quiet, 0.5 * loud)) segs: list[list[float]] = [] for i in np.flatnonzero(rms >= thr): t = i * hop / sr if segs and t - segs[-1][1] < 0.12: segs[-1][1] = t + win / sr else: segs.append([t, t + win / sr]) return [(a, b) for a, b in segs] def render_join_shares(parts: list[str]) -> list[float]: """Where each join between rendered parts falls, as a fraction of the turn's syllables (empty for a single part).""" syl = [_text_syllables(p) for p in parts] total = sum(syl) out: list[float] = [] acc = 0 for n in syl[:-1]: acc += n if n and 0 < acc < total: out.append(acc / total) return out def _text_syllables(text: str) -> int: total = 0 for word in re.findall(r"[A-Za-z']+|\d", text or ""): w = re.sub(r"[^a-z]", "", word.lower()) total += max(1, len(re.findall(r"[aeiouy]+", w))) if w else 1 return total def render_join_steps( path: str, parts: list[str], jump_st: float = JUMP_ST, jump_db: float = JUMP_DB, ) -> list[dict | None]: """The voice step at each join between a turn's rendered parts. Each join's time is estimated from the parts' share of the turn's syllables, then matched to the nearest pause within `JUMP_SNAP_S` (the join itself is a pause). Per join: `{part, t_s, pause_s, d_st, d_db, jump}` — the median-pitch step in semitones and the speech-loudness step in dB over `JUMP_WINDOW_S` either side, and `jump` when pitch rose >= `jump_st` AND loudness >= `jump_db`. None where no pause was close. Empty for a single-part turn. """ if parselmouth is None: raise ImportError(_MISSING_DEPS_MSG) shares = render_join_shares([p for p in parts if p and p.strip()]) if not shares: return [] y, sr = _read_wav(path) segs = _join_segments(y, sr) if len(segs) < 2: return [None] * len(shares) contour = _two_pass_pitch(parselmouth.Sound(y.astype(np.float64), sr)) hop = int(0.010 * sr) n = len(y) // hop db = 20 * np.log10( np.sqrt( np.mean( y[: n * hop].astype(np.float64).reshape(n, hop) ** 2, axis=1 ) ) + 1e-9 ) lt = (np.arange(n) + 0.5) * 0.010 floor = float(db.max()) - JUMP_SPEECH_RANGE_DB span0, span1 = segs[0][0], segs[-1][1] pauses = [(b1, a2) for (_, b1), (a2, _) in pairwise(segs)] used: set[float] = set() out: list[dict | None] = [] for k, share in enumerate(shares): t = span0 + (span1 - span0) * share near = [ p for p in pauses if abs(p[1] - t) <= JUMP_SNAP_S and p[1] not in used ] if not near: out.append(None) continue end, start = min(near, key=lambda p: abs(p[1] - t)) used.add(start) def side(lo: float, hi: float) -> tuple[float | None, float | None]: st = None if contour is not None: xs, f0 = contour hz = f0[(xs >= lo) & (xs < hi) & (f0 > 0)] if len(hz) >= JUMP_MIN_VOICED_FRAMES: st = hz_to_semitones(float(np.median(hz))) seg = db[(lt >= lo) & (lt < hi) & (db > floor)] loud = ( float(10 * np.log10(np.mean(10 ** (seg / 10)))) if len(seg) >= JUMP_MIN_SPEECH_FRAMES else None ) return st, loud p0, l0 = side(max(0.0, end - JUMP_WINDOW_S), end) p1, l1 = side(start, start + JUMP_WINDOW_S) d_st = None if p0 is None or p1 is None else round(p1 - p0, 2) d_db = None if l0 is None or l1 is None else round(l1 - l0, 2) out.append( { "part": k + 2, "t_s": round(start, 2), "pause_s": round(start - end, 2), "d_st": d_st, "d_db": d_db, "jump": d_st is not None and d_db is not None and d_st >= jump_st and d_db >= jump_db, } ) return out def try_render_join_steps( path: str | None, parts: list[str] ) -> list[dict | None]: """`render_join_steps`, best-effort: [] for one part, no audio, or when the measurement can't run (missing extra, unreadable audio).""" if not path or len([p for p in parts if p and p.strip()]) < 2: return [] try: return render_join_steps(path, parts) except Exception: # noqa: BLE001 - optional signal; no JUMP then return [] def turn_metrics(path: str, meter_cache: dict | None = None) -> dict: """Per-turn features for one agent recording. See the module docstring.""" if parselmouth is None or pyloudnorm is None: raise ImportError(_MISSING_DEPS_MSG) y, sr = _read_wav(path) dur = len(y) / sr feat: dict = {"path": path, "dur_s": round(dur, 3), "sr": sr} if dur < MIN_TURN_S: feat["skipped"] = "too_short" return feat if meter_cache is None: meter_cache = {} if sr not in meter_cache: meter_cache[sr] = pyloudnorm.Meter(sr) meter = meter_cache[sr] # --- loudness lufs = meter.integrated_loudness(y) feat["lufs"] = round(float(lufs), 2) if np.isfinite(lufs) else float("nan") feat["peak_dbfs"] = round( 20 * math.log10(float(np.max(np.abs(y))) + 1e-9), 2 ) stl = _short_term_loudness(meter, y, sr) if len(stl) >= 2: feat["lra_within_turn"] = round( float(np.percentile(stl, 95) - np.percentile(stl, 10)), 2 ) else: feat["lra_within_turn"] = float("nan") # --- prosody snd = parselmouth.Sound(y.astype(np.float64), sampling_frequency=sr) pitch = snd.to_pitch(time_step=0.01, pitch_floor=75, pitch_ceiling=400) intensity = snd.to_intensity(minimum_pitch=75, time_step=0.01) f0 = pitch.selected_array["frequency"] f0v = f0[f0 > 0] if len(f0v) > 5: feat["f0_mean_hz"] = round(float(np.mean(f0v)), 1) feat["f0_median_hz"] = round(float(np.median(f0v)), 1) feat["f0_sd_st"] = round( float(np.std([hz_to_semitones(f) for f in f0v])), 2 ) else: feat["f0_mean_hz"] = float("nan") feat["f0_median_hz"] = float("nan") feat["f0_sd_st"] = float("nan") n_syl, phonation_s, voiced_s = _syllable_nuclei(intensity, pitch) feat["n_syl"] = n_syl feat["phonation_s"] = round(phonation_s, 2) feat["voiced_s"] = round(voiced_s, 2) feat["artic_rate"] = ( round(n_syl / phonation_s, 2) if phonation_s > 0.3 else float("nan") ) feat["speak_rate"] = round(n_syl / dur, 2) feat["pause_frac"] = ( round(1.0 - phonation_s / dur, 2) if dur > 0 else float("nan") ) # --- coarse timbre vector: mean MFCC(1..12) over phonated frames mfcc = snd.to_mfcc( number_of_coefficients=12, window_length=0.025, time_step=0.01 ) m = mfcc.to_array() # (13, nframes); row 0 is c0 (energy) ivals = np.interp(mfcc.xs(), intensity.xs(), intensity.values[0]) thr = max( np.percentile(intensity.values[0], 99) - 25.0, np.min(intensity.values[0]) + 2.0, ) mask = ivals > thr if mask.sum() > 10: feat["_mfcc"] = m[1:, mask].mean(axis=1).tolist() else: feat["_mfcc"] = [float("nan")] * 12 # --- within-turn voice seams feat["seams"] = intra_turn_seams(y, sr, snd, mfcc) feat["max_seam_st"] = max( (abs(x["shift_st"]) for x in feat["seams"]), default=0.0 ) return feat def _turn_number(path: str) -> int: m = _TURN_NUM_RE.search(os.path.basename(path)) return int(m.group(1)) if m else 0 def measure_call( wav_paths: list[str], meter_cache: dict | None = None, thresholds: dict[str, float] | None = None, ) -> tuple[dict, list[dict]]: """Per-call consistency scores over a conversation's agent turns. Args: wav_paths: local paths of the `agent-turn-N.wav` files. Sorted by turn number internally. meter_cache: optional dict reused across calls so the loudness meter is built once per sample rate. thresholds: flag levels; defaults to `DEFAULT_THRESHOLDS`. Returns: (call, turns): `call` holds the aggregate scores plus a `flags` list naming every threshold exceeded; `turns` holds one row per file (rows for skipped fragments carry `skipped`). """ if parselmouth is None or pyloudnorm is None: raise ImportError(_MISSING_DEPS_MSG) if meter_cache is None: meter_cache = {} thresholds = dict(thresholds or DEFAULT_THRESHOLDS) turns = [] for path in sorted(wav_paths, key=_turn_number): feat = turn_metrics(path, meter_cache) feat["turn"] = _turn_number(path) turns.append(feat) valid = [ t for t in turns if "skipped" not in t and t["dur_s"] >= MIN_SCORED_TURN_S and np.isfinite(t.get("lufs", float("nan"))) ] call: dict = {"n_turns": len(turns), "n_valid": len(valid)} if len(valid) >= 2: loud = [t["lufs"] for t in valid] rates = [t["artic_rate"] for t in valid if np.isfinite(t["artic_rate"])] pitches = [ hz_to_semitones(t["f0_median_hz"]) for t in valid if np.isfinite(t["f0_median_hz"]) ] call["lufs_mean"] = round(statistics.mean(loud), 2) call["lufs_sd"] = round(statistics.pstdev(loud), 2) call["lufs_range"] = round(max(loud) - min(loud), 2) jumps = [abs(b - a) for a, b in pairwise(loud)] i = int(np.argmax(jumps)) call["max_adj_dlufs"] = round(max(jumps), 2) call["max_adj_dlufs_turns"] = ( f"{valid[i]['turn']}->{valid[i + 1]['turn']}" ) lra = [ t["lra_within_turn"] for t in valid if np.isfinite(t["lra_within_turn"]) ] call["max_lra_within_turn"] = ( round(max(lra), 2) if lra else float("nan") ) if len(rates) >= 2: call["rate_mean"] = round(statistics.mean(rates), 2) call["rate_cov_pct"] = round( 100 * statistics.pstdev(rates) / statistics.mean(rates), 1 ) drate = [ abs(b - a) / ((a + b) / 2) * 100 for a, b in pairwise(rates) ] call["max_adj_drate_pct"] = round(max(drate), 1) if len(pitches) >= 2: call["f0_median_sd_st"] = round(statistics.pstdev(pitches), 2) call["f0_median_range_st"] = round(max(pitches) - min(pitches), 2) first = valid[0]["_mfcc"] vs_first = [ d for d in (_cosine_dist(first, t["_mfcc"]) for t in valid[1:]) if np.isfinite(d) ] vs_prev = [ d for d in ( _cosine_dist(a["_mfcc"], b["_mfcc"]) for a, b in pairwise(valid) ) if np.isfinite(d) ] if vs_first: call["timbre_drift_vs_t1_mean"] = round( statistics.mean(vs_first), 4 ) call["timbre_drift_vs_t1_max"] = round(max(vs_first), 4) if vs_prev: call["timbre_drift_vs_prev_max"] = round(max(vs_prev), 4) measured = [t for t in turns if "skipped" not in t] call["max_seam_st"] = max( (t.get("max_seam_st", 0.0) for t in measured), default=0.0 ) call["seam_speaker_change_turns"] = [ t["turn"] for t in measured if any(x["tier"] == "speaker_change" for x in t.get("seams", ())) ] call["seam_drift_turns"] = [ t["turn"] for t in measured if t.get("seams") and all(x["tier"] == "drift" for x in t["seams"]) ] call["flags"] = sorted( k for k, limit in thresholds.items() if np.isfinite(call.get(k, float("nan"))) and call[k] > limit ) if call["seam_speaker_change_turns"]: call["flags"].append("intra_turn_speaker_change") if call["seam_drift_turns"]: call["flags"].append("intra_turn_voice_drift") for t in turns: t.pop("_mfcc", None) return call, turns