feat(data): POI build pipeline + chunked binary (PR 2 of 5)

.github/workflows/pages.yml

@@ -29,6 +29,9 @@ jobs:
       - name: Assemble walk_graph.bin from chunks
         run: bash scripts/assemble-walk-graph.sh
 
+      - name: Assemble pois.bin from chunks
+        run: bash scripts/assemble-pois.sh
+
       - uses: dtolnay/rust-toolchain@stable
         with:
           targets: wasm32-unknown-unknown

.github/workflows/pr-checks.yml

@@ -16,3 +16,18 @@ jobs:
           node-version: "20"
       - name: Run node --test unit tests
         run: node --test tests/unit/**/*.test.mjs
+
+  pipeline-tests:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install uv
+        uses: astral-sh/setup-uv@v3
+      - name: Assemble walk_graph.bin (needed by walk_graph_reader test)
+        run: bash scripts/assemble-walk-graph.sh
+      - name: Install pipeline deps (incl. dev for pytest)
+        run: uv sync --extra dev
+        working-directory: pipelines
+      - name: Run pytest
+        run: uv run pytest -v
+        working-directory: pipelines

.gitignore

@@ -15,6 +15,7 @@ Cargo.lock
 
 # Built routing blobs (reassembled from tiles/walk-graph.part-* by CI / dev server)
 tiles/walk_graph.bin
+tiles/pois.bin
 
 # Raw OSM extracts (downloaded by build-walk-graph.sh, not committed)
 pipelines/cache/

pipelines/poi_emit.py

@@ -0,0 +1,94 @@
+"""
+Pack POIs into the `tiles/pois.bin` binary format documented in the spec.
+
+Header (24 bytes, little-endian):
+  magic              : 4 bytes = "POI1"
+  version            : u32 = 1
+  walk_graph_version : u32
+  n_pois             : u32
+  names_off          : u32   (byte offset to NAMES section)
+  reserved           : u32
+
+Records: n_pois × 20 bytes, fixed stride. lon/lat as i32×1e7.
+Names: variable-length UTF-8, no terminator. Records index by (name_off, name_len).
+"""
+from __future__ import annotations
+
+import struct
+from dataclasses import dataclass
+
+MAGIC = b"POI1"
+HEADER_FMT = "<4sIIIII"
+HEADER_SIZE = struct.calcsize(HEADER_FMT)  # 24
+RECORD_FMT = "<iiIIHBB"
+RECORD_SIZE = struct.calcsize(RECORD_FMT)  # 20
+MAX_NAME_LEN = 200
+FLAG_UNNAMED = 0x01
+
+
+@dataclass
+class POI:
+    lon: float
+    lat: float
+    walk_node: int
+    category: int  # 1..10
+    name: str = ""
+
+
+def write_poi_blob(pois: list[POI], walk_graph_version: int) -> bytes:
+    names_buf = bytearray()
+    name_index: dict[str, tuple[int, int]] = {}
+
+    def store_name(name: str) -> tuple[int, int]:
+        if not name:
+            return (0, 0)
+        if name in name_index:
+            return name_index[name]
+        encoded = name.encode("utf-8")
+        if len(encoded) > MAX_NAME_LEN:
+            raise ValueError(f"name too long ({len(encoded)} bytes): {name!r}")
+        off = len(names_buf)
+        names_buf.extend(encoded)
+        name_index[name] = (off, len(encoded))
+        return (off, len(encoded))
+
+    records = bytearray()
+    for p in pois:
+        off, n_len = store_name(p.name)
+        flags = 0 if p.name else FLAG_UNNAMED
+        records.extend(struct.pack(
+            RECORD_FMT,
+            int(round(p.lon * 1e7)),
+            int(round(p.lat * 1e7)),
+            p.walk_node,
+            off,
+            n_len,
+            p.category,
+            flags,
+        ))
+
+    names_off = HEADER_SIZE + len(records)
+    header = struct.pack(
+        HEADER_FMT,
+        MAGIC,
+        1,
+        walk_graph_version,
+        len(pois),
+        names_off,
+        0,
+    )
+    return bytes(header + records + names_buf)
+
+
+def read_poi_header(blob: bytes) -> dict:
+    magic, version, walk_graph_version, n_pois, names_off, _ = struct.unpack_from(
+        HEADER_FMT, blob, 0
+    )
+    if magic != MAGIC:
+        raise ValueError(f"bad magic: {magic!r}")
+    return {
+        "version": version,
+        "walk_graph_version": walk_graph_version,
+        "n_pois": n_pois,
+        "names_off": names_off,
+    }

