diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index 8cb0cdf..e46d38a 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -46,6 +46,17 @@ class Settings(BaseSettings): # Verify against https://ai.google.dev/gemini-api/docs/models before changing. corr_cheap_model: str = "gemini-3.6-flash" # was gemini-2.0-flash (shut down) corr_smart_model: str = "gemini-2.5-pro" # was gemini-1.5-pro (shut down) + # Transcript correction (server-26#36). Runs inside transcription, once per + # transcribed call above MIN_WORDS_FOR_CORRECTION, so it is priced like STT + # rather than like the correlation tier — cheap model on purpose. + transcript_correction_enabled: bool = True + transcript_correction_model: str = "gemini-3.6-flash" + # Retry Whisper once when its output is degenerate. The same clip produced a + # 56-word ten-code counting run on one attempt and real speech on the next + # (2026-08-23, call e49ea32c), so a hallucination is a coin-flip rather than + # a property of the audio, and discarding on the first bad roll threw away a + # recoverable transcript. + stt_retry_on_degenerate: bool = True summary_interval_minutes: int = 2 # how often the summary loop runs correlation_window_hours: int = 2 # slow/location path: max hours since last call embedding_similarity_threshold: float = 0.93 # slow-path: requires location corroboration diff --git a/drb-c2-core/app/internal/intelligence.py b/drb-c2-core/app/internal/intelligence.py index 0def62c..26386bd 100644 --- a/drb-c2-core/app/internal/intelligence.py +++ b/drb-c2-core/app/internal/intelligence.py @@ -49,7 +49,6 @@ Response format — a JSON object with a "scenes" array. Each scene: severity: one of "routine" | "minor" | "moderate" | "major" resolved: true if this scene explicitly signals incident closure, false otherwise reassignment: true if a unit is breaking from their current scene to respond to a completely different call — whether dispatch-initiated ("Baker, can you clear and respond to...", "Adam, break from that and go to...") OR unit-initiated ("Show me headed to the vehicle complaint", "Can you show me to that call", a unit going 10-8 and self-requesting a new assignment). False if the unit is reporting in on their current scene, giving a status update, or requesting information about their existing call. - transcript_corrected: corrected text for this scene's transmissions only, or null Rules: - location: prefer intersections > addresses > mile markers > route+town > route alone > town alone. Dispatch-provided addresses take priority over unit-reported positions. Empty string if none. @@ -66,7 +65,6 @@ Rules: - resolved: true only when the scene explicitly signals "Code 4", "all clear", "10-42", "in custody", "patient transported", "fire out", "GOA", "negative contact", "scene clear". - cleared_units: only include units that explicitly stated their own back-in-service status in this recording (e.g. "Unit 7, 10-8", "Baker-1 available", "E-14 back in service", or the department ten-code for available/back-in-service listed above). Silence or absence of a unit is NOT clearance. A scene-wide Code 4 belongs in resolved=true, not here — cleared_units is for individual unit availability signals only. - reassignment: only true when a unit is explicitly being pulled to a completely new call or location. A unit going en route to their first dispatch is NOT a reassignment. Routine status updates, acknowledgements, and scene updates are NOT reassignments. -- transcript_corrected: fix only clear STT/vocoder errors (e.g. "Several" → "10-4", misheard street names, garbled unit IDs). Keep all radio language as-is — do NOT decode codes into plain English. Return null if accurate. System: {system_id} Talkgroup: {talkgroup_name} diff --git a/drb-c2-core/app/internal/transcript_correction.py b/drb-c2-core/app/internal/transcript_correction.py new file mode 100644 index 0000000..99af5a1 --- /dev/null +++ b/drb-c2-core/app/internal/transcript_correction.py @@ -0,0 +1,277 @@ +""" +Transcript correction — the second opinion on what was said. + +Whisper hears a P25 vocoder through a narrowband channel and guesses at proper +nouns it has no reason to know: street names, business names, unit call signs. +It guesses confidently, so the output reads like speech and is wrong in exactly +the places that matter downstream — "Cool Parts, Illinois" and "Shout out to +Optum" both became incident locations. + +Correction used to be a line in intelligence.py's EXTRACTION_PROMPT, which put +it in the wrong place twice over (server-26#36): the same model call that +extracted units, location and severity emitted the correction *afterwards*, so +extraction reasoned over uncorrected text; and it sat behind +`correlation_enabled`, so during a cost-controlled STT-only window nothing was +ever corrected at all. It belongs here, between transcription and everything +that consumes a transcript. + +WHY A SEPARATE PASS AND NOT A WHISPER PROMPT: Whisper treats its prompt as +preceding transcript text and will happily continue a pattern it finds there — +an enumerated ten-code prompt made it emit "10-4. 10-5. 10-6. …" over silence +(see transcription.py). Vocabulary can never be a transcription prior. A +corrector that receives an already-produced transcript plus a reference list has +no series to extend; it can only substitute what it was given. + +SCOPE RESOLUTION: reference data is merged from the talkgroup and the system, +**talkgroup first**. The specific beats the general — a system spanning several +counties may have one talkgroup covering a single municipality, and that +municipality's streets must not be buried under a county-wide list. A +single-municipality system is the degenerate case: populate the system level and +every talkgroup inherits it. +""" + +import asyncio +import json +from typing import Any, Optional + +from app.config import settings +from app.internal import firestore as fstore +from app.internal.logger import logger + +# A transcript this short has no proper nouns to get wrong — "10-4.", "6-2, +# stand by." — and 9 of 29 calls in the 2026-08-23 sample sat at or under this. +# Skipping them is most of the cost saving for none of the value. +MIN_WORDS_FOR_CORRECTION = 4 + +_PROMPT = """You are correcting a police/fire radio transcript produced by an automatic speech recogniser. + +The recogniser hears a low-bitrate vocoded radio channel. It reliably mishears proper nouns — street names, business names, town names, unit call signs — and substitutes common words that sound similar. Your job is to put back what was almost certainly said. + +{context_block} +Rules: +- Change ONLY what is likely a mishearing. If a phrase is already plausible radio traffic, leave it exactly as it is. +- Prefer a name from the reference lists above when the transcript contains something that sounds like it. That is the entire point of this pass. +- NEVER add information. No new sentences, no invented units, no addresses that are not implied by the audio's own words. +- Keep radio language as radio language. Do NOT expand ten-codes or signals into plain English: "10-4" stays "10-4". +- Keep the speaker's structure and order. This is not a rewrite or a summary. +- If the text is clearly not speech at all — a counting run like "10-11. 10-12. 10-13.", or one phrase repeating many times over static — set not_speech to true. + +Return JSON: + corrected: the corrected transcript, or null if nothing needed changing + segments: REQUIRED when numbered transmissions are given below — the corrected + text for each one, as an array of exactly the same length and order. + Never merge, split, reorder or drop a transmission; an unchanged one + is returned verbatim. Omit this field entirely when no transmissions + are numbered. + not_speech: true if this is recogniser noise rather than a transmission + changed: list of ["heard" -> "corrected"] pairs you applied, for audit + +{transcript}""" + + +def _render_input(text: str, segments: Optional[list[dict]]) -> str: + """Numbered transmissions when we have them, so corrections stay aligned.""" + if segments and len(segments) > 1: + lines = [f"{i + 1}. {s.get('text', '')}" for i, s in enumerate(segments)] + body = "\n".join(lines) + return f"Transmissions ({len(segments)}):\n{body}" + return f"Transcript:\n{text}" + + +def _dedupe(items: list[str]) -> list[str]: + """Preserve order, drop case-insensitive duplicates.""" + seen: set[str] = set() + out: list[str] = [] + for item in items: + key = (item or "").strip().lower() + if key and key not in seen: + seen.add(key) + out.append(item.strip()) + return out + + +def _talkgroup_entry(system_doc: dict, talkgroup_id: Optional[int]) -> dict: + """The config.talkgroups[] entry for this talkgroup, or {}.""" + if talkgroup_id is None: + return {} + try: + wanted = int(talkgroup_id) + except (TypeError, ValueError): + return {} + for tg in (system_doc.get("config") or {}).get("talkgroups", []) or []: + try: + if int(tg.get("id", -1)) == wanted: + return tg + except (TypeError, ValueError): + continue + return {} + + +def _area_lines(area: dict, label: str) -> list[str]: + """Render an area_context dict as prompt lines. Empty when nothing is set.""" + if not area: + return [] + parts: list[str] = [] + for key, heading in ( + ("municipality", "Municipality"), + ("county", "County"), + ("roads", "Roads and highways"), + ("landmarks", "Landmarks, businesses and facilities"), + ): + value = area.get(key) + if not value: + continue + if isinstance(value, list): + value = ", ".join(str(v) for v in value if v) + if value: + parts.append(f" {heading}: {value}") + return [f"{label}:"] + parts if parts else [] + + +async def resolve_context(system_id: Optional[str], talkgroup_id: Optional[int]) -> dict: + """ + Merge the reference data a corrector needs, talkgroup ahead of system. + + Returns {"vocabulary": [...], "ten_codes": {...}, "area_lines": [...]}. + Empty everywhere is legitimate — a system nobody has configured yet. + """ + empty: dict[str, Any] = {"vocabulary": [], "ten_codes": {}, "area_lines": []} + if not system_id: + return empty + + system_doc = await fstore.doc_get_cached("systems", system_id) + if not system_doc: + return empty + + tg = _talkgroup_entry(system_doc, talkgroup_id) + + # Talkgroup terms first so they survive any downstream truncation. + vocabulary = _dedupe( + list(tg.get("vocabulary") or []) + list(system_doc.get("vocabulary") or []) + ) + + # Ten-codes: system-wide reference, with talkgroup entries overriding a + # code that means something different on this channel. + ten_codes = dict(system_doc.get("ten_codes") or {}) + ten_codes.update(tg.get("ten_codes") or {}) + + area_lines = ( + _area_lines(tg.get("area_context") or {}, "Area covered by this talkgroup") + + _area_lines(system_doc.get("area_context") or {}, "Area covered by this system") + ) + + return {"vocabulary": vocabulary, "ten_codes": ten_codes, "area_lines": area_lines} + + +def build_context_block(context: dict, talkgroup_name: Optional[str]) -> str: + """Render resolved context into the prompt's reference section.""" + lines: list[str] = [] + if talkgroup_name: + lines.append(f"Channel: {talkgroup_name}") + lines.extend(context.get("area_lines") or []) + vocabulary = context.get("vocabulary") or [] + if vocabulary: + lines.append("Known local names and terms: " + ", ".join(vocabulary)) + ten_codes = context.get("ten_codes") or {} + if ten_codes: + rendered = ", ".join(f"{code}={meaning}" for code, meaning in sorted(ten_codes.items())) + lines.append(f"Ten-codes used on this system: {rendered}") + return ("\n".join(lines) + "\n") if lines else "" + + +def _sync_gemini(model_name: str, prompt: str) -> dict: + import google.generativeai as genai # lazy import — only when needed + + genai.configure(api_key=settings.gemini_api_key) + model = genai.GenerativeModel( + model_name, + generation_config={"response_mime_type": "application/json"}, + ) + return json.loads(model.generate_content(prompt).text) + + +async def correct( + call_id: str, + text: str, + segments: Optional[list[dict]] = None, + system_id: Optional[str] = None, + talkgroup_id: Optional[int] = None, + talkgroup_name: Optional[str] = None, +) -> tuple[Optional[str], Optional[list[dict]], bool]: + """ + Second-opinion pass over a transcript. + + Returns (corrected_text, corrected_segments, not_speech). ``None`` for + either correction means "no change" — the corrector found nothing to fix, + could not run, or returned segments that did not line up. Callers keep the + original in that case; correction is an improvement, never a dependency. + + Segments matter as much as the joined text: intelligence.py builds its + extraction prompt from NUMBERED SEGMENTS whenever there is more than one, + so a correction that only fixed the joined transcript would never reach the + model on exactly the multi-transmission calls that carry the most content. + """ + if not settings.gemini_api_key or not settings.transcript_correction_enabled: + return None, None, False + + if len((text or "").split()) < MIN_WORDS_FOR_CORRECTION: + return None, None, False + + context = await resolve_context(system_id, talkgroup_id) + prompt = _PROMPT.format( + context_block=build_context_block(context, talkgroup_name), + transcript=_render_input(text, segments), + ) + + try: + raw = await asyncio.to_thread( + _sync_gemini, settings.transcript_correction_model, prompt + ) + except Exception as e: + # Never fail the transcript over a failed correction — the raw text is + # still worth having. ai_health reporting is the caller's business. + logger.warning(f"Transcript correction failed for call {call_id}: {e}") + return None, None, False + + not_speech = bool(raw.get("not_speech")) + + corrected = raw.get("corrected") + if not isinstance(corrected, str) or not corrected.strip(): + corrected = None + elif corrected.strip() == (text or "").strip(): + corrected = None + + # Segment alignment is non-negotiable: scene extraction maps scenes back to + # transmissions by INDEX (segment_indices), so a returned array of the wrong + # length would silently attribute the wrong audio to a scene. Wrong length, + # wrong type, or any non-string entry and the segments are discarded whole — + # the joined correction still stands. + corrected_segments: Optional[list[dict]] = None + if segments and len(segments) > 1: + returned = raw.get("segments") + if ( + isinstance(returned, list) + and len(returned) == len(segments) + and all(isinstance(x, str) for x in returned) + ): + corrected_segments = [ + {**seg, "text": new.strip() or seg.get("text", "")} + for seg, new in zip(segments, returned) + ] + if all(s["text"] == o.get("text") for s, o in zip(corrected_segments, segments)): + corrected_segments = None + elif returned is not None: + logger.warning( + f"Transcript correction for call {call_id} returned " + f"{len(returned) if isinstance(returned, list) else type(returned).__name__} " + f"segment(s) against {len(segments)} — discarding segment corrections" + ) + + if corrected or corrected_segments or not_speech: + changed = raw.get("changed") or [] + logger.info( + f"Transcript correction ({settings.transcript_correction_model}): call {call_id} " + f"not_speech={not_speech} segments={'yes' if corrected_segments else 'no'} " + f"changes={changed if isinstance(changed, list) else '?'}" + ) + return corrected, corrected_segments, not_speech diff --git a/drb-c2-core/app/internal/transcription.py b/drb-c2-core/app/internal/transcription.py index b71ed5f..d26354b 100644 --- a/drb-c2-core/app/internal/transcription.py +++ b/drb-c2-core/app/internal/transcription.py @@ -12,6 +12,7 @@ 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.config import settings # Whisper treats `prompt` as preceding transcript text, not instructions. @@ -154,6 +155,7 @@ async def transcribe_call( gcs_uri: str, talkgroup_name: Optional[str] = None, system_id: Optional[str] = None, + talkgroup_id: Optional[int] = None, ) -> tuple[Optional[str], list[dict]]: """ Transcribe audio at the given GCS URI and store the result in Firestore. @@ -166,9 +168,23 @@ async def transcribe_call( return None, [] try: - transcript, segments = await asyncio.to_thread( + transcript, segments, degenerate = await asyncio.to_thread( _sync_transcribe, gcs_uri, talkgroup_name ) + # 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 degenerate and settings.stt_retry_on_degenerate: + logger.info(f"Retrying transcription for call {call_id} after degenerate output") + transcript, segments, degenerate = await asyncio.to_thread( + _sync_transcribe, gcs_uri, talkgroup_name + ) + if degenerate: + logger.warning( + f"Transcription for call {call_id} was degenerate twice — giving up" + ) except Exception as e: await _log_transcribe_failure(call_id, e) return None, [] @@ -183,23 +199,63 @@ async def transcribe_call( updates: dict = {"transcript": transcript} 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. + 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, + ) + 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} " - f"({len(transcript)} chars, {len(segments)} segment(s))" + 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 _sync_transcribe( gcs_uri: str, talkgroup_name: Optional[str] = None, -) -> tuple[Optional[str], list[dict]]: - """Download audio from GCS and transcribe with OpenAI Whisper.""" +) -> tuple[Optional[str], list[dict], bool]: + """Download audio from GCS and transcribe with OpenAI Whisper. + + Third element 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. + """ from google.cloud import storage as gcs from google.oauth2 import service_account from openai import OpenAI @@ -210,7 +266,7 @@ def _sync_transcribe( # Tuple, not a bare None: the caller unpacks two values, so returning # None here raised a TypeError that surfaced as a misleading # "Transcription failed" instead of the real missing-key warning. - return None, [] + return None, [], False without_scheme = gcs_uri[len("gs://"):] bucket_name, blob_path = without_scheme.split("/", 1) @@ -281,16 +337,16 @@ def _sync_transcribe( text = " ".join(s["text"] for s in segments) or None if _is_degenerate(text or "", segments): logger.info(f"Discarded hallucinated transcript for {gcs_uri}: {(text or '')[:80]!r}") - return None, [] - return text, segments + return None, [], True + return text, segments, False else: # 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 "", []): logger.info(f"Discarded hallucinated transcript for {gcs_uri}: {(text or '')[:80]!r}") - return None, [] - return text, [] + return None, [], True + return text, [], False finally: try: os.unlink(tmp_path) diff --git a/drb-c2-core/app/models.py b/drb-c2-core/app/models.py index fca5ab9..99859f1 100644 --- a/drb-c2-core/app/models.py +++ b/drb-c2-core/app/models.py @@ -85,6 +85,11 @@ class SystemRecord(BaseModel): type: str # P25 / DMR / NBFM config: Dict[str, Any] = {} # OP25-compatible config blob ten_codes: Dict[str, str] = {} # {"10-10": "Commercial Alarm", ...} + # Ground truth about the area this system covers, fed to the transcript + # corrector (server-26#36): {municipality, county, roads[], landmarks[]}. + # Per-talkgroup overrides live inside config.talkgroups[] and rank ABOVE + # this, so a multi-county system can narrow per channel. + area_context: Dict[str, Any] = {} class SystemCreate(BaseModel): @@ -92,6 +97,7 @@ class SystemCreate(BaseModel): type: str config: Dict[str, Any] = {} ten_codes: Dict[str, str] = {} + area_context: Dict[str, Any] = {} # --------------------------------------------------------------------------- diff --git a/drb-c2-core/app/routers/systems.py b/drb-c2-core/app/routers/systems.py index d7fa99a..602fa5e 100644 --- a/drb-c2-core/app/routers/systems.py +++ b/drb-c2-core/app/routers/systems.py @@ -1,7 +1,7 @@ import uuid from fastapi import APIRouter, HTTPException, Depends, Query from pydantic import BaseModel -from typing import Dict, Optional +from typing import Dict, List, Optional from app.models import SystemCreate, SystemRecord from app.internal import firestore as fstore from app.internal.auth import ( @@ -23,6 +23,14 @@ class TenCodesBody(BaseModel): ten_codes: Dict[str, str] +class AreaContextBody(BaseModel): + """Ground truth about the area a system covers — see PUT /{id}/area-context.""" + municipality: Optional[str] = None + county: Optional[str] = None + roads: List[str] = [] + landmarks: List[str] = [] + + class AiFlagsBody(BaseModel): stt_enabled: Optional[bool] = None correlation_enabled: Optional[bool] = None @@ -64,8 +72,14 @@ async def update_system(system_id: str, body: SystemCreate, _: dict = Depends(re existing = await fstore.doc_get("systems", system_id) if not existing: raise HTTPException(404, f"System '{system_id}' not found.") - await fstore.doc_update("systems", system_id, body.model_dump()) - return {**existing, **body.model_dump()} + # exclude_unset, or every field the caller omitted gets written as its + # default and silently erases what was there. The systems page PUTs only + # {name, type, config}, so a plain model_dump() wiped ten_codes on every + # save — they are edited through PUT /{id}/ten-codes and were never in this + # payload. area_context (server-26#36) would have been the second casualty. + patch = body.model_dump(exclude_unset=True) + await fstore.doc_update("systems", system_id, patch) + return {**existing, **patch} @router.delete("/{system_id}", status_code=204) @@ -129,6 +143,42 @@ async def update_ten_codes( return {"ok": True, "ten_codes": body.ten_codes} +# ── Area context ────────────────────────────────────────────────────────────── + +@router.get("/{system_id}/area-context") +async def get_area_context(system_id: str, _: dict = Depends(require_admin_token)): + system = await fstore.doc_get("systems", system_id) + if not system: + raise HTTPException(404, f"System '{system_id}' not found.") + return {"area_context": system.get("area_context") or {}} + + +@router.put("/{system_id}/area-context") +async def update_area_context( + system_id: str, + body: AreaContextBody, + _: dict = Depends(require_admin_token), +): + """ + Replace the system-wide area context used by the transcript corrector. + + Ground truth about where this system operates — municipality, county, the + roads and landmarks whose names Whisper mangles. Per-talkgroup overrides + live inside config.talkgroups[] and rank ABOVE this (server-26#36), so a + multi-county system narrows per channel rather than replacing this wholesale. + + Its own route rather than a field on PUT /systems/{id} for the same reason + ten-codes has one: the systems form does not carry it, and folding it into + that payload is how ten_codes kept getting wiped. + """ + existing = await fstore.doc_get("systems", system_id) + if not existing: + raise HTTPException(404, f"System '{system_id}' not found.") + area = body.model_dump(exclude_none=True) + await fstore.doc_update("systems", system_id, {"area_context": area}) + return {"ok": True, "area_context": area} + + # ── Vocabulary endpoints ─────────────────────────────────────────────────────── @router.get("/{system_id}/vocabulary") diff --git a/drb-c2-core/app/routers/upload.py b/drb-c2-core/app/routers/upload.py index dff626d..b0de9a3 100644 --- a/drb-c2-core/app/routers/upload.py +++ b/drb-c2-core/app/routers/upload.py @@ -286,7 +286,8 @@ async def _run_intelligence_pipeline( if gcs_uri: if _flag("stt_enabled"): transcript, segments = await transcription.transcribe_call( - call_id, gcs_uri, talkgroup_name, system_id=system_id + call_id, gcs_uri, talkgroup_name, + system_id=system_id, talkgroup_id=talkgroup_id, ) else: scope = "globally" if not flags["stt_enabled"] else f"system {system_id}" diff --git a/drb-c2-core/tests/test_transcript_correction.py b/drb-c2-core/tests/test_transcript_correction.py new file mode 100644 index 0000000..3869a03 --- /dev/null +++ b/drb-c2-core/tests/test_transcript_correction.py @@ -0,0 +1,211 @@ +""" +Unit tests for the transcript correction pass (server-26#36). + +Two properties carry real risk and are pinned hardest here: + + * SCOPE RESOLUTION — talkgroup reference data must rank ABOVE system data. + A system spanning several counties can have a talkgroup covering one + municipality, and burying that municipality's streets under a county-wide + list is the failure this whole feature exists to avoid. + + * SEGMENT ALIGNMENT — scene extraction maps scenes to transmissions by index + (segment_indices), so a corrected array of the wrong length would silently + attribute the wrong audio to a scene. Anything but an exact 1:1 match must + be discarded whole. +""" +import pytest +from unittest.mock import AsyncMock, patch + +from app.internal import transcript_correction as tc + +SYSTEM = { + "vocabulary": ["Croton-Harmon", "Metro-North"], + "ten_codes": {"10-4": "acknowledged", "10-13": "officer needs assistance"}, + "area_context": {"county": "Westchester", "roads": ["Route 9", "Saw Mill Parkway"]}, + "config": { + "talkgroups": [ + { + "id": 9048, + "name": "Ossining - Police Dispatch", + "vocabulary": ["Snowden Avenue", "Croton-Harmon"], + "area_context": {"municipality": "Ossining", "landmarks": ["Sing Sing"]}, + }, + {"id": 9600, "name": "Harrison - Police/EMS Dispatch"}, + {"id": 9563, "ten_codes": {"10-4": "on scene"}}, + ] + }, +} + + +def _system(doc=SYSTEM): + return patch.object(tc.fstore, "doc_get_cached", AsyncMock(return_value=doc)) + + +@pytest.fixture(autouse=True) +def _api_key(): + """ + The dev venv has no GEMINI_API_KEY, and correct() returns early without one + — which would make every assertion below pass for the wrong reason. + """ + with patch.object(tc.settings, "gemini_api_key", "test-key"): + yield + + +# ── Scope resolution ──────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_talkgroup_vocabulary_ranks_above_system(): + with _system(): + ctx = await tc.resolve_context("sys-1", 9048) + assert ctx["vocabulary"][0] == "Snowden Avenue", "talkgroup terms must come first" + assert "Metro-North" in ctx["vocabulary"], "system terms are still inherited" + + +@pytest.mark.asyncio +async def test_duplicate_terms_are_not_repeated(): + """Croton-Harmon is on both scopes; it should appear once, at talkgroup rank.""" + with _system(): + ctx = await tc.resolve_context("sys-1", 9048) + assert [t.lower() for t in ctx["vocabulary"]].count("croton-harmon") == 1 + + +@pytest.mark.asyncio +async def test_talkgroup_area_precedes_system_area(): + with _system(): + ctx = await tc.resolve_context("sys-1", 9048) + joined = "\n".join(ctx["area_lines"]) + assert joined.index("Ossining") < joined.index("Westchester") + + +@pytest.mark.asyncio +async def test_talkgroup_without_own_data_inherits_system(): + with _system(): + ctx = await tc.resolve_context("sys-1", 9600) + assert ctx["vocabulary"] == ["Croton-Harmon", "Metro-North"] + assert any("Westchester" in line for line in ctx["area_lines"]) + + +@pytest.mark.asyncio +async def test_talkgroup_ten_code_overrides_system_meaning(): + with _system(): + ctx = await tc.resolve_context("sys-1", 9563) + assert ctx["ten_codes"]["10-4"] == "on scene" + assert ctx["ten_codes"]["10-13"] == "officer needs assistance" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("system_id, tgid", [(None, 9048), ("sys-1", None)]) +async def test_missing_scope_is_not_an_error(system_id, tgid): + with _system(): + ctx = await tc.resolve_context(system_id, tgid) + assert isinstance(ctx["vocabulary"], list) + + +@pytest.mark.asyncio +async def test_unconfigured_system_yields_empty_context(): + with _system(doc=None): + ctx = await tc.resolve_context("sys-1", 9048) + assert ctx == {"vocabulary": [], "ten_codes": {}, "area_lines": []} + + +# ── Correction behaviour ──────────────────────────────────────────────────── + +def _gemini(payload): + return patch.object(tc, "_sync_gemini", lambda model, prompt: payload) + + +SEGS = [{"start": 0.0, "end": 1.0, "text": "Headquarters, 11-9."}, + {"start": 1.0, "end": 2.0, "text": "Shout out to Optum."}, + {"start": 2.0, "end": 3.0, "text": "360 north, back to Rose."}] + + +@pytest.mark.asyncio +async def test_short_transcript_is_never_sent(): + """9 of 29 calls in the sample window were <=3 words. Nothing to correct.""" + with patch.object(tc, "_sync_gemini") as m: + out = await tc.correct("c1", "10-4.", None, system_id="sys-1") + assert out == (None, None, False) + m.assert_not_called() + + +@pytest.mark.asyncio +async def test_disabled_by_setting(): + with patch.object(tc.settings, "transcript_correction_enabled", False), \ + patch.object(tc, "_sync_gemini") as m: + assert await tc.correct("c1", "a b c d e", None) == (None, None, False) + m.assert_not_called() + + +@pytest.mark.asyncio +async def test_segments_corrected_when_lengths_match(): + payload = {"corrected": "Headquarters, 11-9. Show it out to Ossining. 360 north, back to Route 9.", + "segments": ["Headquarters, 11-9.", "Show it out to Ossining.", "360 north, back to Route 9."]} + with _system(), _gemini(payload): + text, segs, not_speech = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1", talkgroup_id=9048) + assert not_speech is False + assert segs is not None and len(segs) == 3 + assert segs[1]["text"] == "Show it out to Ossining." + assert segs[1]["start"] == 1.0, "timing must survive correction untouched" + + +@pytest.mark.asyncio +async def test_wrong_segment_count_is_discarded_whole(): + """A short array would silently misattribute audio to the wrong scene.""" + payload = {"corrected": "fine", "segments": ["only", "two"]} + with _system(), _gemini(payload): + text, segs, _ = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1") + assert segs is None + assert text == "fine", "the joined correction still stands" + + +@pytest.mark.asyncio +async def test_non_string_segment_entries_are_discarded(): + payload = {"corrected": None, "segments": ["ok", 42, "ok"]} + with _system(), _gemini(payload): + _, segs, _ = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1") + assert segs is None + + +@pytest.mark.asyncio +async def test_unchanged_segments_report_no_correction(): + payload = {"corrected": None, "segments": [s["text"] for s in SEGS]} + with _system(), _gemini(payload): + text, segs, _ = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1") + assert (text, segs) == (None, None) + + +@pytest.mark.asyncio +async def test_echoed_transcript_counts_as_no_change(): + with _system(), _gemini({"corrected": " x y z w "}): + text, _, _ = await tc.correct("c1", "x y z w", None, system_id="sys-1") + assert text is None + + +@pytest.mark.asyncio +async def test_not_speech_is_surfaced(): + with _system(), _gemini({"corrected": None, "not_speech": True}): + _, _, not_speech = await tc.correct("c1", "10-11. 10-12. 10-13. 10-14.", None, system_id="sys-1") + assert not_speech is True + + +@pytest.mark.asyncio +async def test_model_failure_leaves_the_transcript_alone(): + """Correction is an improvement, never a dependency.""" + def boom(model, prompt): + raise RuntimeError("gemini exploded") + with _system(), patch.object(tc, "_sync_gemini", boom): + assert await tc.correct("c1", "x y z w", SEGS, system_id="sys-1") == (None, None, False) + + +@pytest.mark.asyncio +async def test_reference_data_reaches_the_prompt(): + seen = {} + def capture(model, prompt): + seen["prompt"] = prompt + return {"corrected": None} + with _system(), patch.object(tc, "_sync_gemini", capture): + await tc.correct("c1", "x y z w", None, system_id="sys-1", + talkgroup_id=9048, talkgroup_name="Ossining - Police Dispatch") + p = seen["prompt"] + assert "Snowden Avenue" in p and "Sing Sing" in p and "Ossining - Police Dispatch" in p + assert "10-13=officer needs assistance" in p diff --git a/drb-frontend/app/systems/page.tsx b/drb-frontend/app/systems/page.tsx index 11d728b..995bd53 100644 --- a/drb-frontend/app/systems/page.tsx +++ b/drb-frontend/app/systems/page.tsx @@ -5,7 +5,7 @@ import { useRouter } from "next/navigation"; import { useSystems } from "@/lib/useSystems"; import { c2api } from "@/lib/c2api"; import { useAuth } from "@/components/AuthProvider"; -import type { SystemRecord, VocabularyPendingTerm } from "@/lib/types"; +import type { AreaContext, SystemRecord, VocabularyPendingTerm } from "@/lib/types"; // ── P25 structured config types ─────────────────────────────────────────────── @@ -13,6 +13,27 @@ interface TalkgroupEntry { id: string; name: string; tag: string; + // Local knowledge for the transcript corrector (server-26#36). Optional on + // every talkgroup: unset means "inherit the system's", which is the whole + // point of the scope rule — talkgroup narrows, it does not replace. + vocabulary?: string[]; + area_context?: AreaContext; +} + +/** Comma-separated text field <-> string[], the shape the API stores. */ +const listToText = (v?: string[]) => (v ?? []).join(", "); +const textToList = (v: string) => + v.split(",").map((x) => x.trim()).filter(Boolean); + +/** Drop an area_context whose every field is blank, so we don't store noise. */ +function cleanArea(a?: AreaContext): AreaContext | undefined { + if (!a) return undefined; + const out: AreaContext = {}; + if (a.municipality?.trim()) out.municipality = a.municipality.trim(); + if (a.county?.trim()) out.county = a.county.trim(); + if (a.roads?.length) out.roads = a.roads; + if (a.landmarks?.length) out.landmarks = a.landmarks; + return Object.keys(out).length ? out : undefined; } interface P25Config { @@ -47,10 +68,15 @@ function recordToP25Config(c: Record): P25Config { ? (c.voice_channels as number[]).join(", ") : "", talkgroups: Array.isArray(c.talkgroups) - ? (c.talkgroups as Array<{ id: number; name: string; tag: string }>).map((tg) => ({ + ? (c.talkgroups as Array<{ + id: number; name: string; tag: string; + vocabulary?: string[]; area_context?: AreaContext; + }>).map((tg) => ({ id: String(tg.id), name: tg.name, tag: tg.tag ?? "other", + vocabulary: tg.vocabulary ?? [], + area_context: tg.area_context ?? {}, })) : [], }; @@ -70,7 +96,19 @@ function p25ConfigToRecord(p: P25Config): Record { voice_channels: parseFreqs(p.voice_channels), talkgroups: p.talkgroups .filter((tg) => tg.id && tg.name) - .map((tg) => ({ id: parseInt(tg.id, 10), name: tg.name, tag: tg.tag })), + .map((tg) => { + const area = cleanArea(tg.area_context); + const vocab = (tg.vocabulary ?? []).filter(Boolean); + return { + id: parseInt(tg.id, 10), + name: tg.name, + tag: tg.tag, + // Omitted rather than written empty: an absent key is what + // resolve_context() reads as "inherit from the system". + ...(vocab.length ? { vocabulary: vocab } : {}), + ...(area ? { area_context: area } : {}), + }; + }), }; } @@ -316,6 +354,36 @@ function RRImportModal({ ); } +/** True when this talkgroup overrides anything, so the row can say so. */ +function hasLocalKnowledge(tg: TalkgroupEntry): boolean { + return Boolean((tg.vocabulary ?? []).length || cleanArea(tg.area_context)); +} + +/** One labelled text input in the local-knowledge grid. */ +function LocalKnowledgeField({ + label, + value, + onChange, + placeholder, +}: { + label: string; + value: string; + onChange: (v: string) => void; + placeholder?: string; +}) { + return ( + + ); +} + // ── Talkgroup table editor ──────────────────────────────────────────────────── function TalkgroupEditor({ @@ -329,12 +397,34 @@ function TalkgroupEditor({ const [pasteText, setPasteText] = useState(""); const [rrSystem, setRrSystem] = useState(null); const [rrError, setRrError] = useState(null); + const [expanded, setExpanded] = useState(null); const rrInputRef = useRef(null); function addRow() { onChange([...talkgroups, { id: "", name: "", tag: "other" }]); } + function updateArea(i: number, field: "municipality" | "county", value: string) { + const updated = [...talkgroups]; + updated[i] = { ...updated[i], area_context: { ...updated[i].area_context, [field]: value } }; + onChange(updated); + } + + function updateAreaList(i: number, field: "roads" | "landmarks", value: string) { + const updated = [...talkgroups]; + updated[i] = { + ...updated[i], + area_context: { ...updated[i].area_context, [field]: textToList(value) }, + }; + onChange(updated); + } + + function updateRowList(i: number, field: "vocabulary", value: string) { + const updated = [...talkgroups]; + updated[i] = { ...updated[i], [field]: textToList(value) }; + onChange(updated); + } + function removeRow(i: number) { onChange(talkgroups.filter((_, idx) => idx !== i)); } @@ -478,7 +568,8 @@ function TalkgroupEditor({ {talkgroups.map((tg, i) => ( - + + + + {/* Local knowledge, collapsed by default: a system can carry 125 + talkgroups and most inherit the system's context rather than + override it. */} + {expanded === i && ( + + +

