From 1dcd2af1ce43c696ecef1acf043384a6e2433132 Mon Sep 17 00:00:00 2001 From: rlemke Date: Mon, 13 Jul 2026 19:44:26 -0700 Subject: [PATCH] =?UTF-8?q?feat(osm-equity):=20MappingEquityAtlas=20FFL=20?= =?UTF-8?q?=E2=80=94=20fan=20out=20over=20cities,=20one=20map?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A single parameterised FFL workflow that turns the study into: resolve a sequence of regions + a min-population ("city size") floor into cities, run each city in parallel (andThen foreach — fan-out by subregion), tile each at a chosen size, and MERGE every tile into ONE combined map. FFL (osm_equity.ffl): schema CityRef; event facets ResolveCities / CityTiles / BuildAtlasMap; fan-out facet CollectCityTiles (foreach city -> CityTiles -> aggregate records); workflow MappingEquityAtlas with input params regions, min_population, tile_sqmi, acs_year, osm_kinds, with_footprints, metric, half_deg, update, title. Validates clean under relative scoping. Handlers (handlers/atlas/): ResolveCities (curated US city table filtered by population + region), CityTiles (the fan-out unit — real: TIGER+ACS areal- interpolated onto tiles, OSM by kinds, optional MS footprints; offline: deterministic synthetic tiles), BuildAtlasMap (one combined MapLibre map by `metric`). 2 offline atlas tests (fan-out + combine, population floor); 18 offline tests pass; ruff clean. Co-Authored-By: Claude Opus 4.8 --- examples/osm-equity/README.md | 14 + examples/osm-equity/ffl/osm_equity.ffl | 73 +++++ examples/osm-equity/handlers/__init__.py | 4 + .../osm-equity/handlers/atlas/__init__.py | 0 .../handlers/atlas/atlas_handlers.py | 303 ++++++++++++++++++ examples/osm-equity/tests/test_osm_equity.py | 43 +++ 6 files changed, 437 insertions(+) create mode 100644 examples/osm-equity/handlers/atlas/__init__.py create mode 100644 examples/osm-equity/handlers/atlas/atlas_handlers.py diff --git a/examples/osm-equity/README.md b/examples/osm-equity/README.md index 7430ab7b..3701d5ba 100644 --- a/examples/osm-equity/README.md +++ b/examples/osm-equity/README.md @@ -71,3 +71,17 @@ 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). + +## Atlas workflow — fan out over cities, one map + +`MappingEquityAtlas` (in `ffl/osm_equity.ffl`) turns the whole study into a +single parameterised workflow: it resolves a **sequence of regions** + a +**city-size** floor (min population) into cities, runs each city in parallel +(`andThen foreach` — the fan-out is by subregion/city), tiles each at a chosen +**tile size**, and merges every tile into **one** combined map. + +Parameters: `regions` (e.g. `["north-america"]`), `min_population`, `tile_sqmi`, +`acs_year`, `osm_kinds` (`poi` for large sweeps, `building,road,poi` for detail), +`with_footprints`, `metric` (which per-tile metric colours the map), `half_deg`, +`update`, `title`. Handlers: `ResolveCities` / `CityTiles` (fan-out unit) / +`BuildAtlasMap`, in `handlers/atlas/`. diff --git a/examples/osm-equity/ffl/osm_equity.ffl b/examples/osm-equity/ffl/osm_equity.ffl index 624bbade..de89c19d 100644 --- a/examples/osm-equity/ffl/osm_equity.ffl +++ b/examples/osm-equity/ffl/osm_equity.ffl @@ -194,4 +194,77 @@ namespace osm.equity { ) } + // =================================================================== + // ATLAS - fan out over cities (subregions), tile each, combine to ONE map. + // + // Parameters let the caller choose the region sequence, the "city size" + // (min population), the tile size, and how the OSM/quality layers are + // built; the workflow resolves the region(s) into cities, runs each city + // in parallel (andThen foreach), and merges every tile into a single map. + // =================================================================== + schema CityRef { + name: String, + lon: Double, + lat: Double, + half_deg: Double, + population: Long + } + + // Expand a sequence of regions + a population floor into cities to map. + event facet ResolveCities(regions: [String], min_population: Long, half_deg: Double) => (cities: [CityRef]) + + // Per-city tiled analysis (the fan-out unit): TIGER tracts + ACS equity + // areal-interpolated onto a `tile_sqmi` grid, OSM (kinds) clipped per tile, + // optional Microsoft-footprint extrinsic check. Returns the city's tiles. + event facet CityTiles(city: CityRef, tile_sqmi: Double, acs_year: Long, osm_kinds: String, with_footprints: Boolean, update: Boolean) => (records: [TractQuality]) + + // Render ONE combined MapLibre map of every tile, coloured by `metric`. + event facet BuildAtlasMap(records: [TractQuality], stats: StatResults, metric: String, title: String) => (map_html: String, tile_count: Long, city_count: Long) + + // Fan-out: one parallel branch per city; every city's tiles aggregate into + // the shared `records` collection. + facet CollectCityTiles(cities: [CityRef], tile_sqmi: Double, acs_year: Long, osm_kinds: String, with_footprints: Boolean, update: Boolean) => (records: [TractQuality]) andThen foreach c in $.cities { + ct = CityTiles(city = $.c, tile_sqmi = $.tile_sqmi, acs_year = $.acs_year, osm_kinds = $.osm_kinds, with_footprints = $.with_footprints, update = $.update) + yield CollectCityTiles(records = ct.records) + } + + workflow MappingEquityAtlas( + regions: [String] = ["north-america"], + min_population: Long = 500000, + tile_sqmi: Double = 2.0, + acs_year: Long = 2022, + osm_kinds: String = "poi", + with_footprints: Boolean = false, + metric: String = "attr_completeness", + half_deg: Double = 0.12, + update: Boolean = false, + title: String = "OSM Mapping Equity Atlas - North America" + ) => (map_html: String, tile_count: Long, city_count: Long) andThen { + + // resolve the region sequence + city-size floor into cities + cities = ResolveCities(regions = $.regions, min_population = $.min_population, half_deg = $.half_deg) + + // fan out: tile every city in parallel, aggregate all tiles + tiles = CollectCityTiles( + cities = cities.cities, + tile_sqmi = $.tile_sqmi, + acs_year = $.acs_year, + osm_kinds = $.osm_kinds, + with_footprints = $.with_footprints, + update = $.update + ) + + // national analysis over the combined tiles + stats = AnalyzeEquity(records = tiles.records, snapshot = "atlas") + + // ONE combined map + atlas = BuildAtlasMap(records = tiles.records, stats = stats.results, metric = $.metric, title = $.title) + + yield MappingEquityAtlas( + map_html = atlas.map_html, + tile_count = atlas.tile_count, + city_count = atlas.city_count + ) + } + } diff --git a/examples/osm-equity/handlers/__init__.py b/examples/osm-equity/handlers/__init__.py index 21fd20da..ce1f3cb1 100644 --- a/examples/osm-equity/handlers/__init__.py +++ b/examples/osm-equity/handlers/__init__.py @@ -4,6 +4,7 @@ from .acquisition.acquisition_handlers import register_acquisition_handlers from .analysis.analysis_handlers import register_analysis_handlers +from .atlas.atlas_handlers import register_atlas_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 @@ -16,12 +17,14 @@ def register_all_handlers(poller) -> None: register_metrics_handlers(poller) register_analysis_handlers(poller) register_reporting_handlers(poller) + register_atlas_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 .atlas.atlas_handlers import register_handlers as reg_atlas 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 @@ -31,3 +34,4 @@ def register_all_registry_handlers(runner) -> None: reg_met(runner) reg_ana(runner) reg_rep(runner) + reg_atlas(runner) diff --git a/examples/osm-equity/handlers/atlas/__init__.py b/examples/osm-equity/handlers/atlas/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/examples/osm-equity/handlers/atlas/atlas_handlers.py b/examples/osm-equity/handlers/atlas/atlas_handlers.py new file mode 100644 index 00000000..e746676c --- /dev/null +++ b/examples/osm-equity/handlers/atlas/atlas_handlers.py @@ -0,0 +1,303 @@ +"""Atlas phase - fan out over cities, tile each, combine to ONE map: +ResolveCities, CityTiles, BuildAtlasMap. + +Real mode (FW_EQUITY_SOURCE=real) tiles each US city from live TIGER/ACS/OSM +(+ optional Microsoft footprints); offline mode synthesises deterministic tiles +so the FFL fan-out + combine can be tested with no network.""" + +from __future__ import annotations + +import hashlib +import json +import os +from typing import Any + +import numpy as np + +from handlers.shared import equity_utils as U +from handlers.shared import sources_real as R + +NAMESPACE = "osm.equity" + +# US incorporated places (name, lon, lat, 2020 population). Real ACS/OSM are +# US-scoped, so "north america" resolves to these US cities. +_CITY_TABLE = [ + ("New York, NY", -74.02, 40.70, 8804190), + ("Los Angeles, CA", -118.30, 34.03, 3898747), + ("Chicago, IL", -87.68, 41.86, 2746388), + ("Houston, TX", -95.39, 29.75, 2304580), + ("Phoenix, AZ", -112.07, 33.46, 1608139), + ("Philadelphia, PA", -75.16, 39.95, 1603797), + ("San Antonio, TX", -98.50, 29.44, 1434625), + ("San Diego, CA", -117.15, 32.72, 1386932), + ("Dallas, TX", -96.80, 32.78, 1304379), + ("San Jose, CA", -121.88, 37.32, 1013240), + ("Austin, TX", -97.74, 30.27, 961855), + ("Jacksonville, FL", -81.66, 30.33, 949611), + ("Fort Worth, TX", -97.33, 32.75, 918915), + ("Columbus, OH", -82.99, 39.97, 905748), + ("Charlotte, NC", -80.84, 35.23, 874579), + ("San Francisco, CA", -122.43, 37.77, 873965), + ("Indianapolis, IN", -86.15, 39.77, 887642), + ("Seattle, WA", -122.33, 47.61, 737015), + ("Denver, CO", -104.99, 39.74, 715522), + ("Washington, DC", -77.03, 38.90, 689545), + ("Boston, MA", -71.06, 42.35, 675647), + ("El Paso, TX", -106.44, 31.77, 678815), + ("Nashville, TN", -86.78, 36.17, 689447), + ("Oklahoma City, OK", -97.52, 35.47, 681054), + ("Las Vegas, NV", -115.14, 36.17, 641903), + ("Detroit, MI", -83.06, 42.35, 639111), + ("Portland, OR", -122.66, 45.52, 652503), + ("Memphis, TN", -90.03, 35.13, 633104), + ("Louisville, KY", -85.76, 38.25, 633045), + ("Baltimore, MD", -76.61, 39.29, 585708), + ("Milwaukee, WI", -87.94, 43.04, 577222), + ("Albuquerque, NM", -106.62, 35.10, 564559), + ("Tucson, AZ", -110.94, 32.22, 542629), + ("Fresno, CA", -119.78, 36.74, 542107), + ("Sacramento, CA", -121.49, 38.58, 524943), + ("Mesa, AZ", -111.83, 33.42, 504258), + ("Kansas City, MO", -94.58, 39.10, 508090), + ("Atlanta, GA", -84.39, 33.75, 498715), +] +_US_REGIONS = {"north america", "north-america", "united states", "usa", "us", "u.s."} + + +def _log(params, msg, level="success"): + sl = params.get("_step_log") + if sl is not None: + sl.append({"message": msg, "level": level}) + + +def _truthy(v: Any) -> bool: + return v.strip().lower() in {"1", "true", "yes", "on"} if isinstance(v, str) else bool(v) + + +def _d(v): + return json.loads(v) if isinstance(v, str) else v + + +# --------------------------------------------------------------------------- +# ResolveCities +# --------------------------------------------------------------------------- +def handle_resolve_cities(params: dict[str, Any]) -> dict[str, Any]: + regions = [r.strip().lower() for r in _d(params.get("regions", ["north-america"]))] + min_pop = int(params.get("min_population", 500000)) + half = float(params.get("half_deg", 0.12)) + want_all = any(r in _US_REGIONS for r in regions) + cities = [] + for name, lon, lat, pop in _CITY_TABLE: + if pop < min_pop: + continue + if want_all or any(r in name.lower() for r in regions): + cities.append( + {"name": name, "lon": lon, "lat": lat, "half_deg": half, "population": pop} + ) + _log(params, f"Resolved {len(cities)} cities (>= {min_pop:,} pop) from {regions}") + return {"cities": cities} + + +# --------------------------------------------------------------------------- +# CityTiles - the fan-out unit +# --------------------------------------------------------------------------- +def _bbox(city): + h = float(city["half_deg"]) + return (city["lon"] - h, city["lat"] - h, city["lon"] + h, city["lat"] + h) + + +def _offline_city_tiles(city, tiles): + """Deterministic synthetic tiles (income gradient drives OSM richness).""" + recs, n = [], max(1, len(tiles) - 1) + for i, tile in enumerate(tiles): + seed = int(hashlib.sha256(f"{city['name']}:{tile['geoid']}".encode()).hexdigest()[:8], 16) + rng = np.random.default_rng(seed) + wealth = min(1.0, max(0.0, i / n + float(rng.normal(0, 0.1)))) + eq = { + "geoid": tile["geoid"], + "median_income": round(30000 + wealth * 140000, 1), + "pct_rent_burdened": round(0.5 - wealth * 0.3, 4), + "pct_bipoc": round(0.7 - wealth * 0.4, 4), + "pct_no_hs_diploma": round(0.3 - wealth * 0.2, 4), + "pct_limited_english": round(0.2 - wealth * 0.15, 4), + "pct_zero_vehicle": round(0.3 - wealth * 0.2, 4), + "pct_internet_subscription": round(0.5 + wealth * 0.45, 4), + } + intr = { + "geoid": tile["geoid"], + "building_density": round(20 + wealth * 300, 3), + "road_km_per_km2": round(2 + wealth * 12, 4), + "attr_completeness": round( + min(1.0, max(0.0, 0.2 + wealth * 0.6 + float(rng.normal(0, 0.04)))), 4 + ), + "poi_diversity": round( + min(1.0, max(0.0, wealth * 0.85 + float(rng.normal(0, 0.04)))), 4 + ), + "median_recency_days": round(400 * (1.4 - wealth), 1), + } + recs.append( + { + "geoid": f"{U._slug(city['name'])}:{tile['geoid']}", + "geometry_wkt": tile["geometry_wkt"], + "equity": eq, + "intrinsic": intr, + "extrinsic": { + "geoid": tile["geoid"], + "footprint_completeness": -1.0, + "positional_accuracy_m": -1.0, + "has_reference": False, + }, + } + ) + return recs + + +def handle_city_tiles(params: dict[str, Any]) -> dict[str, Any]: + city = _d(params["city"]) + tile_sqmi = float(params.get("tile_sqmi", 2.0)) + acs_year = int(params.get("acs_year", 2022)) + osm_kinds = params.get("osm_kinds", "poi") + with_fp = _truthy(params.get("with_footprints", False)) + slug = U._slug(city["name"]) + bbox = _bbox(city) + tiles = R.build_tiles(bbox, sqmi=tile_sqmi) + + if U.equity_mode() != "real": + recs = _offline_city_tiles(city, tiles) + _log(params, f"{city['name']}: {len(recs)} synthetic tiles") + return {"records": recs} + + # real: TIGER tracts + ACS -> areal-interpolate equity onto tiles + key = os.environ.get("CENSUS_API_KEY", "") + tract_records = [] + for t in R.real_resolve_tracts(bbox): + try: + eq = R.real_fetch_census_equity(t["geoid"], acs_year, key) + except RuntimeError: + continue + tract_records.append({"geometry_wkt": t["geometry_wkt"], "equity": eq}) + tile_eq = R.areal_interpolate_equity(tiles, tract_records) + + # real OSM (kinds) once for the city, cached; footprints optional + prev = os.environ.get("FW_EQUITY_OSM_KINDS") + os.environ["FW_EQUITY_OSM_KINDS"] = osm_kinds + try: + osm_path = U.out_dir() / f"atlas_osm_{slug}.geojson" + if not osm_path.exists(): + feats, _snap = R.real_fetch_region_osm(bbox) + osm_path.write_text(json.dumps({"type": "FeatureCollection", "features": feats})) + finally: + if prev is None: + os.environ.pop("FW_EQUITY_OSM_KINDS", None) + else: + os.environ["FW_EQUITY_OSM_KINDS"] = prev + + fp_path, fp_ok = "", False + if with_fp: + fp_file = U.out_dir() / f"atlas_fp_{slug}.geojson" + if not fp_file.exists(): + ff = R.real_fetch_footprints(bbox, str(U.out_dir())) + if ff: + fp_file.write_text(json.dumps({"type": "FeatureCollection", "features": ff})) + if fp_file.exists(): + fp_path, fp_ok = str(fp_file), True + + recs = [] + for tile in tiles: + eq = tile_eq.get(tile["geoid"]) + if not eq: + continue + clip = U.clip_tract_osm(str(osm_path), tile["geometry_wkt"]) + intr = U.compute_attribute_quality(clip, tile["area_km2"], tile["geoid"]) + extr = U.compute_extrinsic_quality( + clip, tile["geometry_wkt"], fp_path, fp_ok, tile["geoid"] + ) + recs.append( + { + "geoid": f"{slug}:{tile['geoid']}", + "geometry_wkt": tile["geometry_wkt"], + "equity": eq, + "intrinsic": intr, + "extrinsic": extr, + } + ) + U._clip_index.pop(str(osm_path), None) + _log(params, f"{city['name']}: {len(recs)} tiles (real, footprints={fp_ok})") + return {"records": recs} + + +# --------------------------------------------------------------------------- +# BuildAtlasMap - one combined map +# --------------------------------------------------------------------------- +_METRIC_LABEL = { + "attr_completeness": "Attribute completeness", + "footprint_completeness": "OSM building completeness vs Microsoft", + "poi_diversity": "POI diversity", + "building_density": "Building density (per km2)", +} + + +def handle_build_atlas_map(params: dict[str, Any]) -> dict[str, Any]: + from shapely import wkt as W + from shapely.geometry import mapping + + records = _d(params.get("records", [])) + stats_res = _d(params.get("stats", {})) + metric = params.get("metric", "attr_completeness") + title = params.get("title", "OSM Mapping Equity Atlas") + deserts = U.data_deserts(records) + feats = [] + for r in records: + p = { + "geoid": r["geoid"], + "city": r["geoid"].split(":")[0], + "median_income": r["equity"]["median_income"], + "attr_completeness": r["intrinsic"]["attr_completeness"], + "poi_diversity": r["intrinsic"]["poi_diversity"], + "building_density": r["intrinsic"]["building_density"], + "footprint_completeness": r["extrinsic"]["footprint_completeness"], + "has_reference": r["extrinsic"]["has_reference"], + "data_desert": r["geoid"] in deserts, + } + feats.append( + {"type": "Feature", "properties": p, "geometry": mapping(W.loads(r["geometry_wkt"]))} + ) + html = U.render_maplibre_map( + feats, + title, + stats_res, + deserts, + synthetic=U.equity_mode() != "real", + metric=metric, + metric_label=_METRIC_LABEL.get(metric, metric), + ) + map_path = U.out_dir() / "atlas_map.html" + map_path.write_text(html) + city_count = len({r["geoid"].split(":")[0] for r in records}) + _log(params, f"Atlas map: {len(records)} tiles across {city_count} cities") + return {"map_html": str(map_path), "tile_count": len(records), "city_count": city_count} + + +_DISPATCH: dict[str, Any] = { + f"{NAMESPACE}.ResolveCities": handle_resolve_cities, + f"{NAMESPACE}.CityTiles": handle_city_tiles, + f"{NAMESPACE}.BuildAtlasMap": handle_build_atlas_map, +} + + +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_atlas_handlers(poller) -> None: + for facet_name, fn in _DISPATCH.items(): + poller.register(facet_name, fn) diff --git a/examples/osm-equity/tests/test_osm_equity.py b/examples/osm-equity/tests/test_osm_equity.py index ae3de718..8af4997a 100644 --- a/examples/osm-equity/tests/test_osm_equity.py +++ b/examples/osm-equity/tests/test_osm_equity.py @@ -154,6 +154,49 @@ def _equity(income): } +# --------------------------------------------------------------------------- +# 2c. Atlas: fan out over cities, combine to ONE map (offline) +# --------------------------------------------------------------------------- +class TestAtlas: + def test_atlas_fanout_and_combine(self, tmp_path, monkeypatch): + monkeypatch.setenv("FW_CACHE_ROOT", str(tmp_path)) + from handlers.atlas import atlas_handlers as A + + cities = A.handle_resolve_cities( + {"regions": ["north-america"], "min_population": 500000, "half_deg": 0.05} + )["cities"] + assert len(cities) >= 30 and all(c["population"] >= 500000 for c in cities) + + # fan-out (offline synthetic tiles) over the first 3 cities, aggregate + records = [] + for c in cities[:3]: + recs = A.handle_city_tiles({"city": c, "tile_sqmi": 2.0, "acs_year": 2022})["records"] + assert recs and all(r["geoid"].startswith(A.U._slug(c["name"])) for r in recs) + records.extend(recs) + + # combine to ONE map + out = A.handle_build_atlas_map( + { + "records": records, + "stats": {"spearman_rho": 0.5}, + "metric": "attr_completeness", + "title": "Atlas test", + } + ) + assert out["city_count"] == 3 + assert out["tile_count"] == len(records) + assert os.path.exists(out["map_html"]) + assert "maplibre" in open(out["map_html"]).read().lower() + + def test_resolve_cities_population_floor(self, tmp_path, monkeypatch): + monkeypatch.setenv("FW_CACHE_ROOT", str(tmp_path)) + from handlers.atlas import atlas_handlers as A + + big = A.handle_resolve_cities({"regions": ["us"], "min_population": 1000000})["cities"] + assert all(c["population"] >= 1000000 for c in big) + assert len(big) < 15 # only the largest + + # --------------------------------------------------------------------------- # 3. End-to-end: drive every handler in the workflow's data order # ---------------------------------------------------------------------------