pipelines/pois.py

@@ -0,0 +1,223 @@
+"""
+Extract POIs from an OSM PBF and emit `pois.bin`.
+
+    uv run python pipelines/pois.py <input.osm.pbf> <walk_graph.bin> <output.bin>
+
+Pipeline:
+  1. Pass 0 (ways): record way-node refs for ways with category tags.
+  2. Pass 1 (nodes): resolve coords; emit standalone-node POIs immediately,
+     and collect coords for way nodes we'll need for centroids.
+  3. Compute way centroids (bbox center) for buffered ways.
+  4. Pre-snap each POI to a walk-graph node in the LCC; drop unsnappable.
+  5. Clip to bbox.
+  6. Emit pois.bin (sorted by (category, name) for cache-friendly category scans).
+"""
+from __future__ import annotations
+
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+
+import osmium
+
+# Dual-import: package-style for pytest (runs from repo root with pipelines/
+# on sys.path), sibling-style for direct script invocation matching the
+# walk_graph.py convention (`uv run --directory pipelines python pois.py ...`).
+try:
+    from pipelines.poi_emit import POI, write_poi_blob
+    from pipelines.walk_graph_reader import WalkGraphReader
+except ImportError:
+    from poi_emit import POI, write_poi_blob
+    from walk_graph_reader import WalkGraphReader
+
+
+# Category code constants. Order = priority (lower wins on multi-tagged POIs).
+CATEGORIES: dict[str, int] = {
+    "food":       1,
+    "transit":    2,
+    "park":       3,
+    "culture":    4,
+    "attraction": 5,
+    "shop":       6,
+    "school":     7,
+    "health":     8,
+    "service":    9,
+    "worship":    10,
+}
+
+SHOP_ALLOWED = {
+    "supermarket", "convenience", "bakery", "books", "clothes",
+    "department_store", "mall", "hardware",
+}
+
+
+def classify_tags(tags: dict[str, str]) -> int | None:
+    """Map an OSM tag bundle to a category code; None if no category applies.
+    Order matches the CATEGORIES dict: food > transit > park > culture >
+    attraction > shop > school > health > service > worship.
+    """
+    amenity = tags.get("amenity")
+    tourism = tags.get("tourism")
+    leisure = tags.get("leisure")
+    railway = tags.get("railway")
+    pt = tags.get("public_transport")
+    aeroway = tags.get("aeroway")
+    shop = tags.get("shop")
+    historic = tags.get("historic")
+
+    if amenity in {"restaurant", "cafe", "bar", "fast_food", "pub",
+                   "food_court", "ice_cream", "biergarten"}:
+        return CATEGORIES["food"]
+    if (railway in {"station", "halt", "tram_stop"}
+            or pt == "station"
+            or amenity == "ferry_terminal"
+            or aeroway == "aerodrome"):
+        return CATEGORIES["transit"]
+    if leisure in {"park", "playground", "garden", "nature_reserve"}:
+        return CATEGORIES["park"]
+    if tourism in {"museum", "gallery"} or amenity in {"theatre", "cinema", "arts_centre", "library"}:
+        return CATEGORIES["culture"]
+    if tourism in {"attraction", "viewpoint", "zoo", "aquarium"} or historic:
+        return CATEGORIES["attraction"]
+    if shop in SHOP_ALLOWED:
+        return CATEGORIES["shop"]
+    if amenity in {"school", "university", "college"}:
+        return CATEGORIES["school"]
+    if amenity in {"hospital", "clinic", "pharmacy", "doctors"}:
+        return CATEGORIES["health"]
+    if amenity in {"post_office", "bank", "fuel", "police", "fire_station"}:
+        return CATEGORIES["service"]
+    if amenity == "place_of_worship":
+        return CATEGORIES["worship"]
+    return None
+
+
+# Categories we KEEP even when name is missing (parks/playgrounds).
+KEEP_UNNAMED_CATS = {CATEGORIES["park"]}
+
+
+@dataclass
+class _RawPOI:
+    lon: float
+    lat: float
+    category: int
+    name: str
+
+
+class _WayCollector(osmium.SimpleHandler):
+    """Pass 0: find category-tagged ways; record node refs + tags."""
+
+    def __init__(self):
+        super().__init__()
+        self.way_tags: dict[int, dict[str, str]] = {}
+        self.way_refs: dict[int, list[int]] = {}
+
+    def way(self, w):
+        tags = dict(w.tags)
+        cat = classify_tags(tags)
+        if cat is None:
+            return
+        name = tags.get("name", "")
+        if not name and cat not in KEEP_UNNAMED_CATS:
+            return
+        self.way_tags[w.id] = tags
+        self.way_refs[w.id] = [n.ref for n in w.nodes]
+
+
+class _NodeCollector(osmium.SimpleHandler):
+    """Pass 1: collect standalone POIs (node-tagged) + coords for way nodes."""
+
+    def __init__(self, way_refs_flat: set[int]):
+        super().__init__()
+        self.way_refs_flat = way_refs_flat
+        self.standalone: list[_RawPOI] = []
+        self.way_node_coords: dict[int, tuple[float, float]] = {}
+
+    def node(self, n):
+        if n.id in self.way_refs_flat:
+            self.way_node_coords[n.id] = (n.location.lon, n.location.lat)
+        tags = dict(n.tags)
+        cat = classify_tags(tags)
+        if cat is None:
+            return
+        name = tags.get("name", "")
+        if not name and cat not in KEEP_UNNAMED_CATS:
+            return
+        self.standalone.append(_RawPOI(n.location.lon, n.location.lat, cat, name))
+
+
+def extract_pois(
+    pbf_path: Path,
+    bbox: tuple[float, float, float, float],
+    walk_graph: WalkGraphReader | None = None,
+):
+    """Yield `POI` records from an OSM PBF. If `walk_graph` is None,
+    walk_node = 0 (used by extraction-only tests)."""
+    pbf_path = Path(pbf_path)
+    min_lon, min_lat, max_lon, max_lat = bbox
+
+    # Pass 0: ways.
+    wc = _WayCollector()
+    wc.apply_file(str(pbf_path))
+
+    way_refs_flat: set[int] = set()
+    for refs in wc.way_refs.values():
+        way_refs_flat.update(refs)
+
+    # Pass 1: standalone POIs + way-node coords.
+    nc = _NodeCollector(way_refs_flat)
+    nc.apply_file(str(pbf_path))
+
+    # Compute centroids for ways.
+    way_pois: list[_RawPOI] = []
+    for way_id, tags in wc.way_tags.items():
+        coords = [nc.way_node_coords.get(n) for n in wc.way_refs[way_id]]
+        coords = [c for c in coords if c is not None]
+        if not coords:
+            continue
+        lons, lats = zip(*coords)
+        clon = (min(lons) + max(lons)) / 2
+        clat = (min(lats) + max(lats)) / 2
+        cat = classify_tags(tags)
+        way_pois.append(_RawPOI(clon, clat, cat, tags.get("name", "")))
+
+    all_raw = nc.standalone + way_pois
+
+    for r in all_raw:
+        if not (min_lon <= r.lon <= max_lon and min_lat <= r.lat <= max_lat):
+            continue
+        if walk_graph is None:
+            walk_node = 0
+        else:
+            node = walk_graph.snap(r.lon, r.lat)
+            if node is None or node not in walk_graph.lcc_nodes:
+                node = walk_graph.snap_in_lcc(r.lon, r.lat, max_m=100.0)
+            if node is None:
+                continue
+            walk_node = node
+        yield POI(
+            lon=r.lon, lat=r.lat, walk_node=walk_node,
+            category=r.category, name=r.name,
+        )
+
+
+def main(argv: list[str]) -> int:
+    if len(argv) != 4:
+        print("usage: pois.py <input.osm.pbf> <walk_graph.bin> <output.bin>",
+              file=sys.stderr)
+        return 2
+    pbf, wg, out = map(Path, argv[1:])
+
+    walk = WalkGraphReader(wg.read_bytes())
+    # scripts/bbox.env BASEMAP_BBOX = -74.30,40.49,-71.85,41.20
+    bbox = (-74.30, 40.49, -71.85, 41.20)
+    pois = list(extract_pois(pbf, bbox, walk))
+    pois.sort(key=lambda p: (p.category, p.name))
+    blob = write_poi_blob(pois, walk_graph_version=walk.version)
+    out.write_bytes(blob)
+    print(f"wrote {out} with {len(pois)} POIs ({len(blob)} bytes)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))