+ Given to the transcript corrector for this talkgroup only, ranked + above the system's own list. + Leave blank to inherit the system's. +

+
+ updateArea(i, "municipality", v)} + placeholder="Ossining" + /> + updateArea(i, "county", v)} + placeholder="Westchester" + /> + updateAreaList(i, "roads", v)} + placeholder="Route 9, Croton Ave" + /> + updateAreaList(i, "landmarks", v)} + placeholder="Sing Sing, Phelps Hospital" + /> +
+ updateRowList(i, "vocabulary", v)} + placeholder="Post 4, 11-X-ray, Car 7" + /> +
+
+ + + )} +
))} @@ -996,6 +1146,124 @@ function SourceCallPlayer({ callId }: { callId: string }) { ); } +// ── Area context panel ──────────────────────────────────────────────────────── + +/** + * System-wide ground truth for the transcript corrector (server-26#36). + * + * This is the fallback every talkgroup inherits. A talkgroup that covers one + * municipality inside a multi-county system overrides it from the talkgroup + * table in the edit form, and its entries rank above these. + */ +function AreaContextPanel({ systemId }: { systemId: string }) { + const [open, setOpen] = useState(false); + const [area, setArea] = useState(null); + const [loading, setLoading] = useState(false); + const [saving, setSaving] = useState(false); + const [error, setError] = useState(null); + + async function toggle() { + const next = !open; + setOpen(next); + if (!next || area !== null) return; + setLoading(true); + try { + const data = await c2api.getAreaContext(systemId); + setArea(data.area_context ?? {}); + } catch (e) { + setError(String(e)); + } finally { + setLoading(false); + } + } + + async function save() { + if (!area) return; + setSaving(true); + setError(null); + try { + await c2api.updateAreaContext(systemId, area); + } catch (e) { + setError(String(e)); + } finally { + setSaving(false); + } + } + + const filled = area + ? [area.municipality, area.county, area.roads?.length, area.landmarks?.length].filter(Boolean).length + : 0; + + return ( +
+ + + {open && ( +
+ {loading &&

Loading…

} + {area && ( + <> +

+ Given to the transcript corrector for every talkgroup on this system. + Whisper mishears local names constantly — naming them here is what lets + them be put back. Individual talkgroups can narrow this in the edit form. +

+
+ setArea({ ...area, municipality: v })} + placeholder="Ossining" + /> + setArea({ ...area, county: v })} + placeholder="Westchester" + /> + setArea({ ...area, roads: textToList(v) })} + placeholder="Route 9, Saw Mill Parkway" + /> + setArea({ ...area, landmarks: textToList(v) })} + placeholder="Phelps Hospital, Metro-North station" + /> +
+ + + )} + {error &&

{error}

} +
+ )} +
+ ); +} + // ── Vocabulary panel ────────────────────────────────────────────────────────── function VocabularyPanel({ systemId }: { systemId: string }) { @@ -1291,6 +1559,7 @@ export default function SystemsPage() { + ); diff --git a/drb-frontend/lib/c2api.ts b/drb-frontend/lib/c2api.ts index 5259c33..4f49b80 100644 --- a/drb-frontend/lib/c2api.ts +++ b/drb-frontend/lib/c2api.ts @@ -1,4 +1,5 @@ import { auth } from "@/lib/firebase"; +import type { AreaContext } from "@/lib/types"; const BASE = process.env.NEXT_PUBLIC_C2_URL ?? "http://localhost:8000"; @@ -145,6 +146,17 @@ export const c2api = { updateTenCodes: (systemId: string, ten_codes: Record) => request(`/systems/${systemId}/ten-codes`, { method: "PUT", body: JSON.stringify({ ten_codes }) }), + // Area context — ground truth for the transcript corrector (server-26#36). + // Its own routes rather than fields on updateSystem(), which sends only + // {name, type, config} and would otherwise wipe them on every save. + getAreaContext: (systemId: string) => + request<{ area_context: AreaContext }>(`/systems/${systemId}/area-context`), + updateAreaContext: (systemId: string, area: AreaContext) => + request<{ ok: boolean; area_context: AreaContext }>( + `/systems/${systemId}/area-context`, + { method: "PUT", body: JSON.stringify(area) }, + ), + // Vocabulary getVocabulary: (systemId: string) => request<{ vocabulary: string[]; vocabulary_pending: { term: string; source: "induction" | "correction"; added_at: string }[]; vocabulary_bootstrapped: boolean }>( diff --git a/drb-frontend/lib/types.ts b/drb-frontend/lib/types.ts index b352295..7d8d237 100644 --- a/drb-frontend/lib/types.ts +++ b/drb-frontend/lib/types.ts @@ -226,3 +226,20 @@ export interface AlertEvent { triggered_at: string; acknowledged: boolean; } + +/** + * Ground truth about the area a system or talkgroup covers, handed to the + * transcript corrector so it can recognise local street, town and business + * names that Whisper mangles (server-26#36). + * + * Set on the system, and optionally narrowed per talkgroup inside + * `config.talkgroups[]` — the talkgroup's entries rank ABOVE the system's, so + * a multi-county system can be specific per channel without its wider list + * burying the detail. + */ +export interface AreaContext { + municipality?: string; + county?: string; + roads?: string[]; + landmarks?: string[]; +}