""" 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