feat(examples): osm-equity — Digital Divide & OSM Mapping Equity study

examples/osm-equity/README.md

@@ -0,0 +1,61 @@
+# osm-equity — Digital Divide & OSM Mapping Equity
+
+A 5-phase FFL study of how socioeconomic disparity shapes the completeness/
+quality of OpenStreetMap data, at census-tract scale. The FFL **is** the
+methodology; the runtime schedules it by data flow.
+
+```
+ffl/osm_equity.ffl        the workflow (namespace osm.equity)
+handlers/                 event-facet implementations (real geometry + stats)
+  shared/equity_utils.py    computation + deterministic offline data source
+  design/ acquisition/ metrics/ analysis/ reporting/
+tests/test_osm_equity.py  FFL-compile + unit + end-to-end signal recovery
+```
+
+## Phases → FFL
+
+| Phase | Plan section | FFL |
+|-------|--------------|-----|
+| 1 | Study design | `DefineStudyArea` + workflow params |
+| 2 | Data acquisition | `ResolveStudyTracts`, `FetchRegionOSM`, `FetchRegionFootprints` (region-level, **update-gated**) |
+| 3 | Metric calculation | per-tract `foreach`: `ClipTractOSM` (local), `FetchCensusEquity`, `ComputeAttributeQuality` (intrinsic), `ComputeExtrinsicQuality` (extrinsic, gated) |
+| 4 | Statistical & spatial | `SpearmanCorrelation`, `MoransI`, `GeographicallyWeightedRegression`, `TemporalEvolution` — run concurrently over the full tract set |
+| 5 | Actionable reporting | `BuildEquityReport` → HTML + choropleth GeoJSON + ranked data deserts |
+
+## Design choices (folded-in improvements)
+
+1. **Fetch OSM once, clip per tract locally.** Region OSM is pulled a single
+   time and each tract is obtained by a local geometry clip — no per-tract
+   remote calls (Overpass/ohsome rate limits). `ClipTractOSM` is pure-local.
+2. **Update-gated fetch.** `FetchRegionOSM` / `FetchRegionFootprints` reuse the
+   cached extract and **only re-fetch when `update = true`**.
+3. **Geometry carried into Phase 4** so Moran's I / GWR build a real spatial-
+   weights matrix (`TractQuality.geometry_wkt`).
+4. **POI diversity measured** (Shannon entropy) and correlated with income.
+5. **Extrinsic benchmark is pluggable + gated:** `ExtrinsicQuality.has_reference`
+   is `false` where no authoritative footprint layer exists — metrics are an
+   explicit N/A sentinel (`-1.0`), never silently zeroed.
+6. **Provenance:** `acs_year` param + a content-derived OSM `snapshot` for
+   reproducible re-runs.
+
+## Data source
+
+Handlers do **real** geometry (shapely) and statistics (scipy Spearman, a
+pure-numpy permutation Moran's I, a Gaussian-kernel local GWR). The data source
+defaults to a deterministic offline generator that tiles the region into a grid
+of synthetic tracts whose income gradient is spatially smooth and drives OSM
+richness — so the whole pipeline runs with **no network** and the analysis
+recovers a real signal. Points marked `TODO(real)` in `equity_utils.py` are
+where a deployment swaps in Census TIGER, the ACS API (`CENSUS_API_KEY`), an
+Overpass/ohsome pull, and Microsoft/Google Open Buildings.
+
+## Run the tests
+
+```bash
+pytest examples/osm-equity/tests -q
+```
+
+The end-to-end test drives every handler in the workflow's data order and
+asserts the built-in digital-divide signal is recovered (Spearman ρ > 0.3,
+p < 0.05; Moran's I > 0) and that data deserts are the low-income tracts.
+Set `FW_EQUITY_GRID` to change the grid resolution (default 6 → 36 tracts).

examples/osm-equity/ffl/osm_equity.ffl

@@ -0,0 +1,197 @@
+namespace osm.equity {
+
+  // ===================================================================
+  // Digital Divide & OSM Mapping Equity
+  //
+  // A 5-phase, census-tract-level study of how socioeconomic disparity
+  // shapes the completeness/quality of volunteered geographic
+  // information (OSM). The data flow *is* the methodology.
+  //
+  // Improvements folded in vs. the first sketch:
+  //   1. Region OSM is fetched ONCE (update-gated cache) and clipped
+  //      per tract LOCALLY - no per-tract remote calls (rate limits).
+  //   2. Tract geometry (WKT) is carried into Phase 4 so Moran's I / GWR
+  //      can build a real spatial-weights matrix.
+  //   3. POI diversity is measured (Phase 4.1 correlates it with income).
+  //   4. The extrinsic benchmark is a pluggable source and is GATED:
+  //      ExtrinsicQuality.has_reference is false where no authoritative
+  //      footprint layer exists (no silent zeroing).
+  //   5. Provenance (acs_year, osm snapshot) is carried for reproducibility.
+  // ===================================================================
+
+  // ---- Analysis unit + data layers ----
+  schema StudyArea {
+    region: String,
+    scale: String,
+    hypothesis: String,
+    acs_year: Long
+  }
+
+  schema TractGeo {
+    geoid: String,
+    name: String,
+    // tract boundary as WKT (kept so spatial stats can build weights)
+    geometry_wkt: String,
+    area_km2: Double
+  }
+
+  // Phase 2 "Key Equity Variables" - census/ACS + digital-access layers.
+  schema EquityVars {
+    geoid: String,
+    median_income: Double,
+    pct_rent_burdened: Double,
+    pct_bipoc: Double,
+    pct_no_hs_diploma: Double,
+    pct_limited_english: Double,
+    pct_zero_vehicle: Double,
+    pct_internet_subscription: Double
+  }
+
+  // Phase 3.1 Intrinsic quality (self-referential OSM attributes).
+  schema AttributeQuality {
+    geoid: String,
+    building_density: Double,
+    road_km_per_km2: Double,
+    attr_completeness: Double,
+    // Shannon diversity of amenity/shop types (POI diversity)
+    poi_diversity: Double,
+    median_recency_days: Double
+  }
+
+  // Phase 3.2 Extrinsic quality (OSM vs authoritative references).
+  schema ExtrinsicQuality {
+    geoid: String,
+    footprint_completeness: Double,
+    positional_accuracy_m: Double,
+    // false => no authoritative reference for this tract (metrics are N/A)
+    has_reference: Boolean
+  }
+
+  // Joined per-tract record. Carries geometry for the spatial analysis.
+  schema TractQuality {
+    geoid: String,
+    geometry_wkt: String,
+    equity: EquityVars,
+    intrinsic: AttributeQuality,
+    extrinsic: ExtrinsicQuality
+  }
+
+  schema StatResults {
+    spearman_rho: Double,
+    spearman_pvalue: Double,
+    morans_i: Double,
+    morans_pvalue: Double,
+    gwr_summary_path: String,
+    gwr_mean_r2: Double,
+    temporal_trend: String,
+    temporal_gap_change: Double
+  }
+
+  // ===================================================================
+  // Phase 1 - Study design.
+  // ===================================================================
+  event facet DefineStudyArea(region: String, scale: String, hypothesis: String, acs_year: Long) => (area: StudyArea)
+
+  // ===================================================================
+  // Phase 2 - Data acquisition. Region-level fetches happen ONCE and are
+  // update-gated: if the extract already exists it is reused unless
+  // `update` is set. Per-tract OSM is obtained by LOCAL clipping.
+  // ===================================================================
+  event facet ResolveStudyTracts(area: StudyArea) => (tracts: [TractGeo])
+  event facet FetchRegionOSM(region: String, update: Boolean) => (osm_path: String, snapshot: String)
+  event facet FetchRegionFootprints(region: String, benchmark_source: String, update: Boolean) => (footprints_path: String, available: Boolean)
+
+  // ===================================================================
+  // Phase 3 - Per-tract metric calculation (local clip + compute).
+  // ===================================================================
+  event facet ClipTractOSM(region_osm_path: String, geometry_wkt: String) => (osm_path: String)
+  event facet FetchCensusEquity(geoid: String, acs_year: Long) => (vars: EquityVars)
+  event facet ComputeAttributeQuality(osm_path: String, area_km2: Double, geoid: String) => (quality: AttributeQuality)
+  event facet ComputeExtrinsicQuality(osm_path: String, geometry_wkt: String, footprints_path: String, reference_available: Boolean) => (quality: ExtrinsicQuality)
+  event facet AssembleTractQuality(geoid: String, geometry_wkt: String, equity: EquityVars, intrinsic: AttributeQuality, extrinsic: ExtrinsicQuality) => (record: TractQuality)
+
+  // Fan-out: one parallel branch per census tract. Each branch clips the
+  // shared region extract locally, pulls census equity, computes intrinsic
+  // + extrinsic quality, and appends one joined TractQuality row.
+  facet CollectTractQuality(tracts: [TractGeo], region_osm_path: String, footprints_path: String, reference_available: Boolean, acs_year: Long) => (records: [TractQuality]) andThen foreach t in $.tracts {
+    clip = ClipTractOSM(region_osm_path = $.region_osm_path, geometry_wkt = $.t.geometry_wkt)
+    eq   = FetchCensusEquity(geoid = $.t.geoid, acs_year = $.acs_year)
+    intr = ComputeAttributeQuality(osm_path = clip.osm_path, area_km2 = $.t.area_km2, geoid = $.t.geoid)
+    extr = ComputeExtrinsicQuality(osm_path = clip.osm_path, geometry_wkt = $.t.geometry_wkt, footprints_path = $.footprints_path, reference_available = $.reference_available)
+    rec  = AssembleTractQuality(geoid = $.t.geoid, geometry_wkt = $.t.geometry_wkt, equity = eq.vars, intrinsic = intr.quality, extrinsic = extr.quality)
+    yield CollectTractQuality(records = [rec.record])
+  }
+
+  // ===================================================================
+  // Phase 4 - Statistical & spatial analysis across all tracts.
+  // The four methods are independent, so they run concurrently.
+  // ===================================================================
+  event facet SpearmanCorrelation(records: [TractQuality]) => (rho: Double, pvalue: Double)
+  event facet MoransI(records: [TractQuality]) => (i: Double, pvalue: Double)
+  event facet GeographicallyWeightedRegression(records: [TractQuality]) => (summary_path: String, mean_r2: Double)
+  event facet TemporalEvolution(records: [TractQuality], snapshot: String) => (trend: String, gap_change: Double)
+  event facet CombineStats(spearman_rho: Double, spearman_p: Double, moran_i: Double, moran_p: Double, gwr_summary: String, gwr_r2: Double, temporal_trend: String, temporal_gap_change: Double) => (results: StatResults)
+
+  facet AnalyzeEquity(records: [TractQuality], snapshot: String) => (results: StatResults) andThen {
+    corr     = SpearmanCorrelation(records = $.records)
+    moran    = MoransI(records = $.records)
+    gwr      = GeographicallyWeightedRegression(records = $.records)
+    temporal = TemporalEvolution(records = $.records, snapshot = $.snapshot)
+    combined = CombineStats(
+      spearman_rho = corr.rho, spearman_p = corr.pvalue,
+      moran_i = moran.i, moran_p = moran.pvalue,
+      gwr_summary = gwr.summary_path, gwr_r2 = gwr.mean_r2,
+      temporal_trend = temporal.trend, temporal_gap_change = temporal.gap_change
+    )
+    yield AnalyzeEquity(results = combined.results)
+  }
+
+  // ===================================================================
+  // Phase 5 - Actionable reporting: data deserts + map + HTML report.
+  // ===================================================================
+  event facet BuildEquityReport(records: [TractQuality], stats: StatResults, title: String) => (report_html: String, map_path: String, data_deserts: [String])
+
+  // ===================================================================
+  // Entry point - chains the five phases end to end.
+  // ===================================================================
+  workflow DigitalDivideMappingEquity(
+      region: String = "Oakland, CA",
+      scale: String = "census_tract",
+      hypothesis: String = "Lower-income, lower-connectivity tracts show lower OSM building completeness and road density.",
+      acs_year: Long = 2022,
+      update: Boolean = false,
+      benchmark_source: String = "microsoft_open_buildings",
+      title: String = "Digital Divide and OSM Mapping Equity"
+  ) => (report_html: String, map_path: String, data_deserts: [String]) andThen {
+
+    // Phase 1
+    area  = DefineStudyArea(region = $.region, scale = $.scale, hypothesis = $.hypothesis, acs_year = $.acs_year)
+
+    // Phase 2 - region-level acquisition (update-gated), run concurrently
+    units = ResolveStudyTracts(area = area.area)
+    osm   = FetchRegionOSM(region = $.region, update = $.update)
+    foot  = FetchRegionFootprints(region = $.region, benchmark_source = $.benchmark_source, update = $.update)
+
+    // Phases 2 + 3 - per-tract clip + metrics (fanned out)
+    perTract = CollectTractQuality(
+      tracts = units.tracts,
+      region_osm_path = osm.osm_path,
+      footprints_path = foot.footprints_path,
+      reference_available = foot.available,
+      acs_year = $.acs_year
+    )
+
+    // Phase 4
+    stats = AnalyzeEquity(records = perTract.records, snapshot = osm.snapshot)
+
+    // Phase 5
+    report = BuildEquityReport(records = perTract.records, stats = stats.results, title = $.title)
+
+    yield DigitalDivideMappingEquity(
+      report_html = report.report_html,
+      map_path = report.map_path,
+      data_deserts = report.data_deserts
+    )
+  }
+
+}

examples/osm-equity/handlers/__init__.py

@@ -0,0 +1,33 @@
+"""osm.equity mapping-equity handlers - registration aggregator."""
+
+from __future__ import annotations
+
+from .acquisition.acquisition_handlers import register_acquisition_handlers
+from .analysis.analysis_handlers import register_analysis_handlers
+from .design.design_handlers import register_design_handlers
+from .metrics.metrics_handlers import register_metrics_handlers
+from .reporting.reporting_handlers import register_reporting_handlers
+
+
+def register_all_handlers(poller) -> None:
+    """Register every handler with an AgentPoller."""
+    register_design_handlers(poller)
+    register_acquisition_handlers(poller)
+    register_metrics_handlers(poller)
+    register_analysis_handlers(poller)
+    register_reporting_handlers(poller)
+
+
+def register_all_registry_handlers(runner) -> None:
+    """Register every handler with a RegistryRunner."""
+    from .acquisition.acquisition_handlers import register_handlers as reg_acq
+    from .analysis.analysis_handlers import register_handlers as reg_ana
+    from .design.design_handlers import register_handlers as reg_des
+    from .metrics.metrics_handlers import register_handlers as reg_met
+    from .reporting.reporting_handlers import register_handlers as reg_rep
+
+    reg_des(runner)
+    reg_acq(runner)
+    reg_met(runner)
+    reg_ana(runner)
+    reg_rep(runner)

examples/osm-equity/handlers/acquisition/acquisition_handlers.py

@@ -0,0 +1,84 @@
+"""Phase 2 - data acquisition: ResolveStudyTracts, FetchRegionOSM,
+FetchRegionFootprints. Region-level fetches are update-gated (reuse the
+cached extract unless `update` is set)."""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from handlers.shared import equity_utils as U
+
+NAMESPACE = "osm.equity"
+
+
+def _truthy(v: Any) -> bool:
+    if isinstance(v, str):
+        return v.strip().lower() in {"1", "true", "yes", "on"}
+    return bool(v)
+
+
+def _log(params, msg, level="success"):
+    sl = params.get("_step_log")
+    if sl is not None:
+        sl.append({"message": msg, "level": level})
+
+
+def handle_resolve_study_tracts(params: dict[str, Any]) -> dict[str, Any]:
+    area = params.get("area", {})
+    if isinstance(area, str):
+        import json
+
+        area = json.loads(area)
+    region = area.get("region", "Oakland, CA")
+    tracts = U.make_tracts(region)
+    _log(params, f"Resolved {len(tracts)} census tracts for {region}")
+    return {"tracts": tracts}
+
+
+def handle_fetch_region_osm(params: dict[str, Any]) -> dict[str, Any]:
+    region = params.get("region", "Oakland, CA")
+    update = _truthy(params.get("update", False))
+    tracts = U.make_tracts(region)
+    path, snapshot, cached = U.fetch_region_osm(region, update, tracts)
+    _log(params, f"Region OSM {'reused (cache)' if cached else 'fetched'} -> {snapshot}")
+    return {"osm_path": path, "snapshot": snapshot}
+
+
+def handle_fetch_region_footprints(params: dict[str, Any]) -> dict[str, Any]:
+    region = params.get("region", "Oakland, CA")
+    source = params.get("benchmark_source", "microsoft_open_buildings")
+    update = _truthy(params.get("update", False))
+    tracts = U.make_tracts(region)
+    path, available = U.fetch_region_footprints(region, source, update, tracts)
+    _log(
+        params,
+        f"Footprint reference '{source}': {'available' if available else 'NOT available (extrinsic metrics will be N/A)'}",
+        level="success" if available else "warning",
+    )
+    return {"footprints_path": path, "available": available}
+
+
+_DISPATCH: dict[str, Any] = {
+    f"{NAMESPACE}.ResolveStudyTracts": handle_resolve_study_tracts,
+    f"{NAMESPACE}.FetchRegionOSM": handle_fetch_region_osm,
+    f"{NAMESPACE}.FetchRegionFootprints": handle_fetch_region_footprints,
+}
+
+
+def handle(payload: dict) -> dict:
+    return _DISPATCH[payload["_facet_name"]](payload)
+
+
+def register_handlers(runner) -> None:
+    for facet_name in _DISPATCH:
+        runner.register_handler(
+            facet_name=facet_name,
+            module_uri=f"file://{os.path.abspath(__file__)}",
+            entrypoint="handle",
+        )
+
+
+def register_acquisition_handlers(poller) -> None:
+    for facet_name, fn in _DISPATCH.items():
+        poller.register(facet_name, fn)