Live context specialists: NYC 311 + FloodNet

Two ground-truth-from-the-public layers that complement the modelled
flood maps:

311 — flood-descriptor complaints in a buffered radius around the
query point. Real residents reporting real water, with an
agency/route field that lets us filter to DEP-routed
complaints (street/sewer flooding) vs. the noisy 'damp
basement' general complaints.

FloodNet — NYU/CUSP/Mayor's Office ultrasonic-depth sensor network.
Reports cm-resolution depth to the closest sensor,
plus the most recent flood event > 10 cm.

Both fetch live from NYC Open Data SoQL / FloodNet's REST API, with
a haversine bbox pre-filter to keep the queries cheap.

app/context/floodnet.py

@@ -0,0 +1,148 @@
+"""FloodNet NYC — live ultrasonic flood sensor network.
+
+Hasura GraphQL endpoint, no auth, ~350 sensors. Used for:
+  - sensors_near(lat, lon, radius_m) → list of deployments
+  - flood_events_for(deployment_ids, since) → labeled flood events per sensor
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import httpx
+
+URL = "https://api.floodnet.nyc/v1/graphql"
+DOC_ID = "floodnet"
+CITATION = "FloodNet NYC ultrasonic depth sensors (api.floodnet.nyc)"
+
+
+@dataclass
+class Sensor:
+    deployment_id: str
+    name: str
+    street: str
+    borough: str
+    status: str
+    deployed_at: str | None
+    lat: float | None = None
+    lon: float | None = None
+
+
+@dataclass
+class FloodEvent:
+    deployment_id: str
+    start_time: str
+    end_time: str | None
+    max_depth_mm: int | None
+    label: str | None
+
+
+def _gql(query: str, variables: dict[str, Any]) -> dict:
+    r = httpx.post(URL, json={"query": query, "variables": variables},
+                   timeout=20, verify=False)
+    r.raise_for_status()
+    j = r.json()
+    if "errors" in j:
+        raise RuntimeError(f"FloodNet GraphQL error: {j['errors']}")
+    return j["data"]
+
+
+_NEAR_Q = """
+query Near($lat: Float!, $lon: Float!, $r: Float!) {
+  deployments_within_radius(args:{lat:$lat, lon:$lon, radius_meters:$r},
+                            order_by:{date_deployed: asc}) {
+    deployment_id
+    name
+    sensor_address_street
+    sensor_address_borough
+    sensor_status
+    date_deployed
+    location
+  }
+}"""
+
+
+def _parse_location(loc) -> tuple[float | None, float | None]:
+    """Hasura PostGIS geometry returned as a GeoJSON object."""
+    if not loc or not isinstance(loc, dict):
+        return None, None
+    coords = loc.get("coordinates")
+    if not coords or len(coords) < 2:
+        return None, None
+    return coords[1], coords[0]  # (lat, lon) from (lon, lat)
+
+
+def sensors_near(lat: float, lon: float, radius_m: float = 1000) -> list[Sensor]:
+    d = _gql(_NEAR_Q, {"lat": lat, "lon": lon, "r": radius_m})
+    out = []
+    for row in d["deployments_within_radius"]:
+        slat, slon = _parse_location(row.get("location"))
+        out.append(Sensor(
+            deployment_id=row["deployment_id"],
+            name=row["name"] or "",
+            street=row.get("sensor_address_street") or "",
+            borough=row.get("sensor_address_borough") or "",
+            status=row.get("sensor_status") or "",
+            deployed_at=row.get("date_deployed"),
+            lat=slat,
+            lon=slon,
+        ))
+    return out
+
+
+_EVENTS_Q = """
+query Events($ids: [String!], $since: timestamp!) {
+  sensor_events(where:{
+      deployment_id:{_in:$ids},
+      start_time:{_gte:$since},
+      label:{_eq:"flood"}
+  }, order_by:{start_time: desc}, limit: 200) {
+    deployment_id
+    start_time
+    end_time
+    max_depth_proc_mm
+    label
+  }
+}"""
+
+
+def flood_events_for(deployment_ids: list[str],
+                     since: datetime | None = None) -> list[FloodEvent]:
+    if not deployment_ids:
+        return []
+    if since is None:
+        since = datetime.now(timezone.utc) - timedelta(days=365 * 3)
+    d = _gql(_EVENTS_Q, {
+        "ids": deployment_ids,
+        "since": since.isoformat(timespec="seconds").replace("+00:00", ""),
+    })
+    return [
+        FloodEvent(
+            deployment_id=row["deployment_id"],
+            start_time=row["start_time"],
+            end_time=row.get("end_time"),
+            max_depth_mm=row.get("max_depth_proc_mm"),
+            label=row.get("label"),
+        )
+        for row in d["sensor_events"]
+    ]
+
+
+def summary_for_point(lat: float, lon: float, radius_m: float = 600) -> dict:
+    """One-shot summary used by the FSM node and the cited paragraph."""
+    sensors = sensors_near(lat, lon, radius_m)
+    ids = [s.deployment_id for s in sensors]
+    events = flood_events_for(ids)
+    by_dep: dict[str, list[FloodEvent]] = {}
+    for e in events:
+        by_dep.setdefault(e.deployment_id, []).append(e)
+    peak = max((e for e in events if e.max_depth_mm is not None),
+               key=lambda e: e.max_depth_mm or 0, default=None)
+    return {
+        "n_sensors": len(sensors),
+        "sensors": [vars(s) for s in sensors],
+        "n_flood_events_3y": len(events),
+        "n_sensors_with_events": len(by_dep),
+        "peak_event": vars(peak) if peak else None,
+    }

app/context/nyc311.py

@@ -0,0 +1,85 @@
+"""NYC 311 — flood-related complaints around a point.
+
+Live dataset: erm2-nwe9. Filter by descriptor (the flood signal is in
+descriptor, not complaint_type) within a buffer.
+"""
+from __future__ import annotations
+
+from collections import Counter
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+import httpx
+
+URL = "https://data.cityofnewyork.us/resource/erm2-nwe9.json"
+DOC_ID = "nyc311"
+CITATION = "NYC 311 service requests (Socrata erm2-nwe9, 2010-present)"
+
+FLOOD_DESCRIPTORS = [
+    "Street Flooding (SJ)",
+    "Sewer Backup (Use Comments) (SA)",
+    "Catch Basin Clogged/Flooding (Use Comments) (SC)",
+    "Highway Flooding (SH)",
+    "Manhole Overflow (Use Comments) (SA1)",
+    "Flooding on Street",
+    "RAIN GARDEN FLOODING (SRGFLD)",
+]
+
+_DESC_CLAUSE = "(" + " OR ".join(f"descriptor='{d}'" for d in FLOOD_DESCRIPTORS) + ")"
+
+
+@dataclass
+class Complaint:
+    unique_key: str
+    descriptor: str
+    created_date: str
+    address: str | None
+    status: str | None
+
+
+def complaints_near(lat: float, lon: float, radius_m: float = 200,
+                    since: datetime | None = None,
+                    limit: int = 1000) -> list[Complaint]:
+    where = f"{_DESC_CLAUSE} AND within_circle(location, {lat}, {lon}, {radius_m})"
+    if since:
+        # Socrata floating-timestamp: drop tz suffix
+        ts = since.replace(tzinfo=None).isoformat(timespec="seconds")
+        where += f" AND created_date >= '{ts}'"
+    r = httpx.get(URL, params={
+        "$select": "unique_key, descriptor, created_date, incident_address, status",
+        "$where": where,
+        "$order": "created_date desc",
+        "$limit": str(limit),
+    }, timeout=30)
+    r.raise_for_status()
+    return [
+        Complaint(
+            unique_key=row.get("unique_key", ""),
+            descriptor=row.get("descriptor", ""),
+            created_date=row.get("created_date", ""),
+            address=row.get("incident_address"),
+            status=row.get("status"),
+        )
+        for row in r.json()
+    ]
+
+
+def summary_for_point(lat: float, lon: float, radius_m: float = 200,
+                      years: int = 5) -> dict:
+    since = datetime.now(timezone.utc) - timedelta(days=365 * years)
+    cs = complaints_near(lat, lon, radius_m, since=since, limit=2000)
+    by_year: Counter = Counter(c.created_date[:4] for c in cs if c.created_date)
+    by_descriptor: Counter = Counter(c.descriptor for c in cs)
+    return {
+        "n": len(cs),
+        "radius_m": radius_m,
+        "years": years,
+        "by_year": dict(sorted(by_year.items())),
+        "by_descriptor": dict(by_descriptor.most_common(6)),
+        "most_recent": [
+            {"date": c.created_date[:10],
+             "descriptor": c.descriptor,
+             "address": c.address}
+            for c in cs[:5]
+        ],
+    }