""" Speech-to-text transcription for recorded calls. Audio is downloaded from GCS once, then sent to an OpenAI-compatible /audio/transcriptions endpoint: OpenAI whisper-1 by default, or a cheaper provider (Groq whisper-large-v3-turbo, a self-hosted server) when configured, with OpenAI as the fallback when that provider fails (server-26#297). Returns None on any failure so the intelligence pipeline can still run. """ import asyncio import math import re import tempfile import os from dataclasses import dataclass from typing import Optional from app.internal.logger import logger from app.internal import firestore as fstore from app.internal import ai_health from app.internal import transcript_correction from app.internal import area_context from app.internal.chatter_classifier import gate_chatter from app.config import settings # Whisper treats `prompt` as preceding transcript text, not instructions. # Writing it as actual radio speech primes the vocabulary toward P25 codes # and phrasing before the model hears the audio. # # DO NOT put an enumerated run of ten-codes in here. The original version of # this prompt opened with "10-4. 10-23. 10-20. 10-97. 10-8. ..." and Whisper, # treating that as text it should continue, filled noisy or silent audio with # sequences like "10-4. 10-5. 10-6. ... 10-99." Those hallucinations sailed # straight past the no_speech_prob filter below, because the model is highly # confident the continuation it invented is speech. Codes appear here only # singly and inside a sentence, where there is no series to extend. _WHISPER_PROMPT = ( "Dispatch, go ahead. Copy that, en route. Show me on scene. " "Be advised, units responding. Negative, stand by. " "Post 4, I'm out. Received, thank you. " "Engine and ladder responding to a structure fire. " "Medic on scene with one patient. " "Vehicle accident with injuries, MVA. " "Show me 10-8 and clear." ) # Degenerate-output detection (see _is_degenerate). Tuned to catch Whisper's # repetition failure mode without discarding terse but real radio traffic. _MIN_CODES_FOR_RUN = 6 # ten-codes needed before a run is even considered _RUN_RATIO = 0.7 # share of consecutive pairs that must step by +1 _MIN_SEGMENTS_FOR_REPEAT = 6 # segments needed before repetition is considered _UNIQUE_RATIO = 0.25 # unique/total segment texts at or below this is degenerate _MAX_PHRASE_REPEATS = 8 # identical consecutive phrase repeats allowed in one blob # Whisper's own failure test (openai/whisper transcribe.py, compression_ratio_ # threshold=2.4): text that gzips this well is a loop, whatever its shape. The # two checks above only know two loop shapes; this catches the rest (two # phrases alternating, a list re-emitted) without a model call, so a noise # gate survives with transcript correction switched off (server-26#83). Only # on longer text -- a short string has too little to compress to judge. _MAX_COMPRESSION_RATIO = 2.4 _MIN_CHARS_FOR_COMPRESSION = 200 # Prompt echo (server-26#297): on silence Whisper can emit its own prompt. # With a vocabulary glossary in the prompt that reads as a list of local # street names. See _prompt_echo for why this is kept narrow. _ECHO_MIN_TERMS = 3 _ECHO_WORD_RATIO = 0.6 def _ten_code_run(text: str) -> bool: """True if the text is mostly a counting run of ten-codes. Real traffic uses ten-codes constantly, but never in ascending order — a dispatcher does not say "10-4, 10-5, 10-6". An arithmetic series is the signature of Whisper continuing a pattern rather than hearing one. """ numbers = [int(n) for n in re.findall(r"\b10-(\d{1,2})\b", text)] if len(numbers) < _MIN_CODES_FOR_RUN: return False steps = [b - a for a, b in zip(numbers, numbers[1:])] ascending = sum(1 for s in steps if s == 1) return steps and (ascending / len(steps)) >= _RUN_RATIO def _phrase_loop(text: str) -> bool: """True if one short phrase repeats far more than speech plausibly would. Catches the other repetition mode, e.g. "Dispatch, do you copy?" emitted a dozen times over static. """ parts = [p.strip().lower() for p in re.split(r"[.!?]", text) if p.strip()] if len(parts) <= _MAX_PHRASE_REPEATS: return False repeats = 1 for prev, cur in zip(parts, parts[1:]): repeats = repeats + 1 if cur == prev else 1 if repeats > _MAX_PHRASE_REPEATS: return True return False def _compression_ratio(text: str) -> float: import zlib raw = text.encode("utf-8") return len(raw) / max(1, len(zlib.compress(raw))) def _prompt_echo(text: str, glossary: list[str]) -> bool: """ True if the transcript is the prompt's own glossary read back. Deliberately narrow (review of server-26#297): a real location readback ("Main Street, Croton Avenue, Route 9") names glossary terms too, and that is where map pins come from. So either the lead-in itself came back ("Local names: ..."), or nearly the whole glossary did, in prompt order, with little else around it. """ if not text or not glossary: return False low = text.lower() if "local names" in low: return True if len(glossary) < _ECHO_MIN_TERMS: return False positions = [] for t in glossary: m = re.search(rf"\b{re.escape(t.lower())}\b", low) if m: positions.append(m.start()) if len(positions) < max(_ECHO_MIN_TERMS, math.ceil(0.8 * len(glossary))): return False if positions != sorted(positions): return False words = re.findall(r"[\w'-]+", low) term_words = {w for t in glossary for w in re.findall(r"[\w'-]+", t.lower())} return bool(words) and sum(1 for w in words if w in term_words) / len(words) >= _ECHO_WORD_RATIO def _compression_loop(text: str, segments: list[dict]) -> bool: """ Whisper applies its 2.4 threshold per 30s decode window, and zlib's ratio grows with length, so judging a whole 5-minute call at once would drop busy real traffic. Judge ~30s windows of segments (or ~500-char chunks of a segment-less transcript) and call it a loop only when at least half of the windows long enough to judge are loops. """ windows: list[str] = [] if segments: cur, start = [], None for s in segments: st = float(s.get("start") or 0.0) if start is None: start = st if st - start > 30.0 and cur: windows.append(" ".join(cur)) cur, start = [], st cur.append(s.get("text") or "") if cur: windows.append(" ".join(cur)) else: windows = [text[i:i + 500] for i in range(0, len(text), 500)] judged = [w for w in windows if len(w) >= _MIN_CHARS_FOR_COMPRESSION] if not judged: return False loops = sum(1 for w in judged if _compression_ratio(w) > _MAX_COMPRESSION_RATIO) return loops * 2 >= len(judged) def _is_degenerate(text: str, segments: list[dict], glossary: Optional[list[str]] = None) -> bool: """True if a transcript looks like Whisper output rather than radio traffic. Applied AFTER the per-segment no_speech_prob filter, which does not catch these: the model reports high confidence in text it invented by continuing a pattern, so the only tell is the shape of the output itself. """ if not text: return False if _ten_code_run(text) or _phrase_loop(text): return True if _compression_loop(text, segments): return True if _prompt_echo(text, glossary or []): return True # Near-identical segments repeated across the whole recording. if len(segments) >= _MIN_SEGMENTS_FOR_REPEAT: normalised = {s["text"].strip().lower() for s in segments} if len(normalised) / len(segments) <= _UNIQUE_RATIO: return True return False # ── Providers (server-26#297) ─────────────────────────────────────────────── _GROQ_BASE_URL = "https://api.groq.com/openai/v1" @dataclass(frozen=True) class SttTarget: """One OpenAI-compatible transcription endpoint to try.""" provider: str # "openai" | "groq" | "compat" model: str api_key: Optional[str] base_url: Optional[str] # None -> the OpenAI SDK default vocab_prompt: bool # whether the domain-vocabulary prompt is sent tier: str # ai_health tier this target reports to def stt_targets() -> list[SttTarget]: """ The endpoints to try, in order: the configured provider, then OpenAI as fallback. Empty when nothing usable is configured. `stt_provider="auto"` resolves to Groq only when GROQ_API_KEY is set, so a deploy that has not provisioned the key yet keeps transcribing on OpenAI exactly as before. """ provider = (settings.stt_provider or "auto").strip().lower() if provider == "auto": if settings.groq_api_key: provider = "groq" elif settings.stt_base_url: provider = "compat" # an explicit endpoint is a compat server else: provider = "openai" openai_target = SttTarget( provider="openai", model=settings.stt_model, api_key=settings.openai_api_key, base_url=None, vocab_prompt=False, tier="transcription", ) if provider not in ("groq", "compat"): if provider != "openai": logger.warning(f"Unknown STT_PROVIDER {settings.stt_provider!r} — using OpenAI") return [openai_target] if settings.openai_api_key else [] if provider == "groq": key = settings.stt_api_key or settings.groq_api_key base = settings.stt_base_url or _GROQ_BASE_URL else: # A self-hosted server may not check the key, but the SDK insists on one. key = settings.stt_api_key or "unused" base = settings.stt_base_url targets: list[SttTarget] = [] if key and base: targets.append(SttTarget( provider=provider, model=settings.stt_compat_model, api_key=key, base_url=base, vocab_prompt=settings.stt_vocab_prompt_enabled, tier="transcription", )) else: logger.warning(f"STT provider {provider!r} has no API key/base URL — using OpenAI only") if settings.openai_api_key and (not targets or settings.stt_fallback_to_openai): tier = "transcription_fallback" if targets else "transcription" targets.append(SttTarget(**{**openai_target.__dict__, "tier": tier})) return targets # ── Prompt (server-26#297) ────────────────────────────────────────────────── _CODE_SHAPED = re.compile(r"^\s*\d{1,3}(?:-\d{1,3})+\s*$") _MAX_TERM_CHARS = 40 def _estimate_tokens(text: str) -> int: """ Conservative token estimate for Whisper's GPT-2 tokenizer, without the tokenizer: one per punctuation mark and per digit, ~4 letters per token for words. Over-counts real English slightly, which is the safe side — Whisper drops prompt tokens past 224 from the FRONT, silently. """ n = 0 for piece in re.findall(r"[A-Za-z]+|\d|[^\sA-Za-z\d]", text or ""): n += max(1, math.ceil(len(piece) / 4)) if piece[0].isalpha() else 1 return n def _glossary_terms(context: Optional[dict]) -> list[str]: """ Local names worth spelling for the recogniser, most specific first: the talkgroup/system local_knowledge terms, then vocabulary. Never ten-codes. An enumerated run of codes in the prompt is what made Whisper count "10-4. 10-5. 10-6." over silence (see _WHISPER_PROMPT), so the ten_codes reference and any code-shaped vocabulary term stay out. """ if not context: return [] area = context.get("area") or {} terms = [e.get("term") or "" for e in (area.get("local_knowledge") or []) if isinstance(e, dict)] terms += [t for t in (context.get("vocabulary") or []) if isinstance(t, str)] out: list[str] = [] seen: set[str] = set() for t in terms: t = t.strip() key = t.lower() if not t or key in seen or len(t) > _MAX_TERM_CHARS or _CODE_SHAPED.match(t): continue seen.add(key) out.append(t) return out def build_stt_prompt( talkgroup_name: Optional[str], context: Optional[dict] = None, ) -> tuple[str, list[str]]: """ The STT prompt, and the glossary terms that made it in. Without `context` this is exactly the prompt OpenAI has always been sent. With it, the channel's area and local names go in front of the fixed radio-speech priming, cut to settings.stt_prompt_max_tokens. A glossary is how Whisper's prompting guide says to fix the spelling of proper nouns; the priming text stays last because it is the part that sets style. """ tg_prefix = f"Talkgroup: {talkgroup_name}. " if talkgroup_name else "" if not context: return tg_prefix + _WHISPER_PROMPT, [] area = context.get("area") or {} place = ", ".join(str(area[f]) for f in area_context.PLACE_FIELDS if area.get(f)) head = tg_prefix + (f"{place}. " if place else "") budget = settings.stt_prompt_max_tokens - _estimate_tokens(head + _WHISPER_PROMPT) chosen: list[str] = [] for term in _glossary_terms(context): trial = "Local names: " + ", ".join(chosen + [term]) + ". " if _estimate_tokens(trial) > budget: continue # a shorter later term may still fit chosen.append(term) glossary = ("Local names: " + ", ".join(chosen) + ". ") if chosen else "" prompt = head + glossary + _WHISPER_PROMPT if _estimate_tokens(prompt) > settings.stt_prompt_max_tokens: return tg_prefix + _WHISPER_PROMPT, [] return prompt, chosen # ── Failure reporting ─────────────────────────────────────────────────────── async def _log_transcribe_failure(call_id: str, exc: Exception, target: SttTarget) -> None: """ Log a transcription failure, escalating a permanent condition to ERROR once (via app.internal.ai_health, which also drives the /health/ai endpoint and the Discord degradation alert) and reporting it to the shared registry either way. Transcription failing returns None and the pipeline carries on by design, so a per-call WARNING is invisible: no transcript means no extraction, which means no incident, and the only symptom is calls quietly arriving empty. A network blip is genuinely a warning. An exhausted balance is not -- it will not fix itself and it takes the whole pipeline down with it, so it says so once, loudly, and names the fix. The same failure mode already bit the Gemini correlator twice (a retired model ID, then a depleted balance), which is why this is worth the code. """ text = str(exc) kind = ai_health.classify(text) who = f"{target.provider} ({target.model})" billing_url = { "openai": "https://platform.openai.com/settings/organization/billing", "groq": "https://console.groq.com/settings/billing", }.get(target.provider, "the provider's billing page") if kind == "billing": problem = f"the {target.provider} account cannot be billed" fix = f"top up at {billing_url}" logger.error( f"Transcription: the {target.provider} account cannot be billed -- calls on this " f"provider get no transcript (OpenAI fallback, if configured, is carrying them). " f"Top up at {billing_url}. API said: {text}" ) await ai_health.report_degraded( target.tier, target.provider, target.model, problem, fix, permanent=True ) return if kind == "dead_model": problem = "the STT model is unavailable" fix = "update STT_MODEL / STT_COMPAT_MODEL" logger.error( f"Transcription: the configured model {who} is unavailable -- calls on this provider " f"get no transcript. Update STT_MODEL / STT_COMPAT_MODEL. API said: {text}" ) await ai_health.report_degraded( target.tier, target.provider, target.model, problem, fix, permanent=True ) return logger.warning(f"Transcription via {who} failed for call {call_id}: {text}") await ai_health.report_degraded( target.tier, target.provider, target.model, "transient API error", "no action needed unless this persists", permanent=False, ) # ── Pipeline entry point ──────────────────────────────────────────────────── async def _transcribe_with( call_id: str, tmp_path: str, target: SttTarget, prompt: str, glossary: list[str], ) -> dict: """One target, with the one retry on degenerate output. Raises on API failure.""" result = await asyncio.to_thread(_sync_transcribe_file, tmp_path, target, prompt, glossary) # A hallucination is a coin-flip, not a property of the clip: call # e49ea32c produced a 56-word ten-code counting run on one attempt and # ordinary speech on the next, same audio and temperature=0. Discarding # on the first bad roll threw away a recoverable transcript, so spend # one more request before giving up. if result["degenerate"] and settings.stt_retry_on_degenerate: logger.info(f"Retrying transcription for call {call_id} after degenerate output") result = await asyncio.to_thread(_sync_transcribe_file, tmp_path, target, prompt, glossary) if result["degenerate"]: logger.warning(f"Transcription for call {call_id} was degenerate twice — giving up") return result async def transcribe_call( call_id: str, gcs_uri: str, talkgroup_name: Optional[str] = None, system_id: Optional[str] = None, talkgroup_id: Optional[int] = None, node_id: Optional[str] = None, ) -> tuple[Optional[str], list[dict]]: """ Transcribe audio at the given GCS URI and store the result in Firestore. Returns: (transcript, segments) — segments is a list of {start, end, text} dicts, one per detected transmission. Empty list if transcription failed. """ if not gcs_uri or not gcs_uri.startswith("gs://"): return None, [] targets = stt_targets() if not targets: logger.warning("No STT provider configured (OPENAI_API_KEY / GROQ_API_KEY unset) — transcription disabled.") return None, [] context: Optional[dict] = None if any(t.vocab_prompt for t in targets): try: context = await transcript_correction.resolve_context(system_id, talkgroup_id) except Exception as e: logger.warning(f"Could not resolve STT vocabulary for call {call_id}: {e}") try: tmp_path = await asyncio.to_thread(_download, gcs_uri) except Exception as e: # Not a provider failure: classify() would read a GCS "not found" as a # retired model and page someone to change STT_MODEL. logger.warning(f"Could not download audio for call {call_id} ({gcs_uri}): {e}") return None, [] result: Optional[dict] = None used: Optional[SttTarget] = None try: for i, target in enumerate(targets): prompt, glossary = build_stt_prompt(talkgroup_name, context if target.vocab_prompt else None) try: result = await _transcribe_with(call_id, tmp_path, target, prompt, glossary) except Exception as e: await _log_transcribe_failure(call_id, e, target) if i + 1 < len(targets): logger.warning( f"Transcription for call {call_id}: {target.provider} failed, " f"falling back to {targets[i + 1].provider} ({targets[i + 1].model})" ) continue used = target break finally: try: os.unlink(tmp_path) except OSError: pass if used is None or result is None: return None, [] # No exception means the provider call itself succeeded (this also # covers transcripts discarded as degenerate/hallucinated output — # that's a filtering decision, not a provider failure), so that tier is # healthy and any prior degradation clears. A fallback success clears # only the fallback tier: the primary is still broken and must stay red. await ai_health.report_healthy(used.tier) transcript: Optional[str] = result["text"] segments: list[dict] = result["segments"] if transcript: updates: dict = { "transcript": transcript, # Which endpoint produced this text, and how much audio it billed, # so cost (server-26#45) and an A/B of providers can be read off # the call docs instead of guessed. "stt_provider": used.provider, "stt_model": used.model, } if result.get("duration") is not None: updates["stt_audio_seconds"] = result["duration"] if result.get("confidence") is not None: # Mean per-token probability, exp(avg_logprob) weighted by segment # length. The only STT-side confidence there is; the alert gate # reads it (server-26#300). updates["stt_confidence"] = result["confidence"] if segments: updates["segments"] = segments # Second opinion, before anything downstream sees the text. Whisper # mishears proper nouns confidently, and extraction/embedding/ # correlation all consume the transcript — correcting it afterwards # (which is where it used to live, inside the extraction prompt) meant # every one of them reasoned over known-bad text. server-26#36. # Correction is a second model call plus a Places lookup per proposed # location, so it is real spend that used to be reachable only through # an env var and an ansible run. That made an "STT-only" evaluation # window not STT-only, and its cost unattributable (server-26#76, #45). from app.internal.feature_flags import resolve_flags _, _ai_flag = await resolve_flags(system_id) # server-26#298: radio housekeeping (roll call, a bare "Post 4, 10-8, # copy") has no proper noun to correct, and extraction skips it too # (intelligence.extract_scenes), so a correction pass here is pure # spend. Same pure classifier both places, so they cannot disagree. is_chatter, chatter_reason = ( gate_chatter(transcript) if settings.chatter_gate_enabled else (False, None) ) corrected, corrected_segments, not_speech = (None, None, False) if not _ai_flag("transcript_correction_enabled"): logger.info( f"Transcript correction disabled — saving raw transcript for call {call_id}" ) elif is_chatter: logger.info( f"Transcript correction skipped for call {call_id}: chatter ({chatter_reason})" ) else: corrected, corrected_segments, not_speech = await transcript_correction.correct( call_id, transcript, segments, system_id=system_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name, node_id=node_id, ) if corrected_segments: # Raw stays as evidence; the corrected copy is what extraction reads. updates["segments_corrected"] = corrected_segments if not_speech: # The corrector sees what _is_degenerate misses — novel repetition # shapes rather than the two it pattern-matches. Keep the raw text # (it is evidence) but do not let it reach extraction as fact. updates["transcript_not_speech"] = True logger.info( f"Corrector flagged call {call_id} as recogniser noise: {transcript[:80]!r}" ) elif corrected: updates["transcript_corrected"] = corrected try: await fstore.doc_set("calls", call_id, updates) logger.info( f"Transcript saved for call {call_id} via {used.provider}/{used.model} " f"({len(transcript)} chars, {len(segments)} segment(s)" f"{', corrected' if corrected and not not_speech else ''})" ) except Exception as e: logger.warning(f"Could not save transcript for {call_id}: {e}") if not_speech: return None, [] # Hand the corrected copies downstream. extract_scenes prefers numbered # segments over the joined transcript, so returning corrected text with # raw segments would have thrown the correction away on every call with # more than one transmission. return corrected or transcript, corrected_segments or segments return transcript, segments def _download(gcs_uri: str) -> str: """Download the recording to a temp file once; the caller deletes it.""" from google.cloud import storage as gcs from google.oauth2 import service_account without_scheme = gcs_uri[len("gs://"):] bucket_name, blob_path = without_scheme.split("/", 1) if settings.gcp_credentials_path: creds = service_account.Credentials.from_service_account_file( settings.gcp_credentials_path, scopes=["https://www.googleapis.com/auth/cloud-platform"], ) gcs_client = gcs.Client(credentials=creds) else: gcs_client = gcs.Client() blob = gcs_client.bucket(bucket_name).blob(blob_path) suffix = os.path.splitext(blob_path)[1] or ".mp3" with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp: tmp_path = tmp.name try: blob.download_to_filename(tmp_path) except Exception: try: os.unlink(tmp_path) except OSError: pass raise return tmp_path def _segment_confidence(raw_segments: list) -> Optional[float]: """exp(avg_logprob) per segment, weighted by segment length; None if absent.""" total, weight = 0.0, 0.0 for s in raw_segments: lp = getattr(s, "avg_logprob", None) if lp is None: continue w = max(0.01, float(getattr(s, "end", 0) or 0) - float(getattr(s, "start", 0) or 0)) total += math.exp(float(lp)) * w weight += w return round(total / weight, 3) if weight else None def _sync_transcribe_file( tmp_path: str, target: SttTarget, prompt: str, glossary: Optional[list[str]] = None, ) -> dict: """Transcribe one downloaded file on one target. Returns {"text", "segments", "degenerate", "duration", "confidence"}. `degenerate` is True when output was DISCARDED as degenerate, which the caller distinguishes from ordinary silence so it can retry — the same clip can hallucinate on one attempt and transcribe on the next. Raises on any API failure, so the caller can fall back to the next target. """ from app.internal.ai_clients import stt_client # verbose_json (per-segment timestamps + no_speech_prob) is a Whisper # feature: whisper-1 on OpenAI, every whisper-* model on Groq. OpenAI's # gpt-4o-transcribe / gpt-4o-mini-transcribe only support json/text. use_verbose = target.model.startswith("whisper") client = stt_client(target.api_key, target.base_url) with open(tmp_path, "rb") as f: response = client.audio.transcriptions.create( model=target.model, file=f, language="en", prompt=prompt, response_format="verbose_json" if use_verbose else "json", temperature=0, ) if use_verbose: # Filter hallucinated segments. Two sources of hallucination in P25 recordings: # # 1. Trailing silence / static — Whisper fills silence past real content with # sequential radio codes (10-4, 10-5...). Clamped by audio duration. # # 2. Leading silence — OP25 recordings typically have a short silence at the # start before the first PTT press. Whisper sometimes hallucinates filler # words or codes over this silence. Detected via no_speech_prob > 0.8 # (Whisper's own confidence that a segment contains no real speech). duration = getattr(response, "duration", None) audio_duration: float = duration or float("inf") kept = [ s for s in (response.segments or []) if (_seg(s, "text") or "").strip() and (_seg(s, "start") or 0.0) < audio_duration and (_seg(s, "no_speech_prob") or 0.0) < 0.8 ] segments = [ {"start": round(_seg(s, "start") or 0.0, 2), "end": round(_seg(s, "end") or 0.0, 2), "text": _seg(s, "text").strip()} for s in kept ] # Reconstruct text from non-hallucinated segments only so the two stay # in sync. If every segment was filtered, text becomes None which prevents # the intelligence pipeline from running on hallucinated content. text = " ".join(s["text"] for s in segments) or None base = {"duration": duration, "confidence": _segment_confidence([_SegView(s) for s in kept])} if _is_degenerate(text or "", segments, glossary): logger.info(f"Discarded hallucinated transcript from {target.provider}: {(text or '')[:80]!r}") return {"text": None, "segments": [], "degenerate": True, **base} return {"text": text, "segments": segments, "degenerate": False, **base} # json format returns just {"text": "..."} — no segments or timestamps. # Intelligence extraction falls back to treating the whole transcript as one block. text = (response.text or "").strip() or None if _is_degenerate(text or "", [], glossary): logger.info(f"Discarded hallucinated transcript from {target.provider}: {(text or '')[:80]!r}") return {"text": None, "segments": [], "degenerate": True, "duration": None, "confidence": None} return {"text": text, "segments": [], "degenerate": False, "duration": None, "confidence": None} def _seg(s, field: str): """Segment field from an SDK object or a plain dict (OpenAI-compatible servers do not all come back through the same pydantic model).""" if isinstance(s, dict): return s.get(field) return getattr(s, field, None) class _SegView: """Attribute view over a segment, for _segment_confidence.""" def __init__(self, s): self.avg_logprob = _seg(s, "avg_logprob") self.start = _seg(s, "start") self.end = _seg(s, "end")