path: root/scripts/update-osm-parks.py

#!/usr/bin/env python3
"""小平市の公園ポリゴンをOpenStreetMapから取得して更新する。

使い方:
    py -X utf8 scripts/update-osm-parks.py

出力:
    src/data/kodaira-parks-osm.json
    src/data/kodaira-boundary.json（小平市の境界線）

OpenStreetMapのデータは ODbL 1.0。地図表示時は
「© OpenStreetMap contributors」の帰属表示が必要。
"""
import json
import os
import sys
import time
import unicodedata
import urllib.request

CITY = "小平市"
OURS = "src/data/kodaira-parks.json"
OUT = "src/data/kodaira-parks-osm.json"
BOUNDARY_OUT = "src/data/kodaira-boundary.json"
OVERRIDES = "src/data/park-name-overrides.json"

# OSMの位置補正（メートル）。地図側のLeafletペインに付いていた
# 16pxの余白をCSSで解消したため、現在は0（補正なし）。
# 実位置との差が残る場合だけ正の値を入れる。
NORTH_SHIFT_M = 0
LAT_PER_M = 1 / 111320

ENDPOINTS = [
    "https://overpass-api.de/api/interpreter",
    "https://overpass.kumi.systems/api/interpreter",
    "https://overpass.private.coffee/api/interpreter",
]

QUERY = f"""[out:json][timeout:90];
area["name"="{CITY}"]["boundary"="administrative"]->.a;
(
  way["leisure"="park"](area.a);
  relation["leisure"="park"](area.a);
);
out geom;
"""

BOUNDARY_QUERY = f"""[out:json][timeout:60];
rel["name"="{CITY}"]["boundary"="administrative"];
out geom;
"""

# 名称が一致しないが、住所・座標から同一と確認できた公園。
ALIAS = {
    "小平中央公園": "中央公園",
    "じゃぶじゃぶ公園": "天神じゃぶじゃぶ公園",
    "リサイクルセンター東公園": "リサイクルセンター南公園",
    "小平市立薬用植物園前公園": "薬用植物園前公園",
    "花南ポケットパーク": "花南ポケットパーク公園",
    "花５四季公園": "花五四季公園",
    "天神町のびのび広場": "天神町のびのび広場公園",
}


def norm(s):
    s = unicodedata.normalize("NFKC", str(s)).replace(" ", "").replace("　", "")
    return s.replace("ヶ", "ケ").replace("ツ", "ッ")


def fetch(query):
    last = None
    for ep in ENDPOINTS:
        try:
            req = urllib.request.Request(
                ep,
                data=query.encode("utf-8"),
                headers={"User-Agent": "yasutakeyohei.com park map updater"},
            )
            with urllib.request.urlopen(req, timeout=120) as r:
                data = json.loads(r.read().decode("utf-8"))
            if data.get("elements"):
                return data
            last = "0 elements"
        except Exception as e:  # noqa: BLE001
            last = e
        print(f"  retry: {ep}: {last}", file=sys.stderr)
        time.sleep(2)
    raise SystemExit(f"Overpass API failed: {last}")


def pt(g):
    return (round(g["lon"], 6), round(g["lat"] + NORTH_SHIFT_M * LAT_PER_M, 6))


def ring_from_ways(ways):
    rings = []
    remaining = [list(w) for w in ways if len(w) >= 2]
    while remaining:
        ring = remaining.pop(0)
        while ring[0] != ring[-1] and remaining:
            found = False
            for i, w in enumerate(remaining):
                if w[0] == ring[-1]:
                    ring.extend(w[1:]); remaining.pop(i); found = True; break
                if w[-1] == ring[-1]:
                    ring.extend(list(reversed(w))[1:]); remaining.pop(i); found = True; break
                if w[0] == ring[0]:
                    ring = list(reversed(w))[:-1] + ring; remaining.pop(i); found = True; break
                if w[-1] == ring[0]:
                    ring = w[:-1] + ring; remaining.pop(i); found = True; break
            if not found:
                break
        if ring[0] != ring[-1]:
            ring.append(ring[0])
        if len(ring) >= 4:
            rings.append(ring)
    return rings


