142 lines
5.9 KiB
Python
142 lines
5.9 KiB
Python
"""ZIP/address -> district -> legislators resolution (spec §5, DESIGN §8, D2).
|
|
|
|
Open Decision D2 (inherited from the UX lane) asks which ZIP-to-district data
|
|
source the dashboard owns. The authoritative source is the Legislature's
|
|
full-address tool / a county voter file or data.ca.gov boundary datasets; that
|
|
ETL is the correlation lane's job. This module provides a runnable demonstration
|
|
resolver so the ZIP->legislator flow is real and testable:
|
|
|
|
* a small, documented ZIP sample map (see DEMO_ZIPS) covering a few districts
|
|
of the current 2025-2026 session;
|
|
* a ZIP fast path (0 / 1 / several districts) with the confidence semantics the
|
|
design freezes (single / ambiguous / zero / out-of-state);
|
|
* a manual fallback (pick a district by hand) for the edge cases the design
|
|
requires (P.O. boxes, cross-border ZIPs, post-redistricting gaps).
|
|
|
|
Real-world fact the design grounds this in: a ZIP alone does NOT reliably
|
|
resolve a CA district (a single ZIP can straddle two districts). Fast-path
|
|
ambiguity therefore proposes candidates; the user commits (never auto-watched).
|
|
A real deployment swaps DEMO_ZIPS for the data.ca.gov boundary join, which is
|
|
documented in DASHBOARD.md.
|
|
|
|
Resolve never persists: it proposes (API_CONTRACT section 3.1). Persisting a
|
|
watch is the user's explicit act via /prefs/watch.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Optional, Sequence
|
|
|
|
from . import data
|
|
|
|
# DEMO sample: ZIP -> the (current) district(s) it maps to, each with its
|
|
# assembly/senate member. Explicitly a placeholder for the authoritative
|
|
# boundary join (D2). Legislators reference the session's real member slugs.
|
|
DEMO_ZIPS = {
|
|
"90210": [{"district_id": "A-14", "type": "ASSEMBLY", "number": 14,
|
|
"assembly_member": {"legislator_id": "schiavo",
|
|
"name": "M. Schiavo", "party": "D"}}],
|
|
"94107": [{"district_id": "A-11", "type": "ASSEMBLY", "number": 11,
|
|
"assembly_member": {"legislator_id": "macedo",
|
|
"name": "K. Macedo", "party": "D"}}],
|
|
"90012": [
|
|
# A ZIP straddling two districts -> AMBIGUOUS (the design's example).
|
|
{"district_id": "S-08", "type": "SENATE", "number": 8,
|
|
"senate_member": {"legislator_id": "chen",
|
|
"name": "W. Chen", "party": "R"}},
|
|
{"district_id": "A-09", "type": "ASSEMBLY", "number": 9,
|
|
"assembly_member": {"legislator_id": "ortega",
|
|
"name": "A. Ortega", "party": "D"}},
|
|
],
|
|
"92101": [{"district_id": "A-32", "type": "ASSEMBLY", "number": 32,
|
|
"assembly_member": {"legislator_id": "bryan",
|
|
"name": "B. Bryan", "party": "D"}}],
|
|
"99501": [{"district_id": "A-55", "type": "ASSEMBLY", "number": 55,
|
|
"assembly_member": {"legislator_id": "jackson",
|
|
"name": "H. Jackson", "party": "R"}}],
|
|
}
|
|
|
|
# 5-digit ZIP prefixes that are obviously not California, so the out-of-state
|
|
# confidence branch is demonstrable.
|
|
_OUT_OF_STATE_PREFIX = ("990", "999")
|
|
|
|
|
|
def _to_resolution(mode: str, districts: Sequence[Dict[str, Any]]) -> Dict[str, Any]:
|
|
suggested_watch = []
|
|
for d in districts:
|
|
for member in (d.get("assembly_member"), d.get("senate_member")):
|
|
if member and member.get("legislator_id"):
|
|
suggested_watch.append(member["legislator_id"])
|
|
return {
|
|
"resolved_from": mode,
|
|
"confidence": "single" if len(districts) == 1
|
|
else ("ambiguous" if districts else "zero"),
|
|
"districts": [dict(d) for d in districts],
|
|
"suggested_topics": _suggested_topics(districts),
|
|
"suggested_watch": suggested_watch,
|
|
}
|
|
|
|
|
|
def _manual_district(district: str) -> Dict[str, Any]:
|
|
d = district.strip()
|
|
typ = d[0].upper() if d else "A"
|
|
num = d[1:]
|
|
type_ = "SENATE" if typ == "S" else "ASSEMBLY"
|
|
out: Dict[str, Any] = {
|
|
"district_id": d,
|
|
"type": type_,
|
|
"number": int(num) if num.isdigit() else None,
|
|
}
|
|
member_key = "senate_member" if type_ == "SENATE" else "assembly_member"
|
|
out[member_key] = {}
|
|
for entry in DEMO_ZIPS.values():
|
|
for cand in entry:
|
|
if cand.get("district_id") == d and cand.get(member_key):
|
|
out[member_key] = dict(cand[member_key])
|
|
return out
|
|
|
|
|
|
def _suggested_topics(districts: Sequence[Dict[str, Any]]) -> List[str]:
|
|
topics: List[str] = []
|
|
seen = set()
|
|
for d in districts:
|
|
for member in (d.get("assembly_member"), d.get("senate_member")):
|
|
if not member:
|
|
continue
|
|
lid = member.get("legislator_id")
|
|
for lg in data.load_legislators():
|
|
if lg.get("legislator_id") == lid:
|
|
for x in lg.get("domains") or []:
|
|
if x not in seen:
|
|
seen.add(x)
|
|
topics.append(x)
|
|
return topics
|
|
|
|
|
|
def resolve(zip: Optional[str] = None, address: Optional[str] = None,
|
|
district: Optional[str] = None) -> Dict[str, Any]:
|
|
# Manual pick: no ZIP ambiguity at all.
|
|
if district and not zip and not address:
|
|
d = _manual_district(district)
|
|
return _to_resolution("manual", [d])
|
|
key = None
|
|
mode = "zip"
|
|
if address:
|
|
a = address.strip()
|
|
key = a[-5:] if len(a) >= 5 else a
|
|
mode = "address"
|
|
elif zip is not None:
|
|
key = str(zip).strip()
|
|
else:
|
|
return {"resolved_from": "zip", "confidence": "zero",
|
|
"districts": [], "suggested_topics": [],
|
|
"suggested_watch": []}
|
|
entry = DEMO_ZIPS.get(key)
|
|
if entry is None:
|
|
confidence = "out-of-state" if key[:3] in _OUT_OF_STATE_PREFIX \
|
|
else "zero"
|
|
return {"resolved_from": mode, "confidence": confidence,
|
|
"districts": [], "suggested_topics": [],
|
|
"suggested_watch": []}
|
|
return _to_resolution(mode, entry)
|