Logan CusanoandClaude Opus 5.5
7ecfca7648
transcript_correction now also discards a correction that drops or changes a bare number the recogniser heard (the live "7" -> "ShotSpotter" case) or grows the transcript past max(4 words, 25%). place_verifier.verify() derives an anchor from the talkgroup's own name plus the receiving node's state when no area is configured; no state means no anchor, and a configured place without an anchor keeps the load-bearing skip. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
309 lines
12 KiB
Python
309 lines
12 KiB
Python
"""
|
|
Place verification — is the name the corrector produced a real place *here*?
|
|
|
|
The transcript corrector (`transcript_correction.py`) substitutes sound-alikes
|
|
against a reference list. It has no way to tell whether its own output is a real
|
|
place, so "Cool Parts, Illinois" and "Shout out to Optum" are exactly as
|
|
acceptable to it as a genuine street name. This module is the check
|
|
(server-26#37).
|
|
|
|
MAPS AS A VERIFIER, NOT AS PROMPT STUFFING. Injecting every road and POI in a
|
|
town would be hundreds of names on a pass that runs on every transcribed call.
|
|
Instead we take the handful of location-shaped nouns a transcript actually
|
|
contains and ask one question per noun:
|
|
|
|
1. Geocode it, bounded by the talkgroup's anchor.
|
|
2. Inside the radius -> accept, done.
|
|
3. Outside, or no result -> look for a sound-alike that DOES resolve inside.
|
|
4. Found one -> correct to it, and propose {term, meaning} to that
|
|
talkgroup's local_knowledge as pending.
|
|
|
|
Cost scales with location nouns, not call volume, and every verified miss
|
|
permanently improves the reference data for that channel.
|
|
|
|
NO ANCHOR MEANS SKIP, NOT ACCEPT. An anchor too wide to discriminate is not
|
|
stored at all (see `area_context`), and without one this module returns
|
|
immediately. A channel nobody described may still get one from its own
|
|
talkgroup name plus the node's state (server-26#160, `_talkgroup_anchor`). A statewide radius would confirm anything inside it, which is worse
|
|
than not checking — it looks like verification and is not.
|
|
|
|
THE FREE TIER RUNS FIRST. A sound-alike among the terms the operator already
|
|
entered costs nothing and is more trustworthy than anything Maps guesses, so
|
|
`local_knowledge` and `vocabulary` are searched before any request goes out.
|
|
"""
|
|
|
|
import re
|
|
from difflib import SequenceMatcher
|
|
from typing import Any, Optional
|
|
|
|
from app.config import settings
|
|
from app.internal import area_context
|
|
from app.internal.logger import logger
|
|
|
|
# Soundex-style consonant classes. Letters that a vocoder + Whisper routinely
|
|
# swap land in the same bucket, so "Optum"/"Ossining" stay far apart while
|
|
# "Snowden"/"Snowdon" collapse together.
|
|
_CLASSES = {
|
|
"b": "1", "f": "1", "p": "1", "v": "1",
|
|
"c": "2", "g": "2", "j": "2", "k": "2", "q": "2", "s": "2", "x": "2", "z": "2",
|
|
"d": "3", "t": "3",
|
|
"l": "4",
|
|
"m": "5", "n": "5",
|
|
"r": "6",
|
|
}
|
|
_DIGRAPHS = (("ph", "f"), ("gh", "g"), ("ck", "k"), ("wr", "r"), ("kn", "n"), ("wh", "w"))
|
|
|
|
|
|
def _norm(text: str) -> str:
|
|
return re.sub(r"[^a-z0-9]+", " ", (text or "").lower()).strip()
|
|
|
|
|
|
def phonetic_key(text: str) -> str:
|
|
"""
|
|
Consonant-class skeleton of a name. Vowels drop out; a run of the same class
|
|
collapses unless a vowel separates it.
|
|
"""
|
|
letters = re.sub(r"[^a-z]", "", (text or "").lower())
|
|
for a, b in _DIGRAPHS:
|
|
letters = letters.replace(a, b)
|
|
out: list[str] = []
|
|
prev = ""
|
|
for ch in letters:
|
|
code = _CLASSES.get(ch, "")
|
|
if code and code != prev:
|
|
out.append(code)
|
|
prev = code if ch not in "aeiouyhw" else ""
|
|
return "".join(out)
|
|
|
|
|
|
def sounds_like(heard: str, candidate: str) -> float:
|
|
"""
|
|
0..1 similarity, the better of the phonetic and the literal comparison.
|
|
|
|
Both are needed: Whisper errors are sometimes phonetic ("5 acre" for
|
|
"5-baker") and sometimes near-spellings ("Croton Ave" for "Croton Avenue"),
|
|
and a key comparison alone scores the second one poorly.
|
|
"""
|
|
literal = SequenceMatcher(None, _norm(heard), _norm(candidate)).ratio()
|
|
ka, kb = phonetic_key(heard), phonetic_key(candidate)
|
|
phonetic = SequenceMatcher(None, ka, kb).ratio() if ka and kb else 0.0
|
|
return max(literal, phonetic)
|
|
|
|
|
|
# -- Maps ----------------------------------------------------------------------
|
|
|
|
def _place_suffix(area: dict) -> str:
|
|
parts = [area[f] for f in area_context.PLACE_FIELDS if area.get(f)]
|
|
return ", ".join(parts)
|
|
|
|
|
|
async def _geocode_in_anchor(query: str, anchor: dict) -> Optional[dict]:
|
|
"""Geocode `query` and return its coords only if they land inside the anchor."""
|
|
from app.internal.intelligence import _geocode_location
|
|
|
|
coords = await _geocode_location(query, anchor=anchor)
|
|
return coords
|
|
|
|
|
|
async def _places_soundalike(heard: str, anchor: dict) -> Optional[dict]:
|
|
"""
|
|
Ask Maps for places near the anchor matching the misheard text.
|
|
|
|
Places Text Search does its own fuzzy matching against a biased region, which
|
|
is usually enough — but "usually" is not a standard, so the result still has
|
|
to pass `sounds_like` before it is allowed to rewrite a transcript. Without
|
|
that guard the API happily returns the nearest gas station for any garbage
|
|
string.
|
|
"""
|
|
if not settings.google_maps_api_key:
|
|
return None
|
|
|
|
import httpx
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=5.0) as client:
|
|
r = await client.get(
|
|
"https://maps.googleapis.com/maps/api/place/textsearch/json",
|
|
params={
|
|
"query": heard,
|
|
"location": f"{anchor['lat']},{anchor['lng']}",
|
|
"radius": int(anchor["radius_km"] * 1000),
|
|
"key": settings.google_maps_api_key,
|
|
},
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
except Exception as e:
|
|
logger.warning(f"Place search failed for {heard!r}: {e}")
|
|
return None
|
|
|
|
if data.get("status") not in ("OK", "ZERO_RESULTS"):
|
|
logger.warning(f"Place search for {heard!r} returned {data.get('status')}")
|
|
return None
|
|
|
|
for result in (data.get("results") or [])[:5]:
|
|
name = (result.get("name") or "").strip()
|
|
loc = (result.get("geometry") or {}).get("location") or {}
|
|
if not name or "lat" not in loc:
|
|
continue
|
|
distance = area_context.geo_dist_km(
|
|
anchor["lat"], anchor["lng"], float(loc["lat"]), float(loc["lng"])
|
|
)
|
|
if distance > anchor["radius_km"]:
|
|
continue
|
|
score = sounds_like(heard, name)
|
|
if score >= settings.place_soundalike_min_ratio:
|
|
return {"term": name, "meaning": result.get("formatted_address") or None, "score": score}
|
|
return None
|
|
|
|
|
|
def _known_soundalike(heard: str, area: dict) -> Optional[dict]:
|
|
"""Best sound-alike among terms the operator already entered. Free."""
|
|
best: Optional[dict] = None
|
|
for entry in area.get("local_knowledge") or []:
|
|
term = entry.get("term") or ""
|
|
if not term or _norm(term) == _norm(heard):
|
|
continue
|
|
score = sounds_like(heard, term)
|
|
if score >= settings.place_soundalike_min_ratio and (best is None or score > best["score"]):
|
|
best = {"term": term, "meaning": entry.get("meaning"), "score": score, "known": True}
|
|
return best
|
|
|
|
|
|
async def _talkgroup_anchor(
|
|
talkgroup_name: Optional[str], node_id: Optional[str],
|
|
) -> tuple[Optional[dict], str]:
|
|
"""
|
|
An anchor derived from the talkgroup's own name plus the receiving node's
|
|
state, or (None, ""). Same width rule as an operator anchor
|
|
(area_context.resolve_anchor drops anything wider than
|
|
area_anchor_max_radius_km, and caches per place for the process).
|
|
|
|
A STATE IS REQUIRED. "Springfield PD" alone geocodes to whichever
|
|
Springfield Google prefers, and an anchor in the wrong half of the
|
|
country would let Maps rewrite real local streets into sound-alikes from
|
|
there (the failure server-26#36 added `state` to prevent). The node's
|
|
reverse-geocoded state is coarse enough to be right even when the node
|
|
sits far from the area the channel covers.
|
|
"""
|
|
from app.internal import firestore as fstore
|
|
from app.internal.intelligence import _get_node_state, _municipality_from_tg
|
|
|
|
muni = _municipality_from_tg(talkgroup_name)
|
|
if not muni or not node_id:
|
|
return None, ""
|
|
node = await fstore.doc_get_cached("nodes", node_id) or {}
|
|
lat, lon = node.get("lat"), node.get("lon")
|
|
if lat is None or lon is None:
|
|
return None, ""
|
|
state = await _get_node_state(node_id, lat, lon)
|
|
if not state:
|
|
return None, ""
|
|
place = {"municipality": muni, "state": state}
|
|
resolved = await area_context.resolve_anchor(place)
|
|
if not resolved:
|
|
return None, ""
|
|
try:
|
|
anchor = {
|
|
"lat": float(resolved["center"]["lat"]),
|
|
"lng": float(resolved["center"]["lng"]),
|
|
"radius_km": float(resolved["radius_km"]),
|
|
}
|
|
except (KeyError, TypeError, ValueError):
|
|
return None, ""
|
|
return anchor, f"{muni}, {state}"
|
|
|
|
|
|
# -- Public --------------------------------------------------------------------
|
|
|
|
def _substitute(text: str, swaps: list[tuple[str, str]]) -> str:
|
|
for heard, replacement in swaps:
|
|
text = re.sub(rf"\b{re.escape(heard)}\b", replacement, text, flags=re.IGNORECASE)
|
|
return text
|
|
|
|
|
|
async def verify(
|
|
call_id: str,
|
|
text: str,
|
|
segments: Optional[list[dict]],
|
|
locations: list[str],
|
|
system_area: Optional[dict],
|
|
tg_area: Optional[dict],
|
|
system_id: Optional[str] = None,
|
|
talkgroup_id: Optional[Any] = None,
|
|
talkgroup_name: Optional[str] = None,
|
|
node_id: Optional[str] = None,
|
|
) -> tuple[Optional[str], Optional[list[dict]]]:
|
|
"""
|
|
Check the corrector's location nouns against the talkgroup's anchor.
|
|
|
|
Returns (text, segments) with verified substitutions applied, or (None, None)
|
|
when nothing changed. Like correction itself, this is an improvement and
|
|
never a dependency: any failure leaves the transcript exactly as it was.
|
|
"""
|
|
if not settings.place_verification_enabled or not locations:
|
|
return None, None
|
|
|
|
area = area_context.effective(system_area, tg_area)
|
|
anchor = area_context.anchor_for(system_area, tg_area)
|
|
suffix = _place_suffix(area)
|
|
if not anchor and not area_context.has_place(area):
|
|
# server-26#160: nobody described this channel, but its own name may
|
|
# (e.g. "New York City - NYPD Citywide 2 Patch") — the same signal
|
|
# #159 trusted in intelligence.py's geocoder. Only when an operator set
|
|
# NOTHING: a place that is set but has no anchor was too wide or is
|
|
# stale, and that skip stays load-bearing.
|
|
anchor, suffix = await _talkgroup_anchor(talkgroup_name, node_id)
|
|
if not anchor:
|
|
return None, None # load-bearing: no anchor means skip, never accept
|
|
swaps: list[tuple[str, str]] = []
|
|
proposals: list[dict] = []
|
|
|
|
for heard in locations[: settings.place_verify_max_per_call]:
|
|
heard = (heard or "").strip()
|
|
if not heard:
|
|
continue
|
|
query = f"{heard}, {suffix}" if suffix else heard
|
|
try:
|
|
if await _geocode_in_anchor(query, anchor):
|
|
continue # real place, in the right area — nothing to do
|
|
candidate = _known_soundalike(heard, area) or await _places_soundalike(heard, anchor)
|
|
except Exception as e:
|
|
logger.warning(f"Place verification failed for {heard!r} on call {call_id}: {e}")
|
|
continue
|
|
if not candidate:
|
|
logger.info(
|
|
f"Place verification: {heard!r} (call {call_id}) does not resolve near the "
|
|
f"anchor and has no sound-alike that does — leaving it alone"
|
|
)
|
|
continue
|
|
swaps.append((heard, candidate["term"]))
|
|
if not candidate.get("known"):
|
|
proposals.append({
|
|
"term": candidate["term"],
|
|
"meaning": candidate.get("meaning"),
|
|
"source": "place_verifier",
|
|
"source_call_ids": [call_id],
|
|
})
|
|
logger.info(
|
|
f"Place verification: {heard!r} -> {candidate['term']!r} "
|
|
f"(score {candidate['score']:.2f}, call {call_id})"
|
|
)
|
|
|
|
if not swaps:
|
|
return None, None
|
|
|
|
if proposals and system_id and talkgroup_id is not None:
|
|
try:
|
|
await area_context.add_pending(system_id, talkgroup_id, proposals)
|
|
except Exception as e:
|
|
logger.warning(f"Could not queue verified terms for call {call_id}: {e}")
|
|
|
|
new_text = _substitute(text or "", swaps)
|
|
new_segments = None
|
|
if segments:
|
|
new_segments = [{**s, "text": _substitute(s.get("text", ""), swaps)} for s in segments]
|
|
if all(a["text"] == b.get("text") for a, b in zip(new_segments, segments)):
|
|
new_segments = None
|
|
return (new_text if new_text != (text or "") else None), new_segments
|