diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index e46d38a..06d51f2 100644 --- a/drb-c2-core/app/config.py +++ b/drb-c2-core/app/config.py @@ -57,6 +57,19 @@ class Settings(BaseSettings): # a property of the audio, and discarding on the first bad roll threw away a # recoverable transcript. stt_retry_on_degenerate: bool = True + # Place verification (server-26#37). Checks the corrector's location nouns + # against the talkgroup's own anchor instead of stuffing every road in town + # into the prompt, so cost scales with location nouns rather than call volume. + place_verification_enabled: bool = True + place_verify_max_per_call: int = 3 + # How close a candidate has to sound before it may rewrite a transcript. + # Below this, Places Text Search will confidently hand back the nearest + # business for any garbage string. + place_soundalike_min_ratio: float = 0.6 + # An anchor wider than this is not stored at all. A statewide radius would + # confirm any location inside it, so the check would rubber-stamp everything + # while appearing to work — absent anchor means SKIP, never "accept anything". + area_anchor_max_radius_km: float = 60.0 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/area_context.py b/drb-c2-core/app/internal/area_context.py new file mode 100644 index 0000000..67cc4bc --- /dev/null +++ b/drb-c2-core/app/internal/area_context.py @@ -0,0 +1,544 @@ +""" +Area context — the ground truth an operator sets about where a channel operates. + +One shape, used at two scopes (server-26#36): + + area_context: { + municipality?, county?, state?, + center?, radius_km?, resolved_from?, resolved_at?, # backend-written + local_knowledge?: [ { term, meaning } ] + } + +WHY EVERY FIELD IS NULLABLE. The system level is only meaningful when it is true +of *every* talkgroup on that system. White Plains PD — it is, so an operator +fills it once and every talkgroup inherits. A statewide Colorado system — it is +not, so they leave it null and fill each talkgroup. Which level someone fills IS +their declaration of how homogeneous the system is, which is what lets one +schema serve both without a `system_type` flag to get out of sync. + +WHY `local_knowledge` REPLACED `roads[]`/`landmarks[]`. Radio traffic references +intersections, schools, housing developments, rail stations and nicknames ("the +flats"), none of which fit two lists. And a bare term is half the information: +`11-X-ray` tells a corrector nothing, `11-X-ray — MTA PD patrol unit` is what +lets it recognise the sound. + +WHY THE ANCHOR CAN BE ABSENT ON PURPOSE. `center`/`radius_km` exist so a +geocoded place name can be sanity-checked against the area the channel actually +covers (server-26#37). If municipality/county/state only resolve to something as +wide as a state, that check would confirm anything inside it while appearing to +work — worse than useless. So an anchor wider than +`settings.area_anchor_max_radius_km` is not written at all, and an absent anchor +means SKIP THE CHECK, never "accept anything". + +THE CLIENT DOES NOT WRITE THE DERIVED FIELDS. `center`, `radius_km`, +`resolved_from` and `resolved_at` are computed here and merged in by the server. +Taking them from the request body is the same defect as the `ten_codes` wipe +fixed in 58efdbd: the frontend does not decide what is in a system document. +""" + +import asyncio +import math +from datetime import datetime, timezone +from typing import Any, Optional + +from app.config import settings +from app.internal.logger import logger + +# Fields an operator sets. Anything else in an incoming body is dropped. +CLIENT_FIELDS = ("municipality", "county", "state", "local_knowledge") +# Fields this module owns. Carried forward from the stored document on every +# write, never read from the request. +SERVER_FIELDS = ("center", "radius_km", "resolved_from", "resolved_at") +# The three that identify a place, in the order they are geocoded. +PLACE_FIELDS = ("municipality", "county", "state") + +_anchor_cache: dict[str, Optional[dict]] = {} + + +def geo_dist_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float: + """Haversine distance in km between two lat/lon points.""" + R = 6371.0 + dlat = math.radians(lat2 - lat1) + dlon = math.radians(lon2 - lon1) + a = ( + math.sin(dlat / 2) ** 2 + + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon / 2) ** 2 + ) + return R * 2 * math.asin(math.sqrt(a)) + + +# -- Normalisation ------------------------------------------------------------- + +def normalize_local_knowledge(raw: Any) -> list[dict]: + """ + Coerce whatever arrived into [{term, meaning}], dropping junk. + + Accepts a bare string list too — that is what `roads[]`/`landmarks[]` and the + old flat `vocabulary` look like, and a term with no meaning is still worth + having in the reference list. + """ + if not isinstance(raw, list): + return [] + out: list[dict] = [] + seen: set[str] = set() + for item in raw: + if isinstance(item, str): + term, meaning = item.strip(), None + elif isinstance(item, dict): + term = str(item.get("term") or "").strip() + meaning = str(item.get("meaning") or "").strip() or None + else: + continue + key = term.lower() + if not term or key in seen: + continue + seen.add(key) + out.append({"term": term, "meaning": meaning} if meaning else {"term": term}) + return out + + +def knowledge_of(area: Optional[dict]) -> list[dict]: + """ + This scope's local knowledge, folding the pre-#36 shape forward. + + `roads[]` and `landmarks[]` were the original fields and real systems still + have them stored. Reading them as bare terms means the corrector keeps the + ground truth an operator already entered instead of silently losing it the + day this shipped; they disappear from the document the next time that scope + is saved. + """ + area = area or {} + legacy = list(area.get("roads") or []) + list(area.get("landmarks") or []) + return normalize_local_knowledge(list(area.get("local_knowledge") or []) + legacy) + + +def normalize(raw: Any) -> dict: + """Client-supplied area_context -> the stored shape, server fields excluded.""" + if not isinstance(raw, dict): + return {} + out: dict[str, Any] = {} + for field in PLACE_FIELDS: + value = raw.get(field) + if isinstance(value, str) and value.strip(): + out[field] = value.strip() + knowledge = knowledge_of(raw) + if knowledge: + out["local_knowledge"] = knowledge + return out + + +def merge_server_fields(incoming: dict, existing: Optional[dict]) -> dict: + """Carry the backend-owned anchor forward across a client write.""" + out = dict(incoming) + for field in SERVER_FIELDS: + if existing and existing.get(field) is not None: + out[field] = existing[field] + return out + + +def merge_config(incoming: Any, existing: Optional[dict]) -> Any: + """ + Reconcile a client-sent config blob with what the server already owns. + + The systems form sends `config.talkgroups[]` in full, so writing it verbatim + destroys everything the backend put there — the resolved anchor and the + pending term queue. That is the same defect as the `ten_codes` wipe fixed in + 58efdbd, and the same fix applies: the backend merges its own fields back in + rather than taking dictation from the frontend. + """ + if not isinstance(incoming, dict): + return incoming + incoming_tgs = incoming.get("talkgroups") + if not isinstance(incoming_tgs, list): + return incoming + + by_id: dict[int, dict] = {} + for tg in ((existing or {}).get("talkgroups") or []): + if isinstance(tg, dict): + try: + by_id[int(tg.get("id", -1))] = tg + except (TypeError, ValueError): + continue + + merged: list[Any] = [] + for tg in incoming_tgs: + if not isinstance(tg, dict): + merged.append(tg) + continue + try: + prior = by_id.get(int(tg.get("id", -1))) or {} + except (TypeError, ValueError): + prior = {} + out = dict(tg) + area = normalize(tg.get("area_context")) + prior_area = prior.get("area_context") or {} + if area: + out["area_context"] = merge_server_fields(area, prior_area) + else: + out.pop("area_context", None) + if prior.get(PENDING_KEY): + out[PENDING_KEY] = prior[PENDING_KEY] + merged.append(out) + + return {**incoming, "talkgroups": merged} + + +# -- Scope resolution ---------------------------------------------------------- + +def effective(system_area: Optional[dict], tg_area: Optional[dict]) -> dict: + """ + Merge the two scopes: talkgroup wins where set, system fills the gaps. + + `local_knowledge` concatenates rather than replaces, talkgroup entries first + so they survive any downstream truncation and outrank a system entry for the + same term. A multi-county system whose talkgroup covers one municipality must + not have that municipality's terms buried under a county-wide list. + """ + system_area = system_area or {} + tg_area = tg_area or {} + out: dict[str, Any] = {} + for field in PLACE_FIELDS: + value = tg_area.get(field) or system_area.get(field) + if value: + out[field] = value + knowledge = normalize_local_knowledge(knowledge_of(tg_area) + knowledge_of(system_area)) + if knowledge: + out["local_knowledge"] = knowledge + return out + + +def talkgroup_entry(system_doc: Optional[dict], talkgroup_id: Any) -> dict: + """The `config.talkgroups[]` entry for this talkgroup, or `{}`.""" + if not system_doc or talkgroup_id is None: + return {} + talkgroups = (system_doc.get("config") or {}).get("talkgroups") or [] + idx = _tg_index(talkgroups, talkgroup_id) + return talkgroups[idx] if idx >= 0 else {} + + +def anchor_key(area: Optional[dict]) -> str: + """ + Stable identity of the place an anchor was resolved from. + + Stored as `resolved_from`, which is what makes "did this actually change?" + decidable — so the geocode happens when someone edits a town name, not on + every read or every five minutes. + """ + area = area or {} + return "|".join((area.get(f) or "").strip().lower() for f in PLACE_FIELDS) + + +def has_place(area: Optional[dict]) -> bool: + return bool(anchor_key(area).strip("|")) + + +def anchor_for(system_area: Optional[dict], tg_area: Optional[dict]) -> Optional[dict]: + """ + The anchor to sanity-check geocoded locations against, or None. + + None has one meaning and it is load-bearing: SKIP THE CHECK. It covers an + unconfigured system, an area too wide to discriminate, and a stored anchor + whose `resolved_from` no longer matches the place it was computed for (an + edit landed and the refresh has not run). Accepting a stale or oversized + anchor would rubber-stamp locations while looking like verification. + """ + key = anchor_key(effective(system_area, tg_area)) + for area in (tg_area, system_area): + if not area: + continue + center, radius = area.get("center"), area.get("radius_km") + if area.get("resolved_from") == key and center and radius: + try: + return { + "lat": float(center["lat"]), + "lng": float(center["lng"]), + "radius_km": float(radius), + } + except (KeyError, TypeError, ValueError): + continue + return None + + +# -- Anchor geocoding ---------------------------------------------------------- + +def _query(area: dict) -> str: + return ", ".join(area[f] for f in PLACE_FIELDS if area.get(f)) + + +async def resolve_anchor(area: dict) -> Optional[dict]: + """ + Geocode municipality/county/state into {center, radius_km}, or None. + + The radius comes from the result's own viewport — half its diagonal — so a + village anchors tightly and a county loosely, which is the real difference + we care about. Anything wider than `area_anchor_max_radius_km` is discarded + rather than stored: see the module docstring. + """ + if not has_place(area): + return None + query = _query(area) + if query in _anchor_cache: + return _anchor_cache[query] + if not settings.google_maps_api_key: + logger.warning("GOOGLE_MAPS_API_KEY not set — area anchors cannot be resolved") + return None + + import httpx + + try: + async with httpx.AsyncClient(timeout=5.0) as client: + r = await client.get( + "https://maps.googleapis.com/maps/api/geocode/json", + params={"address": query, "region": "us", "key": settings.google_maps_api_key}, + ) + r.raise_for_status() + data = r.json() + if data.get("status") != "OK" or not data.get("results"): + logger.warning(f"Area anchor: {query!r} did not geocode ({data.get('status')})") + _anchor_cache[query] = None + return None + geometry = data["results"][0].get("geometry") or {} + loc = geometry.get("location") or {} + lat, lng = float(loc["lat"]), float(loc["lng"]) + viewport = geometry.get("viewport") or {} + ne, sw = viewport.get("northeast"), viewport.get("southwest") + if ne and sw: + radius_km = geo_dist_km(sw["lat"], sw["lng"], ne["lat"], ne["lng"]) / 2 + else: + radius_km = settings.geocode_max_km + except Exception as e: + logger.warning(f"Area anchor geocoding failed for {query!r}: {e}") + return None # not cached — a transient failure should be retried + + if radius_km > settings.area_anchor_max_radius_km: + logger.info( + f"Area anchor: {query!r} spans ~{radius_km:.0f}km, wider than " + f"area_anchor_max_radius_km={settings.area_anchor_max_radius_km} — storing no " + f"anchor, so verification skips rather than rubber-stamps" + ) + _anchor_cache[query] = None + return None + + anchor = { + "center": {"lat": lat, "lng": lng}, + "radius_km": round(radius_km, 2), + "resolved_from": anchor_key(area), + "resolved_at": datetime.now(timezone.utc).isoformat(), + } + _anchor_cache[query] = anchor + logger.info(f"Area anchor: {query!r} -> ({lat:.4f}, {lng:.4f}) r={radius_km:.1f}km") + return anchor + + +def _apply(area: dict, anchor: Optional[dict], key: str) -> dict: + """Write (or clear) the derived fields on one scope's area_context.""" + out = {k: v for k, v in area.items() if k not in SERVER_FIELDS} + if anchor: + out.update(anchor) + elif key.strip("|"): + # A place is set but produced no usable anchor. Record that we tried, so + # the next write does not geocode it again for the same answer. + out["resolved_from"] = key + out["resolved_at"] = datetime.now(timezone.utc).isoformat() + return out + + +async def refresh_anchors(system_doc: dict) -> dict: + """ + Recompute anchors for a system and every talkgroup that sets a place. + + Returns a Firestore patch — `{}` when nothing needed resolving. Talkgroups + are refreshed alongside the system because a talkgroup's anchor is derived + from its EFFECTIVE place (its own fields over the system's), so editing the + system's county silently changes what every talkgroup should be anchored to. + + Only scopes whose `resolved_from` no longer matches are geocoded, and the + per-query cache means N talkgroups in one town cost one request. + """ + system_area = dict(system_doc.get("area_context") or {}) + patch: dict[str, Any] = {} + + system_key = anchor_key(system_area) + if system_area.get("resolved_from") != system_key: + anchor = await resolve_anchor(system_area) if has_place(system_area) else None + patch["area_context"] = _apply(system_area, anchor, system_key) + system_area = patch["area_context"] + + config = system_doc.get("config") or {} + talkgroups = config.get("talkgroups") + if not isinstance(talkgroups, list): + return patch + + updated: list[dict] = [] + changed = False + for tg in talkgroups: + if not isinstance(tg, dict): + updated.append(tg) + continue + tg_area = tg.get("area_context") or {} + # No place of its own means it inherits the system's anchor wholesale — + # nothing to store here, and anchor_for() falls back to the system. + if not has_place(tg_area): + if any(tg_area.get(f) is not None for f in SERVER_FIELDS): + tg = {**tg, "area_context": {k: v for k, v in tg_area.items() if k not in SERVER_FIELDS}} + changed = True + updated.append(tg) + continue + key = anchor_key(effective(system_area, tg_area)) + if tg_area.get("resolved_from") == key: + updated.append(tg) + continue + anchor = await resolve_anchor(effective(system_area, tg_area)) + updated.append({**tg, "area_context": _apply(tg_area, anchor, key)}) + changed = True + + if changed: + patch["config"] = {**config, "talkgroups": updated} + return patch + + +# -- Talkgroup-level pending terms --------------------------------------------- +# +# Proposals land on the TALKGROUP and are never promoted to the system +# automatically (server-26#37). The argument is blast radius: a wrong term on a +# talkgroup misleads one channel, the same term at system level misleads every +# channel on that system — including one 400km away on a statewide system, which +# is exactly the context poisoning the scope rule exists to prevent. If a term +# genuinely applies system-wide, carrying it on several talkgroups costs almost +# nothing; auto-promoting a wrong one is expensive to notice. + +PENDING_KEY = "local_knowledge_pending" + + +def _tg_index(talkgroups: list, talkgroup_id: Any) -> int: + try: + wanted = int(talkgroup_id) + except (TypeError, ValueError): + return -1 + for i, tg in enumerate(talkgroups): + if not isinstance(tg, dict): + continue + try: + if int(tg.get("id", -1)) == wanted: + return i + except (TypeError, ValueError): + continue + return -1 + + +def _known_terms(tg: dict, system_doc: dict) -> set[str]: + """Everything this talkgroup already knows, at either scope, plus pending.""" + known = { + e["term"].lower() + for e in effective(system_doc.get("area_context"), tg.get("area_context")) + .get("local_knowledge", []) + } + known |= {str(t).lower() for t in (tg.get("vocabulary") or [])} + known |= {str(t).lower() for t in (system_doc.get("vocabulary") or [])} + known |= {str(p.get("term", "")).lower() for p in (tg.get(PENDING_KEY) or [])} + return known + + +async def add_pending(system_id: str, talkgroup_id: Any, entries: list[dict]) -> int: + """ + Queue proposed {term, meaning} entries on one talkgroup for human review. + + Returns how many were actually queued. Nothing here writes to + `local_knowledge` — approval is a person's decision, always. + """ + from app.internal import firestore as fstore + + if not system_id or talkgroup_id is None or not entries: + return 0 + system_doc = await fstore.doc_get("systems", system_id) + if not system_doc: + return 0 + config = dict(system_doc.get("config") or {}) + talkgroups = list(config.get("talkgroups") or []) + idx = _tg_index(talkgroups, talkgroup_id) + if idx < 0: + return 0 + + tg = dict(talkgroups[idx]) + known = _known_terms(tg, system_doc) + now = datetime.now(timezone.utc).isoformat() + queued: list[dict] = [] + for entry in entries: + term = str(entry.get("term") or "").strip() + if not term or term.lower() in known: + continue + known.add(term.lower()) + queued.append({ + "term": term, + "meaning": entry.get("meaning") or None, + "source": entry.get("source") or "verifier", + "added_at": now, + "source_call_ids": entry.get("source_call_ids") or [], + }) + if not queued: + return 0 + + tg[PENDING_KEY] = list(tg.get(PENDING_KEY) or []) + queued + talkgroups[idx] = tg + config["talkgroups"] = talkgroups + await fstore.doc_update("systems", system_id, {"config": config}) + logger.info( + f"Local knowledge: {len(queued)} term(s) proposed for talkgroup " + f"{talkgroup_id} on system {system_id}: {[q['term'] for q in queued]}" + ) + return len(queued) + + +async def resolve_pending(system_id: str, talkgroup_id: Any, term: str, approve: bool) -> bool: + """Approve a pending term onto the talkgroup, or dismiss it. Never promotes.""" + from app.internal import firestore as fstore + + system_doc = await fstore.doc_get("systems", system_id) + if not system_doc: + return False + config = dict(system_doc.get("config") or {}) + talkgroups = list(config.get("talkgroups") or []) + idx = _tg_index(talkgroups, talkgroup_id) + if idx < 0: + return False + + tg = dict(talkgroups[idx]) + pending = list(tg.get(PENDING_KEY) or []) + match = next((p for p in pending if str(p.get("term", "")).lower() == term.lower()), None) + if match is None: + return False + tg[PENDING_KEY] = [p for p in pending if p is not match] + if approve: + area = dict(tg.get("area_context") or {}) + area["local_knowledge"] = normalize_local_knowledge( + list(area.get("local_knowledge") or []) + + [{"term": match["term"], "meaning": match.get("meaning")}] + ) + tg["area_context"] = area + talkgroups[idx] = tg + config["talkgroups"] = talkgroups + await fstore.doc_update("systems", system_id, {"config": config}) + return True + + +async def refresh_anchors_bg(system_id: str) -> None: + """Fire-and-forget refresh, for callers that must not block on Maps.""" + from app.internal import firestore as fstore + + try: + doc = await fstore.doc_get("systems", system_id) + if not doc: + return + patch = await refresh_anchors(doc) + if patch: + await fstore.doc_update("systems", system_id, patch) + except Exception as e: + logger.warning(f"Area anchor refresh failed for system {system_id}: {e}") + + +def schedule_refresh(system_id: str) -> None: + """Kick a refresh without making the caller wait for the geocoder.""" + try: + asyncio.get_running_loop().create_task(refresh_anchors_bg(system_id)) + except RuntimeError: # no loop (tests, scripts) — nothing to schedule + pass diff --git a/drb-c2-core/app/internal/intelligence.py b/drb-c2-core/app/internal/intelligence.py index 26386bd..282c312 100644 --- a/drb-c2-core/app/internal/intelligence.py +++ b/drb-c2-core/app/internal/intelligence.py @@ -15,6 +15,7 @@ import re from typing import Optional from app.internal.logger import logger from app.internal import firestore as fstore +from app.internal import area_context # Location validity is defined once, by the module that owns the incident's # location/pin invariant. incident_correlator does not import this module, so # this is not a cycle. @@ -225,6 +226,20 @@ async def extract_scenes( node_lat = node_doc.get("lat") node_lon = node_doc.get("lon") + # The talkgroup's own anchor and place, when an operator has described it + # (server-26#36). This is what "where is this channel" should mean; the node + # position below is only the fallback for a system nobody has described. + tg_anchor: Optional[dict] = None + tg_area: dict = {} + if system_id: + system_doc = await fstore.doc_get_cached("systems", system_id) + if system_doc: + system_area = system_doc.get("area_context") or {} + tg_entry = area_context.talkgroup_entry(system_doc, talkgroup_id) + own_area = tg_entry.get("area_context") or {} + tg_area = area_context.effective(system_area, own_area) + tg_anchor = area_context.anchor_for(system_area, own_area) + processed: list[dict] = [] for scene in raw_scenes: tags: list[str] = scene.get("tags") or [] @@ -273,20 +288,27 @@ async def extract_scenes( # Build the most specific query possible: location + municipality + state. # e.g. "High Street" → "High Street, Yorktown, New York" # This prevents generic street names from resolving to wrong-country results. + # + # Prefer the place an operator actually set over the one guessed from + # the talkgroup's name and the node's reverse-geocoded position. A name + # like "Ossining PD" gives a municipality with no state behind it, which + # is how a generic street name ends up resolving in the wrong half of + # the country. location_coords: Optional[dict] = None - if location and node_lat is not None and node_lon is not None: - muni = _municipality_from_tg(talkgroup_name) - state = await _get_node_state(node_id or "", node_lat, node_lon) if node_id else "" - county = _node_county_cache.get(node_id or "") if node_id else "" - parts = [location] - if muni: - parts.append(muni) - if county: - parts.append(county) - if state: - parts.append(state) + if location: + parts = [location] + if tg_area.get("municipality") or tg_area.get("county") or tg_area.get("state"): + parts += [tg_area[f] for f in area_context.PLACE_FIELDS if tg_area.get(f)] + elif node_lat is not None and node_lon is not None: + muni = _municipality_from_tg(talkgroup_name) + state = await _get_node_state(node_id or "", node_lat, node_lon) if node_id else "" + county = _node_county_cache.get(node_id or "") if node_id else "" + parts += [p for p in (muni, county, state) if p] query = ", ".join(parts) - location_coords = await _geocode_location(query, node_lat, node_lon) + if tg_anchor or (node_lat is not None and node_lon is not None): + location_coords = await _geocode_location( + query, node_lat, node_lon, anchor=tg_anchor + ) # Embed this scene's content scene_text = _build_scene_embed_text( @@ -411,12 +433,25 @@ async def _get_node_state(node_id: str, lat: float, lon: float) -> str: async def _geocode_location( - location_str: str, node_lat: float, node_lon: float + location_str: str, + node_lat: Optional[float] = None, + node_lon: Optional[float] = None, + anchor: Optional[dict] = None, ) -> Optional[dict]: """ - Geocode using Google Maps Geocoding API, biased toward the node's area. - Returns {"lat": float, "lng": float} or None if geocoding fails or the - result is farther than geocode_max_km from the node (wrong-jurisdiction guard). + Geocode using Google Maps Geocoding API, biased toward the channel's area. + + Returns {"lat": float, "lng": float}, or None if geocoding fails or the + result lands outside the area this channel covers. + + THE REFERENCE POINT IS THE TALKGROUP, NOT THE NODE (server-26#6 / #37). This + used to reject anything more than geocode_max_km (40km) from the receiving + node, which conflates an antenna with a jurisdiction: a system can span a + county or several, so a node legitimately sits far from the area a talkgroup + covers, and real dispatch locations were being thrown away for it. When the + talkgroup has a resolved anchor, that is the reference and its own radius is + the bound. Distance-from-node stays only as the fallback for a system nobody + has described yet — it was always a stand-in for this. """ import httpx from app.config import settings @@ -425,9 +460,24 @@ async def _geocode_location( logger.warning("GOOGLE_MAPS_API_KEY not set — geocoding disabled") return None + if anchor: + ref_lat, ref_lon = anchor["lat"], anchor["lng"] + max_km = anchor["radius_km"] + # Bias box scaled to the anchor rather than a fixed half-degree, so a + # village biases tightly and a county loosely. + delta = max(max_km / 111.0, 0.05) + ref_label = "anchor" + elif node_lat is not None and node_lon is not None: + ref_lat, ref_lon = node_lat, node_lon + max_km = settings.geocode_max_km + delta = _GEO_DELTA + ref_label = "node" + else: + return None + bounds = ( - f"{node_lat - _GEO_DELTA},{node_lon - _GEO_DELTA}" - f"|{node_lat + _GEO_DELTA},{node_lon + _GEO_DELTA}" + f"{ref_lat - delta},{ref_lon - delta}" + f"|{ref_lat + delta},{ref_lon + delta}" ) params = { "address": location_str, @@ -465,15 +515,18 @@ async def _geocode_location( return None loc = result["geometry"]["location"] lat, lng = float(loc["lat"]), float(loc["lng"]) - dist_km = _geo_dist_km(node_lat, node_lon, lat, lng) - if dist_km > settings.geocode_max_km: + dist_km = _geo_dist_km(ref_lat, ref_lon, lat, lng) + if dist_km > max_km: logger.warning( f"Geocoding rejected '{location_str}' → ({lat:.4f}, {lng:.4f}) " - f"— {dist_km:.1f}km from node exceeds geocode_max_km={settings.geocode_max_km}" + f"— {dist_km:.1f}km from {ref_label} exceeds {max_km:.1f}km" ) return None coords = {"lat": lat, "lng": lng} - logger.info(f"Geocoded '{location_str}' → {coords} ({dist_km:.1f}km from node) [{location_type}]") + logger.info( + f"Geocoded '{location_str}' → {coords} " + f"({dist_km:.1f}km from {ref_label}) [{location_type}]" + ) return coords except Exception as e: logger.warning(f"Geocoding failed for '{location_str}': {e}") diff --git a/drb-c2-core/app/internal/place_verifier.py b/drb-c2-core/app/internal/place_verifier.py new file mode 100644 index 0000000..d83394b --- /dev/null +++ b/drb-c2-core/app/internal/place_verifier.py @@ -0,0 +1,255 @@ +""" +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 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 + + +# -- 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, +) -> 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 + + anchor = area_context.anchor_for(system_area, tg_area) + if not anchor: + return None, None # load-bearing: no anchor means skip, never accept + + area = area_context.effective(system_area, tg_area) + suffix = _place_suffix(area) + 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 diff --git a/drb-c2-core/app/internal/transcript_correction.py b/drb-c2-core/app/internal/transcript_correction.py index 99af5a1..b5d0809 100644 --- a/drb-c2-core/app/internal/transcript_correction.py +++ b/drb-c2-core/app/internal/transcript_correction.py @@ -35,7 +35,9 @@ import json from typing import Any, Optional from app.config import settings +from app.internal import area_context from app.internal import firestore as fstore +from app.internal import place_verifier from app.internal.logger import logger # A transcript this short has no proper nouns to get wrong — "10-4.", "6-2, @@ -65,6 +67,10 @@ Return JSON: are numbered. not_speech: true if this is recogniser noise rather than a transmission changed: list of ["heard" -> "corrected"] pairs you applied, for audit + locations: every place name in your corrected output, exactly as it appears + there — streets, intersections, businesses, schools, towns, + landmarks. Include ones you are unsure of; that is the point. + A unit call sign or a person's name is NOT a location. {transcript}""" @@ -107,35 +113,45 @@ def _talkgroup_entry(system_doc: dict, talkgroup_id: Optional[int]) -> dict: return {} -def _area_lines(area: dict, label: str) -> list[str]: - """Render an area_context dict as prompt lines. Empty when nothing is set.""" +def _area_lines(area: dict) -> list[str]: + """ + Render a merged area_context as prompt lines. Empty when nothing is set. + + One block, not one per scope: by the time this runs the two scopes have + already been merged with talkgroup ahead of system, and showing the model + two competing lists invites it to pick from the wrong one. + """ 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 [] + lines: list[str] = [] + place = ", ".join( + str(area[f]) for f in area_context.PLACE_FIELDS if area.get(f) + ) + if place: + lines.append(f"Area covered by this channel: {place}") + knowledge = area.get("local_knowledge") or [] + if knowledge: + lines.append("Local names heard on this channel:") + lines.extend( + f" {e['term']} — {e['meaning']}" if e.get("meaning") else f" {e['term']}" + for e in knowledge + ) + return lines 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. + Returns {"vocabulary", "ten_codes", "area_lines", "area", "system_area", + "tg_area"}. Empty everywhere is legitimate — a system nobody has configured + yet. The two raw scopes come back alongside the merge because the place + verifier needs them to pick an anchor (server-26#37). """ - empty: dict[str, Any] = {"vocabulary": [], "ten_codes": {}, "area_lines": []} + empty: dict[str, Any] = { + "vocabulary": [], "ten_codes": {}, "area_lines": [], + "area": {}, "system_area": {}, "tg_area": {}, + } if not system_id: return empty @@ -155,12 +171,18 @@ async def resolve_context(system_id: Optional[str], talkgroup_id: Optional[int]) 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") - ) + system_area = system_doc.get("area_context") or {} + tg_area = tg.get("area_context") or {} + area = area_context.effective(system_area, tg_area) - return {"vocabulary": vocabulary, "ten_codes": ten_codes, "area_lines": area_lines} + return { + "vocabulary": vocabulary, + "ten_codes": ten_codes, + "area_lines": _area_lines(area), + "area": area, + "system_area": system_area, + "tg_area": tg_area, + } def build_context_block(context: dict, talkgroup_name: Optional[str]) -> str: @@ -267,6 +289,34 @@ async def correct( f"segment(s) against {len(segments)} — discarding segment corrections" ) + # Maps has the last word on place names (server-26#37). The corrector can + # only match against the list it was handed, so a plausible-sounding invention + # — "Cool Parts, Illinois" — reads exactly like a real street to it. The + # verifier geocodes each location noun against the talkgroup's anchor and, + # on a miss, looks for a sound-alike that does resolve there. It runs on the + # corrected copy so it judges the text everything downstream will actually + # read, and it skips entirely when there is no discriminating anchor. + if not not_speech: + locations = [x for x in (raw.get("locations") or []) if isinstance(x, str)] + try: + verified_text, verified_segments = await place_verifier.verify( + call_id, + corrected or text, + corrected_segments or segments, + locations, + context.get("system_area"), + context.get("tg_area"), + system_id=system_id, + talkgroup_id=talkgroup_id, + ) + except Exception as e: + logger.warning(f"Place verification failed for call {call_id}: {e}") + verified_text, verified_segments = None, None + if verified_text: + corrected = verified_text + if verified_segments: + corrected_segments = verified_segments + if corrected or corrected_segments or not_speech: changed = raw.get("changed") or [] logger.info( diff --git a/drb-c2-core/app/internal/vocabulary_learner.py b/drb-c2-core/app/internal/vocabulary_learner.py index 3b6e84d..fa90dc1 100644 --- a/drb-c2-core/app/internal/vocabulary_learner.py +++ b/drb-c2-core/app/internal/vocabulary_learner.py @@ -20,8 +20,9 @@ import json import random import re from datetime import datetime, timezone, timedelta -from typing import Optional +from typing import Any, Optional from app.internal.logger import logger +from app.internal import area_context from app.internal import firestore as fstore from app.config import settings @@ -57,26 +58,32 @@ Do NOT include common English words. Max 80 terms. Only include what you are con accurate for this specific area; return fewer terms rather than guessing.""" _INDUCTION_PROMPT = """\ -You are analyzing P25 emergency radio transcripts to find vocabulary terms that should be \ -added to improve future speech-to-text accuracy for this system. +You are analyzing P25 emergency radio transcripts from ONE talkgroup (a single radio channel) \ +to find local terms that should be added to improve future speech-to-text accuracy for that \ +channel. System: {system_name} -Existing approved vocabulary (do not re-propose these): {existing_vocab} +Channel: {talkgroup_name} +Area: {area_hint} +Terms this channel already knows (do not re-propose these): {existing_vocab} Sampled transcripts: {transcript_block} -Find terms that are LIKELY STT errors or local terms missing from the vocabulary: +Find terms that are LIKELY STT errors or local terms missing from the list: - Unit IDs that appear garbled (e.g. "5 acre" → "5-baker") - Agency acronyms spelled out phonetically (e.g. "why vac" → "YVAC") - Street names or locations that look misspelled or oddly transcribed -- Callsigns or local codes not yet in the vocabulary +- Callsigns or local codes not yet known Return ONLY a JSON object: -{{"new_terms": ["term1", "term2", ...]}} +{{"new_terms": [{{"term": "YVAC", "meaning": "Yorktown Volunteer Ambulance Corps"}}, ...]}} -Only include high-confidence additions not already in existing vocabulary. -Return {{"new_terms": []}} if nothing new is found.""" +`meaning` is what the term refers to — an agency, a road, a unit type. Omit it or use null \ +when you genuinely do not know; a term with no meaning is still worth proposing. + +Only propose what is specific to THIS channel and this area. Do not propose a term just \ +because it appears often. Return {{"new_terms": []}} if nothing new is found.""" # ───────────────────────────────────────────────────────────────────────────── @@ -97,10 +104,17 @@ async def bootstrap_system_vocabulary(system_id: str) -> list[str]: system_name = system_doc.get("name", "Unknown") system_type = system_doc.get("type", "P25") - # Build area hint from configured talkgroup names - talkgroups = system_doc.get("config", {}).get("talkgroups", []) - tg_names = [tg.get("name", "") for tg in talkgroups if tg.get("name")][:8] - area_hint = f"Talkgroups include: {', '.join(tg_names)}" if tg_names else "Unknown area" + # Prefer the place an operator actually set. Guessing the area from talkgroup + # names is thin for a single-municipality system and close to useless for a + # multi-county one (server-26#36), so it is only the fallback now. + area = system_doc.get("area_context") or {} + place = ", ".join(str(area[f]) for f in area_context.PLACE_FIELDS if area.get(f)) + if place: + area_hint = place + else: + talkgroups = system_doc.get("config", {}).get("talkgroups", []) + tg_names = [tg.get("name", "") for tg in talkgroups if tg.get("name")][:8] + area_hint = f"Talkgroups include: {', '.join(tg_names)}" if tg_names else "Unknown area" terms = await asyncio.to_thread(_sync_bootstrap, system_name, system_type, area_hint) if not terms: @@ -279,9 +293,20 @@ async def _run_induction_pass() -> None: async def _induct_system(system_id: str, system_doc: dict) -> None: - """Sample random transcripts for a system and propose new vocabulary.""" - system_name = system_doc.get("name", "Unknown") - existing_vocab: list[str] = system_doc.get("vocabulary") or [] + """ + Sample recent transcripts per TALKGROUP and propose local knowledge there. + + Proposals used to land at system level, which is the wrong blast radius + (server-26#37). A wrong term on a talkgroup misleads one channel; the same + term at system level misleads every channel on that system — including one + 400km away on a statewide system, which is exactly the context poisoning the + scope rule exists to prevent. If a term really does apply system-wide, + carrying it on several talkgroups costs almost nothing, while auto-promoting + a wrong one is expensive to notice. So: talkgroup-level pending terms only, + and nothing here ever promotes upward or approves itself. + """ + system_name = system_doc.get("name", "Unknown") + system_area = system_doc.get("area_context") or {} # Fetch calls from the last 7 days only — avoids scanning the entire history. # Active calls have ended_at=None and are excluded by the range filter automatically. @@ -294,57 +319,87 @@ async def _induct_system(system_id: str, system_doc: dict) -> None: if not all_calls: return - # Random sample up to the token budget (4 chars ≈ 1 token) - random.shuffle(all_calls) - char_budget = settings.vocabulary_induction_sample_tokens * 4 + by_tg: dict[Any, list[dict]] = {} + for call in all_calls: + tgid = call.get("talkgroup_id") + if tgid is None: + continue + by_tg.setdefault(tgid, []).append(call) + + # The sample budget is per system, split across the talkgroups that have + # traffic — a channel with 400 calls should not starve one with 12. + char_budget = max( + (settings.vocabulary_induction_sample_tokens * 4) // max(len(by_tg), 1), 800 + ) + + for talkgroup_id, calls in by_tg.items(): + try: + await _induct_talkgroup( + system_id, system_doc, system_name, system_area, + talkgroup_id, calls, char_budget, + ) + except Exception as e: + logger.warning( + f"Induction failed for talkgroup {talkgroup_id} on system {system_id}: {e}" + ) + + +async def _induct_talkgroup( + system_id: str, + system_doc: dict, + system_name: str, + system_area: dict, + talkgroup_id: Any, + calls: list[dict], + char_budget: int, +) -> None: + tg_entry = area_context.talkgroup_entry(system_doc, talkgroup_id) + tg_area = tg_entry.get("area_context") or {} + area = area_context.effective(system_area, tg_area) + + talkgroup_name = ( + tg_entry.get("name") + or calls[0].get("talkgroup_name") + or f"TGID {talkgroup_id}" + ) + known = area_context._known_terms(tg_entry, system_doc) + + random.shuffle(calls) transcript_block = "" sampled_call_docs: list[dict] = [] - sampled = 0 - for call in all_calls: + for call in calls: text = call.get("transcript_corrected") or call.get("transcript") or "" if not text: continue if len(transcript_block) + len(text) > char_budget: break - tg = call.get("talkgroup_name") or f"TGID {call.get('talkgroup_id', '?')}" - transcript_block += f"[{tg}] {text}\n" + transcript_block += f"{text}\n" sampled_call_docs.append(call) - sampled += 1 - if sampled < 3: - return # not enough data to learn from yet + if len(sampled_call_docs) < 3: + return # not enough data on this channel to learn from yet - new_terms = await asyncio.to_thread( - _sync_induct, system_name, existing_vocab, transcript_block + place = ", ".join(str(area[f]) for f in area_context.PLACE_FIELDS if area.get(f)) + proposed = await asyncio.to_thread( + _sync_induct, + system_name, talkgroup_name, place or "not set", + sorted(known)[:80], transcript_block, ) - if not new_terms: + if not proposed: return - now = datetime.now(timezone.utc).isoformat() - existing_pending: list[dict] = system_doc.get("vocabulary_pending") or [] - pending_lower = {p["term"].lower() for p in existing_pending} - vocab_lower = {t.lower() for t in existing_vocab} - - to_queue = [] - for t in new_terms: - if t.lower() in vocab_lower or t.lower() in pending_lower: - continue - to_queue.append({ - "term": t, + entries = [ + { + "term": p["term"], + "meaning": p.get("meaning"), "source": "induction", - "added_at": now, - "source_call_ids": _find_source_calls(t, sampled_call_docs), - }) - if not to_queue: - return - - await fstore.doc_set("systems", system_id, { - "vocabulary_pending": existing_pending + to_queue, - }) - logger.info( - f"Vocabulary induction: {len(to_queue)} new term(s) proposed for " - f"system {system_id} ({system_name}): {[p['term'] for p in to_queue]}" - ) + "source_call_ids": _find_source_calls(p["term"], sampled_call_docs), + } + for p in proposed + if p.get("term") and p["term"].lower() not in known + ] + if entries: + await area_context.add_pending(system_id, talkgroup_id, entries) # ───────────────────────────────────────────────────────────────────────────── @@ -441,8 +496,13 @@ def _sync_bootstrap(system_name: str, system_type: str, area_hint: str) -> list[ def _sync_induct( - system_name: str, existing_vocab: list[str], transcript_block: str -) -> list[str]: + system_name: str, + talkgroup_name: str, + area_hint: str, + existing_vocab: list[str], + transcript_block: str, +) -> list[dict]: + """Returns [{term, meaning}] — a bare string is still accepted from the model.""" from app.config import settings as cfg from openai import OpenAI @@ -452,6 +512,8 @@ def _sync_induct( vocab_str = ", ".join(existing_vocab[:80]) if existing_vocab else "(none yet)" prompt = _INDUCTION_PROMPT.format( system_name=system_name, + talkgroup_name=talkgroup_name, + area_hint=area_hint, existing_vocab=vocab_str, transcript_block=transcript_block[:8000], ) @@ -463,8 +525,7 @@ def _sync_induct( response_format={"type": "json_object"}, ) data = json.loads(response.choices[0].message.content) - terms = data.get("new_terms") or [] - return [str(t).strip() for t in terms if str(t).strip()] + return area_context.normalize_local_knowledge(data.get("new_terms") or []) except Exception as e: logger.warning(f"Vocabulary induction GPT call failed: {e}") return [] diff --git a/drb-c2-core/app/models.py b/drb-c2-core/app/models.py index 99859f1..f7233a2 100644 --- a/drb-c2-core/app/models.py +++ b/drb-c2-core/app/models.py @@ -78,6 +78,49 @@ class CommandPayload(BaseModel): # Systems # --------------------------------------------------------------------------- +class LocalKnowledgeEntry(BaseModel): + """A local name and what it is. Both scopes, one shape (server-26#36).""" + term: str + meaning: Optional[str] = None + + +class AreaContextBody(BaseModel): + """ + Ground truth about the area a system or talkgroup covers. + + Every field is nullable on purpose: which SCOPE an operator fills is their + declaration of how homogeneous the system is. A single-municipality system + is described once at system level and inherited by every talkgroup; a + statewide one is left null there and described per talkgroup. See + `internal/area_context.py` for the merge rules and the anchor. + + `center`/`radius_km`/`resolved_from`/`resolved_at` are absent here by + design — the backend geocodes and writes those. A client that sends them is + ignored. + """ + municipality: Optional[str] = None + county: Optional[str] = None + state: Optional[str] = None + local_knowledge: List[LocalKnowledgeEntry] = [] + + +class TalkgroupEntry(BaseModel): + """ + One entry in `config.talkgroups[]`. + + Declared so the talkgroup copy of `area_context` stops being unvalidated + JSON riding inside the config blob — it is the same shape as the system's + and gets the same validator (server-26#36). + """ + model_config = {"extra": "allow"} + + id: int + name: str = "" + tag: str = "other" + vocabulary: List[str] = [] + area_context: Optional[AreaContextBody] = None + + class SystemRecord(BaseModel): system_id: str org_id: Optional[str] = None @@ -86,9 +129,10 @@ class SystemRecord(BaseModel): 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. + # corrector and the place verifier (server-26#36 / #37). Shape is + # AreaContextBody plus the backend-owned anchor. Per-talkgroup overrides + # live inside config.talkgroups[] and rank ABOVE this, so a multi-county + # system narrows per channel rather than replacing this wholesale. area_context: Dict[str, Any] = {} diff --git a/drb-c2-core/app/routers/systems.py b/drb-c2-core/app/routers/systems.py index 602fa5e..0bc15fd 100644 --- a/drb-c2-core/app/routers/systems.py +++ b/drb-c2-core/app/routers/systems.py @@ -2,8 +2,9 @@ import uuid from fastapi import APIRouter, HTTPException, Depends, Query from pydantic import BaseModel from typing import Dict, List, Optional -from app.models import SystemCreate, SystemRecord +from app.models import AreaContextBody, SystemCreate, SystemRecord from app.internal import firestore as fstore +from app.internal import area_context as area_ctx from app.internal.auth import ( require_admin_token, require_node_service_or_firebase_token, @@ -23,12 +24,9 @@ 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 PendingTermBody(BaseModel): + talkgroup_id: int + term: str class AiFlagsBody(BaseModel): @@ -78,7 +76,21 @@ async def update_system(system_id: str, body: SystemCreate, _: dict = Depends(re # 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) + # The form sends config.talkgroups[] in full, which would erase the resolved + # anchor and the pending-term queue the backend put there. Same class of bug + # as ten_codes above; the backend merges its own fields back rather than + # taking dictation from the client (server-26#36). + if "config" in patch: + patch["config"] = area_ctx.merge_config(patch["config"], existing.get("config")) + if "area_context" in patch: + patch["area_context"] = area_ctx.merge_server_fields( + area_ctx.normalize(patch["area_context"]), existing.get("area_context") + ) await fstore.doc_update("systems", system_id, patch) + # Geocoding the anchor is a write-time job — a place changes when someone + # edits a town name, not every five minutes — but the operator should not + # wait on Maps to see their save land. + area_ctx.schedule_refresh(system_id) return {**existing, **patch} @@ -160,12 +172,17 @@ async def update_area_context( _: dict = Depends(require_admin_token), ): """ - Replace the system-wide area context used by the transcript corrector. + Replace the system-wide area context used by the corrector and the verifier. - 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 + Ground truth about where this system operates — municipality, county, state, + and the local names whose sound 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. + Leaving it entirely empty is legitimate and meaningful: it says nothing here + is true of every talkgroup. + + The derived anchor (`center`, `radius_km`, `resolved_from`, `resolved_at`) is + never taken from the body — it is carried forward and then recomputed here. 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 @@ -174,11 +191,69 @@ async def update_area_context( 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) + area = area_ctx.merge_server_fields( + area_ctx.normalize(body.model_dump(exclude_none=True)), + existing.get("area_context"), + ) await fstore.doc_update("systems", system_id, {"area_context": area}) + # Awaited, not scheduled: this route exists to edit the place, so the caller + # should get back the anchor its edit produced. Talkgroups are refreshed with + # it because their anchor derives from the merged place, not their own. + patch = await area_ctx.refresh_anchors({**existing, "area_context": area}) + if patch: + await fstore.doc_update("systems", system_id, patch) + area = patch.get("area_context", area) return {"ok": True, "area_context": area} +# -- Talkgroup-level pending local knowledge (server-26#37) -------------------- + +@router.get("/{system_id}/talkgroup-pending") +async def list_talkgroup_pending(system_id: str, _: dict = Depends(require_admin_token)): + """ + Every pending local-knowledge proposal on this system, by talkgroup. + + Proposals are made at talkgroup level and are never promoted to the system + automatically — a wrong term on one channel misleads one channel, the same + term system-wide misleads every channel on it. + """ + system = await fstore.doc_get("systems", system_id) + if not system: + raise HTTPException(404, f"System '{system_id}' not found.") + out = [] + for tg in ((system.get("config") or {}).get("talkgroups") or []): + if not isinstance(tg, dict): + continue + pending = tg.get(area_ctx.PENDING_KEY) or [] + if pending: + out.append({ + "talkgroup_id": tg.get("id"), + "talkgroup_name": tg.get("name"), + "pending": pending, + }) + return {"talkgroups": out} + + +@router.post("/{system_id}/talkgroup-pending/approve") +async def approve_talkgroup_pending( + system_id: str, body: PendingTermBody, _: dict = Depends(require_admin_token) +): + """Move a pending term into that talkgroup's local_knowledge.""" + if not await area_ctx.resolve_pending(system_id, body.talkgroup_id, body.term, approve=True): + raise HTTPException(404, "No such pending term on that talkgroup.") + return {"ok": True} + + +@router.post("/{system_id}/talkgroup-pending/dismiss") +async def dismiss_talkgroup_pending( + system_id: str, body: PendingTermBody, _: dict = Depends(require_admin_token) +): + """Drop a pending term without adding it.""" + if not await area_ctx.resolve_pending(system_id, body.talkgroup_id, body.term, approve=False): + raise HTTPException(404, "No such pending term on that talkgroup.") + return {"ok": True} + + # ── Vocabulary endpoints ─────────────────────────────────────────────────────── @router.get("/{system_id}/vocabulary") diff --git a/drb-c2-core/tests/test_area_context.py b/drb-c2-core/tests/test_area_context.py new file mode 100644 index 0000000..99fbb95 --- /dev/null +++ b/drb-c2-core/tests/test_area_context.py @@ -0,0 +1,305 @@ +""" +Unit tests for the area_context schema and anchor (server-26#36). + +Three properties carry the real risk: + + * NULLABILITY IS THE MECHANISM. Which scope an operator fills is their + declaration of how homogeneous the system is. Merging must let a talkgroup + narrow the system without dropping what the system already said — a + talkgroup that sets only a town must still inherit the state, or "Ossining" + is nationally ambiguous again. + + * NO ANCHOR IS BETTER THAN A USELESS ONE. An anchor wider than + area_anchor_max_radius_km, or one whose resolved_from no longer matches the + place it came from, must read as ABSENT. Verification then skips. Treating + either as usable would rubber-stamp any location while looking like a check. + + * THE CLIENT DOES NOT WRITE SERVER FIELDS. The systems form sends + config.talkgroups[] in full; taking it verbatim destroys the resolved anchor + and the pending queue, which is the same bug as the ten_codes wipe. +""" +import pytest +from unittest.mock import AsyncMock, patch + +from app.internal import area_context as ac + +SYSTEM_AREA = { + "county": "Westchester", + "state": "New York", + "local_knowledge": [{"term": "Route 9", "meaning": "state highway"}], + "center": {"lat": 41.1, "lng": -73.8}, + "radius_km": 30.0, + "resolved_from": "|westchester|new york", + "resolved_at": "2026-08-23T00:00:00+00:00", +} + +TG_AREA = { + "municipality": "Ossining", + "local_knowledge": [{"term": "Sing Sing", "meaning": "state prison"}], + "center": {"lat": 41.16, "lng": -73.86}, + "radius_km": 6.0, + "resolved_from": "ossining|westchester|new york", + "resolved_at": "2026-08-23T00:00:00+00:00", +} + + +# -- Merging ------------------------------------------------------------------- + +def test_talkgroup_narrows_without_dropping_the_system(): + merged = ac.effective(SYSTEM_AREA, TG_AREA) + assert merged["municipality"] == "Ossining" + assert merged["county"] == "Westchester" + assert merged["state"] == "New York", "the state must survive the narrowing" + + +def test_talkgroup_knowledge_ranks_first_and_dedupes(): + system = {"local_knowledge": [{"term": "Route 9"}, {"term": "Metro-North"}]} + tg = {"local_knowledge": [{"term": "route 9", "meaning": "the local name"}]} + merged = ac.effective(system, tg) + assert [e["term"] for e in merged["local_knowledge"]] == ["route 9", "Metro-North"] + assert merged["local_knowledge"][0]["meaning"] == "the local name" + + +def test_empty_at_both_scopes_is_legal(): + assert ac.effective(None, None) == {} + assert ac.effective({}, {}) == {} + + +def test_bare_strings_are_accepted_as_terms(): + """roads[]/landmarks[] from the old shape, and anything a model returns.""" + assert ac.normalize_local_knowledge(["Route 9", "", "Route 9", 7]) == [{"term": "Route 9"}] + + +def test_pre_36_roads_and_landmarks_are_read_forward(): + """ + Real systems still have the old shape stored. Dropping it the day this + shipped would silently discard ground truth an operator already entered. + """ + legacy = {"county": "Westchester", "roads": ["Route 9"], "landmarks": ["Sing Sing"]} + merged = ac.effective(legacy, None) + assert [e["term"] for e in merged["local_knowledge"]] == ["Route 9", "Sing Sing"] + assert ac.normalize(legacy) == { + "county": "Westchester", + "local_knowledge": [{"term": "Route 9"}, {"term": "Sing Sing"}], + }, "and the next save writes them in the new shape" + + +def test_normalize_drops_client_sent_server_fields(): + out = ac.normalize({"municipality": " Ossining ", "radius_km": 5000, "center": {"lat": 0}}) + assert out == {"municipality": "Ossining"} + + +# -- Anchor selection ---------------------------------------------------------- + +def test_talkgroup_anchor_wins(): + anchor = ac.anchor_for(SYSTEM_AREA, TG_AREA) + assert anchor == {"lat": 41.16, "lng": -73.86, "radius_km": 6.0} + + +def test_system_anchor_used_when_talkgroup_sets_no_place(): + anchor = ac.anchor_for(SYSTEM_AREA, {"local_knowledge": [{"term": "Post 4"}]}) + assert anchor == {"lat": 41.1, "lng": -73.8, "radius_km": 30.0} + + +def test_no_anchor_when_nothing_is_configured(): + assert ac.anchor_for({}, {}) is None + + +def test_stale_anchor_reads_as_absent(): + """ + Someone edited the town and the refresh has not run yet. The stored centre + is for the OLD place, so using it would validate locations against an area + the channel no longer covers. + """ + stale = {**TG_AREA, "municipality": "Croton"} + assert ac.anchor_for(SYSTEM_AREA, stale) is None + + +def test_anchor_key_ignores_case_and_padding(): + assert ac.anchor_key({"municipality": " OSSINING "}) == ac.anchor_key({"municipality": "ossining"}) + + +# -- Anchor resolution --------------------------------------------------------- + +def _maps(viewport_span_deg: float): + """A geocode response whose viewport spans roughly the given degrees.""" + payload = { + "status": "OK", + "results": [{ + "geometry": { + "location": {"lat": 41.0, "lng": -73.0}, + "viewport": { + "northeast": {"lat": 41.0 + viewport_span_deg, "lng": -73.0 + viewport_span_deg}, + "southwest": {"lat": 41.0 - viewport_span_deg, "lng": -73.0 - viewport_span_deg}, + }, + } + }], + } + + class _Resp: + def raise_for_status(self): pass + def json(self): return payload + + class _Client: + async def __aenter__(self): return self + async def __aexit__(self, *a): return False + async def get(self, *a, **k): return _Resp() + + return patch("httpx.AsyncClient", lambda *a, **k: _Client()) + + +@pytest.fixture(autouse=True) +def _clear_cache(): + ac._anchor_cache.clear() + with patch.object(ac.settings, "google_maps_api_key", "test-key"): + yield + ac._anchor_cache.clear() + + +@pytest.mark.asyncio +async def test_small_place_produces_an_anchor(): + with _maps(0.05): + anchor = await ac.resolve_anchor({"municipality": "Ossining", "state": "New York"}) + assert anchor is not None + assert anchor["radius_km"] < 10 + assert anchor["resolved_from"] == "ossining||new york" + + +@pytest.mark.asyncio +async def test_statewide_place_produces_no_anchor(): + """ + A radius that covers a state would confirm any location inside it. Storing + it would make the geocode check worse than useless — it would look like + verification and pass everything. + """ + with _maps(4.0), patch.object(ac.settings, "area_anchor_max_radius_km", 60.0): + assert await ac.resolve_anchor({"state": "Colorado"}) is None + + +@pytest.mark.asyncio +async def test_no_place_never_calls_maps(): + with patch("httpx.AsyncClient") as client: + assert await ac.resolve_anchor({"local_knowledge": [{"term": "Post 4"}]}) is None + client.assert_not_called() + + +@pytest.mark.asyncio +async def test_refresh_skips_scopes_whose_place_is_unchanged(): + doc = {"area_context": SYSTEM_AREA, "config": {"talkgroups": [{"id": 1, "area_context": TG_AREA}]}} + with patch("httpx.AsyncClient") as client: + assert await ac.refresh_anchors(doc) == {} + client.assert_not_called() + + +@pytest.mark.asyncio +async def test_editing_the_system_place_re_anchors_its_talkgroups(): + """ + A talkgroup's anchor derives from its EFFECTIVE place, so changing the + system's county silently changes what every talkgroup should be anchored to. + """ + doc = { + "area_context": {"county": "Putnam", "state": "New York"}, + "config": {"talkgroups": [{"id": 1, "area_context": {"municipality": "Ossining"}}]}, + } + with _maps(0.05): + patch_out = await ac.refresh_anchors(doc) + tg = patch_out["config"]["talkgroups"][0] + assert tg["area_context"]["resolved_from"] == "ossining|putnam|new york" + assert tg["area_context"]["center"]["lat"] == 41.0 + + +# -- Client writes ------------------------------------------------------------- + +def test_merge_config_preserves_the_anchor_and_the_pending_queue(): + existing = {"talkgroups": [{ + "id": 9048, + "area_context": TG_AREA, + ac.PENDING_KEY: [{"term": "Snowden Avenue"}], + }]} + # What the systems form actually sends: no anchor, no pending queue. + incoming = {"talkgroups": [{"id": 9048, "name": "Ossining PD", + "area_context": {"municipality": "Ossining"}}]} + merged = ac.merge_config(incoming, existing) + tg = merged["talkgroups"][0] + assert tg["area_context"]["center"] == TG_AREA["center"] + assert tg[ac.PENDING_KEY] == [{"term": "Snowden Avenue"}] + assert tg["name"] == "Ossining PD", "the client still owns the fields it owns" + + +def test_merge_config_drops_an_emptied_area(): + existing = {"talkgroups": [{"id": 1, "area_context": TG_AREA}]} + merged = ac.merge_config({"talkgroups": [{"id": 1}]}, existing) + assert "area_context" not in merged["talkgroups"][0] + + +# -- Pending terms ------------------------------------------------------------- + +def _store(doc): + saved = {} + + async def _get(_col, _id): + return doc + + async def _update(_col, _id, patch): + saved.update(patch) + + return saved, patch.multiple( + "app.internal.firestore", doc_get=AsyncMock(side_effect=_get), + doc_update=AsyncMock(side_effect=_update), + ) + + +@pytest.mark.asyncio +async def test_pending_terms_land_on_the_talkgroup(): + doc = {"config": {"talkgroups": [{"id": 9048}]}, "vocabulary": []} + saved, store = _store(doc) + with store: + assert await ac.add_pending("sys-1", 9048, [{"term": "Snowden Avenue"}]) == 1 + assert saved["config"]["talkgroups"][0][ac.PENDING_KEY][0]["term"] == "Snowden Avenue" + assert "vocabulary" not in saved, "nothing writes to the system" + + +@pytest.mark.asyncio +async def test_already_known_terms_are_not_re_proposed(): + doc = { + "vocabulary": ["Metro-North"], + "area_context": {"local_knowledge": [{"term": "Route 9"}]}, + "config": {"talkgroups": [{"id": 9048, "local_knowledge_pending": [{"term": "Sing Sing"}]}]}, + } + saved, store = _store(doc) + with store: + queued = await ac.add_pending("sys-1", 9048, [ + {"term": "route 9"}, {"term": "Metro-North"}, {"term": "sing sing"}, + ]) + assert queued == 0 + assert saved == {} + + +@pytest.mark.asyncio +async def test_approving_writes_to_the_talkgroup_and_never_the_system(): + """ + Blast radius: the same term at system level misleads every channel on the + system, including one 400km away on a statewide system. + """ + doc = {"config": {"talkgroups": [{"id": 9048, ac.PENDING_KEY: [ + {"term": "Snowden Avenue", "meaning": "residential street"}]}]}} + saved, store = _store(doc) + with store: + assert await ac.resolve_pending("sys-1", 9048, "snowden avenue", approve=True) is True + tg = saved["config"]["talkgroups"][0] + assert tg["area_context"]["local_knowledge"] == [ + {"term": "Snowden Avenue", "meaning": "residential street"} + ] + assert tg[ac.PENDING_KEY] == [] + assert "vocabulary" not in saved and "area_context" not in saved + + +@pytest.mark.asyncio +async def test_dismissing_adds_nothing(): + doc = {"config": {"talkgroups": [{"id": 9048, ac.PENDING_KEY: [{"term": "Optum"}]}]}} + saved, store = _store(doc) + with store: + assert await ac.resolve_pending("sys-1", 9048, "Optum", approve=False) is True + tg = saved["config"]["talkgroups"][0] + assert tg[ac.PENDING_KEY] == [] + assert not (tg.get("area_context") or {}).get("local_knowledge") diff --git a/drb-c2-core/tests/test_place_verifier.py b/drb-c2-core/tests/test_place_verifier.py new file mode 100644 index 0000000..a84081e --- /dev/null +++ b/drb-c2-core/tests/test_place_verifier.py @@ -0,0 +1,192 @@ +""" +Unit tests for Maps-based place verification (server-26#37). + +The property that matters most is the one that looks like a no-op: WITHOUT AN +ANCHOR, NOTHING HAPPENS. A system whose area is too wide to discriminate stores +no anchor, and verification must then skip entirely rather than accept whatever +geocodes. A check that passes everything is worse than no check, because it +reads as verification in the logs and in the data. + +After that: a candidate may only rewrite a transcript if it actually sounds like +what was heard. Places Text Search will return the nearest plausible business +for any garbage string, so the API answering at all is not evidence. +""" +import pytest +from unittest.mock import AsyncMock, patch + +from app.internal import place_verifier as pv + +ANCHOR_AREA = { + "municipality": "Ossining", + "county": "Westchester", + "state": "New York", + "local_knowledge": [{"term": "Snowden Avenue", "meaning": "residential street"}], + "center": {"lat": 41.16, "lng": -73.86}, + "radius_km": 6.0, + "resolved_from": "ossining|westchester|new york", +} + +SEGS = [{"start": 0.0, "end": 1.0, "text": "Shout out to Optum."}, + {"start": 1.0, "end": 2.0, "text": "Copy that."}] + + +@pytest.fixture(autouse=True) +def _enabled(): + with patch.object(pv.settings, "place_verification_enabled", True), \ + patch.object(pv.settings, "google_maps_api_key", "test-key"): + yield + + +def _geocode(result): + return patch.object(pv, "_geocode_in_anchor", AsyncMock(return_value=result)) + + +def _places(result): + return patch.object(pv, "_places_soundalike", AsyncMock(return_value=result)) + + +# -- Phonetics ----------------------------------------------------------------- + +@pytest.mark.parametrize("heard, real", [ + ("Snowden Avenue", "Snowdon Ave"), + ("5 acre", "5-baker"), + ("why vac", "YVAC"), + ("Croton Ave", "Croton Avenue"), +]) +def test_real_mishearings_score_above_the_threshold(heard, real): + assert pv.sounds_like(heard, real) >= pv.settings.place_soundalike_min_ratio + + +@pytest.mark.parametrize("heard, unrelated", [ + ("Optum", "Ossining"), + ("Cool Parts", "Croton Point"), +]) +def test_unrelated_names_score_below_it(heard, unrelated): + assert pv.sounds_like(heard, unrelated) < pv.settings.place_soundalike_min_ratio + + +# -- The skip path ------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_no_anchor_means_skip_not_accept(): + """A statewide system stores no anchor. Nothing may be checked or rewritten.""" + with patch.object(pv, "_geocode_in_anchor") as geo: + out = await pv.verify("c1", "text here", SEGS, ["Optum"], {"state": "Colorado"}, {}) + assert out == (None, None) + geo.assert_not_called() + + +@pytest.mark.asyncio +async def test_no_locations_means_no_requests(): + with patch.object(pv, "_geocode_in_anchor") as geo: + assert await pv.verify("c1", "t", None, [], ANCHOR_AREA, {}) == (None, None) + geo.assert_not_called() + + +@pytest.mark.asyncio +async def test_disabled_by_setting(): + with patch.object(pv.settings, "place_verification_enabled", False), \ + patch.object(pv, "_geocode_in_anchor") as geo: + assert await pv.verify("c1", "t", None, ["Optum"], ANCHOR_AREA, {}) == (None, None) + geo.assert_not_called() + + +# -- The accept path ----------------------------------------------------------- + +@pytest.mark.asyncio +async def test_a_place_that_resolves_inside_the_anchor_is_left_alone(): + with _geocode({"lat": 41.16, "lng": -73.86}), _places(None) as places: + out = await pv.verify("c1", "Units to Snowden Avenue.", None, + ["Snowden Avenue"], ANCHOR_AREA, {}) + assert out == (None, None) + places.assert_not_called() # a hit must not cost a second request + + +@pytest.mark.asyncio +async def test_the_query_carries_the_full_place(): + seen = {} + + async def capture(query, anchor): + seen["query"] = query + return {"lat": 41.16, "lng": -73.86} + + with patch.object(pv, "_geocode_in_anchor", capture): + await pv.verify("c1", "t", None, ["High Street"], ANCHOR_AREA, {}) + assert seen["query"] == "High Street, Ossining, Westchester, New York" + + +# -- The correction path ------------------------------------------------------- + +@pytest.mark.asyncio +async def test_known_term_is_preferred_and_costs_nothing(): + """ + A sound-alike the operator already entered is both free and more trustworthy + than anything Maps guesses, so it must be tried before any request goes out. + """ + with _geocode(None), _places(None) as places, \ + patch.object(pv.area_context, "add_pending", AsyncMock()) as add: + text, segs = await pv.verify( + "c1", "Units to Snowdon Ave.", None, ["Snowdon Ave"], ANCHOR_AREA, {} + ) + assert text == "Units to Snowden Avenue." + places.assert_not_called() + add.assert_not_called() # already known — nothing to propose + + +@pytest.mark.asyncio +async def test_a_maps_soundalike_is_applied_and_proposed_to_the_talkgroup(): + candidate = {"term": "Croton Point", "meaning": "Croton Point Ave, Croton NY", "score": 0.8} + with _geocode(None), _places(candidate), \ + patch.object(pv.area_context, "add_pending", AsyncMock(return_value=1)) as add: + text, segs = await pv.verify( + "c1", "Respond to Cool Parts.", None, ["Cool Parts"], ANCHOR_AREA, {}, + system_id="sys-1", talkgroup_id=9048, + ) + assert text == "Respond to Croton Point." + args = add.await_args.args + assert args[0] == "sys-1" and args[1] == 9048 + assert args[2][0]["term"] == "Croton Point" + assert args[2][0]["source_call_ids"] == ["c1"] + + +@pytest.mark.asyncio +async def test_nothing_plausible_leaves_the_transcript_untouched(): + """ + An invented name with no real counterpart nearby stays as it is. Guessing + would put a fabricated location into the incident record, which is the + outcome this whole pass exists to avoid. + """ + with _geocode(None), _places(None): + assert await pv.verify("c1", "Shout out to Optum.", SEGS, + ["Optum"], ANCHOR_AREA, {}) == (None, None) + + +@pytest.mark.asyncio +async def test_segments_are_corrected_alongside_the_joined_text(): + """Extraction reads numbered segments, so a joined-only fix reaches nothing.""" + with _geocode(None), _places(None), \ + patch.object(pv.area_context, "add_pending", AsyncMock()): + text, segs = await pv.verify( + "c1", "Shout out to Snowdon Ave. Copy that.", + [{"start": 0.0, "end": 1.0, "text": "Shout out to Snowdon Ave."}, + {"start": 1.0, "end": 2.0, "text": "Copy that."}], + ["Snowdon Ave"], ANCHOR_AREA, {}, + ) + assert segs is not None + assert segs[0]["text"] == "Shout out to Snowden Avenue." + assert segs[0]["start"] == 0.0, "timing survives untouched" + assert segs[1]["text"] == "Copy that." + + +@pytest.mark.asyncio +async def test_a_geocoder_failure_never_breaks_the_transcript(): + with patch.object(pv, "_geocode_in_anchor", AsyncMock(side_effect=RuntimeError("boom"))): + assert await pv.verify("c1", "t here", None, ["Optum"], ANCHOR_AREA, {}) == (None, None) + + +@pytest.mark.asyncio +async def test_only_a_bounded_number_of_nouns_is_checked(): + with patch.object(pv.settings, "place_verify_max_per_call", 2), \ + patch.object(pv, "_geocode_in_anchor", AsyncMock(return_value={"lat": 41.16, "lng": -73.86})) as geo: + await pv.verify("c1", "t", None, ["a", "b", "c", "d"], ANCHOR_AREA, {}) + assert geo.await_count == 2 diff --git a/drb-c2-core/tests/test_transcript_correction.py b/drb-c2-core/tests/test_transcript_correction.py index 3869a03..40eda3b 100644 --- a/drb-c2-core/tests/test_transcript_correction.py +++ b/drb-c2-core/tests/test_transcript_correction.py @@ -21,14 +21,24 @@ 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"]}, + "area_context": { + "county": "Westchester", + "state": "New York", + "local_knowledge": [ + {"term": "Route 9", "meaning": "north-south state highway"}, + {"term": "Saw Mill Parkway"}, + ], + }, "config": { "talkgroups": [ { "id": 9048, "name": "Ossining - Police Dispatch", "vocabulary": ["Snowden Avenue", "Croton-Harmon"], - "area_context": {"municipality": "Ossining", "landmarks": ["Sing Sing"]}, + "area_context": { + "municipality": "Ossining", + "local_knowledge": [{"term": "Sing Sing", "meaning": "state prison"}], + }, }, {"id": 9600, "name": "Harrison - Police/EMS Dispatch"}, {"id": 9563, "ten_codes": {"10-4": "on scene"}}, @@ -83,6 +93,7 @@ async def test_talkgroup_without_own_data_inherits_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"]) + assert ctx["area"].get("municipality") is None, "inherits, invents nothing" @pytest.mark.asyncio @@ -105,7 +116,10 @@ async def test_missing_scope_is_not_an_error(system_id, tgid): 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": []} + assert ctx == { + "vocabulary": [], "ten_codes": {}, "area_lines": [], + "area": {}, "system_area": {}, "tg_area": {}, + } # ── Correction behaviour ──────────────────────────────────────────────────── @@ -207,5 +221,7 @@ async def test_reference_data_reaches_the_prompt(): 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 "Snowden Avenue" in p and "Ossining - Police Dispatch" in p + assert "Sing Sing — state prison" in p, "a term without its meaning is half the information" + assert "Ossining, Westchester, New York" in p, "state must reach the prompt (server-26#36)" 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 995bd53..afe9473 100644 --- a/drb-frontend/app/systems/page.tsx +++ b/drb-frontend/app/systems/page.tsx @@ -5,7 +5,13 @@ import { useRouter } from "next/navigation"; import { useSystems } from "@/lib/useSystems"; import { c2api } from "@/lib/c2api"; import { useAuth } from "@/components/AuthProvider"; -import type { AreaContext, SystemRecord, VocabularyPendingTerm } from "@/lib/types"; +import type { + AreaContext, + LocalKnowledgeEntry, + SystemRecord, + TalkgroupPending, + VocabularyPendingTerm, +} from "@/lib/types"; // ── P25 structured config types ─────────────────────────────────────────────── @@ -25,17 +31,77 @@ 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. */ +/** + * Local knowledge is one entry per line, `term — meaning`. + * + * A bare term is valid — half the information is still worth having, and + * forcing a meaning would make people invent one. An em dash, a spaced hyphen + * or an equals sign all separate; whichever comes first wins. + */ +const knowledgeToText = (v?: LocalKnowledgeEntry[]) => + (v ?? []).map((e) => (e.meaning ? `${e.term} — ${e.meaning}` : e.term)).join("\n"); + +function textToKnowledge(v: string): LocalKnowledgeEntry[] { + const out: LocalKnowledgeEntry[] = []; + for (const line of v.split("\n")) { + const match = line.match(/^(.*?)\s*(?:—|–|\s-\s|=)\s*(.*)$/); + const term = (match ? match[1] : line).trim(); + const meaning = match ? match[2].trim() : ""; + if (term) out.push(meaning ? { term, meaning } : { term }); + } + return out; +} + +/** + * Fold the pre-#36 shape forward on load. + * + * `roads[]` and `landmarks[]` were the original fields and real systems still + * have them stored. Showing them as local-knowledge lines means an operator + * sees what they already entered instead of an empty box, and the next save + * writes them in the new shape. + */ +function migrateArea(a?: AreaContext & { roads?: string[]; landmarks?: string[] }): AreaContext { + if (!a) return {}; + const legacy = [...(a.roads ?? []), ...(a.landmarks ?? [])].map((term) => ({ term })); + if (!legacy.length) return a; + const seen = new Set((a.local_knowledge ?? []).map((e) => e.term.toLowerCase())); + const { roads: _roads, landmarks: _landmarks, ...rest } = a; + return { + ...rest, + local_knowledge: [ + ...(a.local_knowledge ?? []), + ...legacy.filter((e) => !seen.has(e.term.toLowerCase())), + ], + }; +} + +/** + * Drop an area_context whose every field is blank, so we don't store noise. + * + * Also strips the derived anchor: `center`/`radius_km`/`resolved_*` belong to + * the backend, which geocodes them from the place and merges them back. Sending + * them up would be the frontend deciding what is in a system document. + */ 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; + if (a.state?.trim()) out.state = a.state.trim(); + if (a.local_knowledge?.length) out.local_knowledge = a.local_knowledge; return Object.keys(out).length ? out : undefined; } +/** What the backend resolved this area to, in one line. */ +function anchorLabel(a?: AreaContext): string { + if (!a) return ""; + const place = [a.municipality, a.county, a.state].filter(Boolean).join(", "); + if (!place) return "no area set — location checking is off for this scope"; + if (a.center && a.radius_km) return `anchored to ${place}, ${Math.round(a.radius_km)} km radius`; + if (a.resolved_from) return `${place} — too wide to check locations against, so that check is skipped`; + return `${place} — not yet resolved`; +} + interface P25Config { nac: string; system_id: string; @@ -76,7 +142,7 @@ function recordToP25Config(c: Record): P25Config { name: tg.name, tag: tg.tag ?? "other", vocabulary: tg.vocabulary ?? [], - area_context: tg.area_context ?? {}, + area_context: migrateArea(tg.area_context), })) : [], }; @@ -359,27 +425,41 @@ function hasLocalKnowledge(tg: TalkgroupEntry): boolean { return Boolean((tg.vocabulary ?? []).length || cleanArea(tg.area_context)); } -/** One labelled text input in the local-knowledge grid. */ +/** One labelled field in the local-knowledge grid. */ function LocalKnowledgeField({ label, value, onChange, placeholder, + multiline, }: { label: string; value: string; onChange: (v: string) => void; placeholder?: string; + multiline?: boolean; }) { + const cls = + "w-full mt-0.5 bg-gray-900 border border-gray-700 rounded px-2 py-1 text-white text-xs focus:outline-none focus:border-indigo-500"; return (