def build_boundary():
    """小平市の行政境界をGeoJSONにする。"""
    data = fetch(BOUNDARY_QUERY)
    if not data.get("elements"):
        print("  boundary relation not found", file=sys.stderr)
        return None
    rel = data["elements"][0]
    ways = []
    for m in rel.get("members", []):
        if m.get("role") == "outer" and m.get("geometry"):
            ways.append([pt(p) for p in m["geometry"]])
    rings = ring_from_ways(ways)
    if not rings:
        print("  boundary ring not assembled", file=sys.stderr)
        return None
    if len(rings) == 1:
        geom = {"type": "Polygon", "coordinates": [rings[0]]}
    else:
        geom = {"type": "MultiPolygon", "coordinates": [[r] for r in rings]}
    props = {
        "name": rel.get("tags", {}).get("name", CITY),
        "boundary": "administrative",
        "admin_level": rel.get("tags", {}).get("admin_level", ""),
    }
    fc = {"type": "FeatureCollection", "features": [{"type": "Feature", "properties": props, "geometry": geom}]}
    return fc, len(rings), sum(len(r) for r in rings)


def main():
    ours = json.load(open(OURS, encoding="utf-8"))
    our_names = {norm(r["公園名"]): r["公園名"] for r in ours}
    alias_norm = {norm(k): v for k, v in ALIAS.items()}

    # OSM ID単位の手動上書き（osm_type/osm_id -> 市の公園名）
    overrides = {}
    if os.path.exists(OVERRIDES):
        raw = json.load(open(OVERRIDES, encoding="utf-8"))
        for key, entry in raw.items():
            city = entry.get("city", "") if isinstance(entry, dict) else str(entry)
            city = str(city).strip()
            if not city:
                continue
            canonical = our_names.get(norm(city))
            if canonical:
                overrides[key] = canonical
            else:
                print(f"  warning: override city not found: {key} -> {city}", file=sys.stderr)

    def match(name):
        k = norm(name)
        if k in our_names:
            return our_names[k]
        if k in alias_norm:
            return alias_norm[k]
        # 「〇〇児童公園」→「〇〇公園」のような表記差を吸収
        k2 = k.replace("児童", "")
        if k2 in our_names:
            return our_names[k2]
        return ""

    data = fetch(QUERY)
    features = []
    applied = 0
    for e in data["elements"]:
        tags = e.get("tags", {})
        name = tags.get("name", "")
        key = f"{e['type']}/{e['id']}"
        our_name = overrides.get(key) or match(name)
        if key in overrides:
            applied += 1
        props = {
            "name": name,
            "our_name": our_name,
            "matched": bool(our_name),
            "osm_type": e["type"],
            "osm_id": e["id"],
        }
        for k in ["leisure", "addr:city", "addr:street", "addr:housenumber", "opening_hours"]:
            if k in tags:
                props[k] = tags[k]

        if e["type"] == "way":
            ring = [pt(p) for p in e.get("geometry", [])]
            if len(ring) < 3:
                continue
            if ring[0] != ring[-1]:
                ring.append(ring[0])
            geom = {"type": "Polygon", "coordinates": [ring]}
        else:
            outers, inners = [], []
            for m in e.get("members", []):
                pts = [pt(p) for p in m.get("geometry", [])]
                if len(pts) < 2:
                    continue
                (inners if m.get("role") == "inner" else outers).append(pts)
            outer_rings = ring_from_ways(outers)
            inner_rings = ring_from_ways(inners)
            if not outer_rings:
                continue
            if len(outer_rings) == 1:
                geom = {"type": "Polygon", "coordinates": [outer_rings[0]] + inner_rings}
            else:
                geom = {"type": "MultiPolygon", "coordinates": [[r] + inner_rings for r in outer_rings]}

        features.append({"type": "Feature", "properties": props, "geometry": geom})

    features.sort(key=lambda f: (f["properties"].get("name", ""), f["properties"]["osm_type"], f["properties"]["osm_id"]))
    out = {"type": "FeatureCollection", "features": features}
    json.dump(out, open(OUT, "w", encoding="utf-8"), ensure_ascii=False, separators=(",", ":"))

    matched = sum(1 for f in features if f["properties"]["matched"])
    named = sum(1 for f in features if f["properties"].get("name"))
    print(f"features {len(features)} (named {named}, matched {matched}) -> {OUT}")
    if overrides:
        print(f"  overrides applied {applied} / {len(overrides)}")

    boundary = build_boundary()
    if boundary:
        boundary_data, ring_count, point_count = boundary
        json.dump(boundary_data, open(BOUNDARY_OUT, "w", encoding="utf-8"), ensure_ascii=False, separators=(",", ":"))
        print(f"boundary rings {ring_count}, points {point_count} -> {BOUNDARY_OUT}")


if __name__ == "__main__":
    main()