Logan Cusano
97013e1505
Two independent sources of garbage in the AI pipeline, both visible in the 2026-08-16 correlation dump. 1. Hallucinated transcripts. The Whisper prompt opened with an enumerated run of ten-codes: 10-4, 10-23, 10-20, 10-97 and so on. Whisper treats prompt text as preceding transcript, so on noisy or silent audio it continued the series, emitting transcripts that count upward from 10-4 to 10-99. The existing no_speech_prob filter could not catch these: the model is highly confident in text it invented by continuing a pattern. The prompt no longer contains a series to extend, and _is_degenerate() rejects the three shapes this failure takes: ascending ten-code runs, one phrase looping, and near-identical segments across a whole recording. Verified against 13 transcripts from production: all four known hallucinations rejected, all nine real ones kept, including terse traffic containing legitimate codes. 2. Duplicate recordings. node-002 and node-PI-2 both cover TG 9048 and both uploaded the same transmissions, ~1.1s apart. Nine pairs appeared in one dump. Each was transcribed, billed and correlated twice, and the resulting incident listed two units where there was one. Canonical selection is by earliest started_at, tie-broken on call_id, NOT by upload order: upload order varies with encode time and network latency, so it would make the authoritative recording non-deterministic. Call documents are created from MQTT call_start before uploads arrive, so both nodes independently reach the same verdict. The loser keeps its audio (it may be the cleaner capture) but is excluded from STT, correlation, the re-correlation sweep and the orphan debug view. Also fixes _sync_transcribe returning a bare None when OPENAI_API_KEY is missing, where the caller unpacks two values. A missing key surfaced as a misleading "Transcription failed" instead of the real warning. Adds tests/test_dedup.py (15 cases). dedup.py reaches Firestore through an injected callable so it stays importable without firebase-admin present.
113 lines
4.6 KiB
Python
113 lines
4.6 KiB
Python
"""
|
|
Cross-node duplicate detection for call recordings.
|
|
|
|
Two edge nodes within range of the same trunked system both decode and upload
|
|
the same transmission. That is the normal case for a distributed network, not
|
|
an error — but without this, one transmission is transcribed twice, billed
|
|
twice, and correlated twice, and the resulting incident shows two "units"
|
|
where there was one.
|
|
|
|
CANONICAL SELECTION IS DELIBERATELY NOT "FIRST UPLOAD WINS". Upload order
|
|
depends on encode time and network latency, so it varies run to run; picking
|
|
by it would make which recording is authoritative non-deterministic. The call
|
|
document is created from the MQTT call_start event *before* the upload
|
|
arrives, so by upload time every node's document for the transmission already
|
|
exists and can be ranked. Canonical is the earliest ``started_at``, breaking
|
|
ties on ``call_id`` so both nodes independently reach the same verdict.
|
|
|
|
The loser keeps its audio — it is ~60 KB and may be the cleaner capture if the
|
|
winner's node had a weak signal — but is excluded from the AI pipeline.
|
|
"""
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Awaitable, Callable, Optional
|
|
from app.config import settings
|
|
from app.internal.logger import logger
|
|
|
|
# Firestore is reached through an injected callable rather than a module-level
|
|
# import. app.internal.firestore initialises firebase-admin at import time,
|
|
# which needs credentials and the SDK present — so importing it here would make
|
|
# this module unimportable in a unit test. Same reasoning as the deferred
|
|
# import in app/internal/auth.py.
|
|
QueryFn = Callable[[str, list], Awaitable[list[dict]]]
|
|
|
|
|
|
def _parse_dt(value) -> Optional[datetime]:
|
|
"""Firestore hands back Timestamp, datetime, or ISO string depending on writer."""
|
|
if not value:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
|
|
try:
|
|
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
|
|
|
|
|
|
def _is_canonical(call: dict, others: list[dict]) -> bool:
|
|
"""True if `call` is the one recording of this transmission that should be processed."""
|
|
started = _parse_dt(call.get("started_at"))
|
|
call_id = call.get("call_id") or ""
|
|
for other in others:
|
|
other_started = _parse_dt(other.get("started_at"))
|
|
if not other_started or not started:
|
|
continue
|
|
if other_started < started:
|
|
return False
|
|
if other_started == started and (other.get("call_id") or "") < call_id:
|
|
return False
|
|
return True
|
|
|
|
|
|
async def find_duplicate_of(call: dict, query: Optional[QueryFn] = None) -> Optional[str]:
|
|
"""Return the canonical call_id if `call` duplicates another node's recording.
|
|
|
|
Returns None when this call is the canonical one, or when there is nothing
|
|
to compare against (single node in range, or the call lacks the talkgroup
|
|
and system identifiers the match is keyed on).
|
|
"""
|
|
system_id = call.get("system_id")
|
|
talkgroup_id = call.get("talkgroup_id")
|
|
call_id = call.get("call_id")
|
|
started = _parse_dt(call.get("started_at"))
|
|
if not (system_id and talkgroup_id is not None and call_id and started):
|
|
return None
|
|
|
|
if query is None:
|
|
from app.internal import firestore as fstore
|
|
query = fstore.collection_where
|
|
|
|
window = timedelta(seconds=settings.duplicate_window_seconds)
|
|
try:
|
|
# Range-scan on started_at, then filter the rest in Python — Firestore
|
|
# allows a range on only one field per query.
|
|
nearby = await query("calls", [
|
|
("system_id", "==", system_id),
|
|
("started_at", ">=", started - window),
|
|
("started_at", "<=", started + window),
|
|
])
|
|
except Exception as e:
|
|
# Never block an upload on dedup — worst case is the pre-existing
|
|
# behaviour of processing both copies.
|
|
logger.warning(f"Duplicate check failed for call {call_id}: {e}")
|
|
return None
|
|
|
|
matches = [
|
|
c for c in nearby
|
|
if c.get("call_id") != call_id
|
|
and c.get("talkgroup_id") == talkgroup_id
|
|
and c.get("node_id") != call.get("node_id") # same node twice is a real repeat
|
|
and not c.get("duplicate_of") # never point at another duplicate
|
|
]
|
|
if not matches:
|
|
return None
|
|
|
|
if _is_canonical(call, matches):
|
|
return None
|
|
|
|
canonical = min(
|
|
matches,
|
|
key=lambda c: (_parse_dt(c.get("started_at")) or started, c.get("call_id") or ""),
|
|
)
|
|
return canonical.get("call_id")
|