diff options
Diffstat (limited to 'scripts/make-park-match-review.py')
| -rw-r--r-- | scripts/make-park-match-review.py | 305 |
1 files changed, 305 insertions, 0 deletions
diff --git a/scripts/make-park-match-review.py b/scripts/make-park-match-review.py new file mode 100644 index 0000000..b8a0d4f --- /dev/null +++ b/scripts/make-park-match-review.py @@ -0,0 +1,305 @@ +"""OSMの公園ポリゴンのうち、市の公園一覧と一致しなかったものをレビュー用に出力する。 + +出力: + reference/park-matching-review.csv … Excelで開ける一覧（UTF-8 BOM） + reference/park-matching-review.geojson … 未一致ポリゴン（QGIS用） + reference/kodaira-parks-points.geojson … 市の公園の位置（QGIS用） + src/data/park-name-overrides.json … OSM ID→市の公園名の手動上書き（未作成時のみ） + +CSVやQGISで確認したうえで、JSONの "city" を埋めて `npm run osm:update` を +実行すると照合結果に反映される。市の公園一覧にない施設・広場は +"status": "unlisted" にすると、候補の提示を止めて「一覧外」として記録する。 +""" + +import csv +import difflib +import json +import math +import os +import sys +import time +import unicodedata + +GEOJSON = "src/data/kodaira-parks-osm.json" +OURS = "src/data/kodaira-parks.json" +OUT_CSV = "reference/park-matching-review.csv" +OUT_REVIEW_GEOJSON = "reference/park-matching-review.geojson" +OUT_POINTS_GEOJSON = "reference/kodaira-parks-points.geojson" +OUT_OVERRIDES = "src/data/park-name-overrides.json" + +CSV_FIELDS = [ + "osm_type", + "osm_id", + "OSM名", + "OSM URL", + "緯度", + "経度", + "面積㎡", + "頂点数", + "候補1", + "距離1m", + "候補2", + "距離2m", + "候補3", + "距離3m", + "名前類似候補", + "類似度", + "類似候補距離m", + "推定", + "判定", + "メモ", +] + + +def norm(s): + s = unicodedata.normalize("NFKC", str(s)).replace(" ", "").replace(" ", "") + return s.replace("ヶ", "ケ").replace("ツ", "ッ") + + +def rings(geom): + if geom["type"] == "Polygon": + return [geom["coordinates"][0]] + if geom["type"] == "MultiPolygon": + return [poly[0] for poly in geom["coordinates"]] + return [] + + +def center(geom): + pts = [p for ring in rings(geom) for p in ring] + if not pts: + return None, None + lats = [p[1] for p in pts] + lons = [p[0] for p in pts] + return (min(lats) + max(lats)) / 2, (min(lons) + max(lons)) / 2 + + +def area_m2(geom): + total = 0.0 + for ring in rings(geom): + if len(ring) < 4: + continue + lat0 = sum(p[1] for p in ring) / len(ring) + kx = 111320 * math.cos(math.radians(lat0)) + ky = 111320 + s = 0.0 + for (x1, y1), (x2, y2) in zip(ring, ring[1:] + ring[:1]): + s += (x1 * kx) * (y2 * ky) - (x2 * kx) * (y1 * ky) + total += abs(s) / 2 + return total + + +def dist_m(lat1, lon1, lat2, lon2): + dy = (lat1 - lat2) * 111320 + dx = (lon1 - lon2) * 111320 * math.cos(math.radians((lat1 + lat2) / 2)) + return math.hypot(dx, dy) + + +def main(): + osm = json.load(open(GEOJSON, encoding="utf-8")) + ours = json.load(open(OURS, encoding="utf-8")) + parks = [] + for r in ours: + try: + lat = float(r.get("緯度") or "") + lon = float(r.get("経度") or "") + except ValueError: + continue + parks.append({"name": r["公園名"], "lat": lat, "lon": lon, "address": r.get("住所", "")}) + park_names = {q["name"] for q in parks} + + decisions = {} + if os.path.exists(OUT_OVERRIDES): + raw = json.load(open(OUT_OVERRIDES, encoding="utf-8")) + decisions = {k: v for k, v in raw.items() if isinstance(v, dict)} + + unmatched = [f for f in osm["features"] if not f["properties"].get("matched")] + rows = [] + review_features = [] + for f in unmatched: + p = f["properties"] + lat, lon = center(f["geometry"]) + if lat is None: + continue + area = area_m2(f["geometry"]) + pts = sum(len(r) for r in rings(f["geometry"])) + near = sorted( + ( + { + "name": q["name"], + "dist": dist_m(lat, lon, q["lat"], q["lon"]), + "similar": difflib.SequenceMatcher(None, norm(p.get("name", "")), norm(q["name"])).ratio(), + } + for q in parks + ), + key=lambda x: x["dist"], + )[:3] + same = sorted( + ( + { + "name": q["name"], + "dist": dist_m(lat, lon, q["lat"], q["lon"]), + "similar": difflib.SequenceMatcher(None, norm(p.get("name", "")), norm(q["name"])).ratio(), + } + for q in parks + if p.get("name") + ), + key=lambda x: -x["similar"], + ) + best_similar = same[0] if same and same[0]["similar"] >= 0.6 else None + + suggestion = "" + if near[0]["dist"] <= 30: + suggestion = near[0]["name"] + elif best_similar and best_similar["similar"] >= 0.75 and best_similar["dist"] <= 300: + suggestion = best_similar["name"] + elif best_similar and best_similar["dist"] <= 100: + suggestion = best_similar["name"] + + memo = [] + if area < 5: + memo.append("面積が極小（形状エラーの可能性）") + if near[0]["dist"] <= 30: + memo.append("最寄り公園の30m以内（同一の可能性）") + if p.get("name") and best_similar: + memo.append(f"名前が類似: {best_similar['name']} ({best_similar['similar']:.2f})") + + key = f"{p.get('osm_type', '')}/{p.get('osm_id', '')}" + decision = decisions.get(key, {}) + decided_city = str(decision.get("city", "") or "").strip() + status = str(decision.get("status", "") or "").strip() + invalid_city = bool(decided_city) and decided_city not in park_names + + if invalid_city: + memo.append(f"上書き先が一覧にありません: {decided_city}") + decided_city = "" + if decided_city: + suggestion = decided_city + memo.append("手動で対応済み") + elif status == "unlisted": + suggestion = "" + memo = [m for m in memo if not m.startswith("名前が類似")] + memo.append("一覧外（手動確認済み）") + + if decided_city: + verdict = f"対応済み: {decided_city}" + elif status == "unlisted": + verdict = "一覧外" + elif invalid_city: + verdict = "要確認: 一覧にない名前" + else: + verdict = "" + + row = { + "osm_type": p.get("osm_type", ""), + "osm_id": p.get("osm_id", ""), + "OSM名": p.get("name", ""), + "OSM URL": f"https://www.openstreetmap.org/{p.get('osm_type', '')}/{p.get('osm_id', '')}", + "緯度": f"{lat:.6f}", + "経度": f"{lon:.6f}", + "面積㎡": f"{area:.1f}", + "頂点数": pts, + "候補1": near[0]["name"] if len(near) > 0 else "", + "距離1m": f"{near[0]['dist']:.0f}" if len(near) > 0 else "", + "候補2": near[1]["name"] if len(near) > 1 else "", + "距離2m": f"{near[1]['dist']:.0f}" if len(near) > 1 else "", + "候補3": near[2]["name"] if len(near) > 2 else "", + "距離3m": f"{near[2]['dist']:.0f}" if len(near) > 2 else "", + "名前類似候補": best_similar["name"] if best_similar else "", + "類似度": f"{best_similar['similar']:.2f}" if best_similar else "", + "類似候補距離m": f"{best_similar['dist']:.0f}" if best_similar else "", + "推定": suggestion, + "判定": verdict, + "メモ": " / ".join(memo), + } + rows.append(row) + review_features.append( + { + "type": "Feature", + "properties": { + "osm": f"{row['osm_type']}/{row['osm_id']}", + "osm_name": row["OSM名"], + "area_m2": round(area, 1), + "nearest": row["候補1"], + "dist_m": row["距離1m"], + "similar": row["名前類似候補"], + "similar_ratio": row["類似度"], + "similar_dist_m": row["類似候補距離m"], + "status": status or ("mapped" if decided_city else ""), + "memo": row["メモ"], + }, + "geometry": f["geometry"], + } + ) + + rows.sort(key=lambda r: (not r["OSM名"], float(r["距離1m"] or 1e9), r["OSM名"])) + os.makedirs(os.path.dirname(OUT_CSV), exist_ok=True) + out_csv = OUT_CSV + try: + fh = open(out_csv, "w", encoding="utf-8-sig", newline="") + except PermissionError: + out_csv = OUT_CSV.replace(".csv", f"-{time.strftime('%H%M%S')}.csv") + print(f"warning: {OUT_CSV} is locked; wrote {out_csv}", file=sys.stderr) + fh = open(out_csv, "w", encoding="utf-8-sig", newline="") + with fh: + w = csv.DictWriter(fh, fieldnames=CSV_FIELDS) + w.writeheader() + w.writerows(rows) + + json.dump( + {"type": "FeatureCollection", "features": review_features}, + open(OUT_REVIEW_GEOJSON, "w", encoding="utf-8"), + ensure_ascii=False, + separators=(",", ":"), + ) + points = [] + for q in parks: + points.append( + { + "type": "Feature", + "properties": {"name": q["name"], "address": q["address"]}, + "geometry": {"type": "Point", "coordinates": [q["lon"], q["lat"]]}, + } + ) + json.dump( + {"type": "FeatureCollection", "features": points}, + open(OUT_POINTS_GEOJSON, "w", encoding="utf-8"), + ensure_ascii=False, + separators=(",", ":"), + ) + + if not os.path.exists(OUT_OVERRIDES): + data = { + f"{r['osm_type']}/{r['osm_id']}": { + "osm_name": r["OSM名"], + "suggested": r.get("推定", ""), + "status": "", + "city": "", + "candidates": [c for c in (r["候補1"], r["候補2"], r["候補3"]) if c], + } + for r in rows + } + with open(OUT_OVERRIDES, "w", encoding="utf-8") as fh: + json.dump(data, fh, ensure_ascii=False, indent=2) + fh.write("\n") + print(f"overrides template -> {OUT_OVERRIDES}") + else: + print(f"overrides exists (not overwritten) -> {OUT_OVERRIDES}") + + named = [r for r in rows if r["OSM名"]] + print(f"unmatched {len(rows)} (named {len(named)}, unnamed {len(rows) - len(named)})") + print(f" csv: {out_csv}") + print(f" geojson: {OUT_REVIEW_GEOJSON} / {OUT_POINTS_GEOJSON}") + print(f" near<=30m: {sum(1 for r in rows if r['距離1m'] and float(r['距離1m']) <= 30)}") + print(f" suggested: {sum(1 for r in rows if r.get('推定'))}") + print(f" area<5m2: {sum(1 for r in rows if float(r['面積㎡']) < 5)}") + print("named:") + for r in named: + print( + f" {r['osm_type']}/{r['osm_id']} {r['OSM名']} -> " + f"類似 {r['名前類似候補'] or '-'} ({r['類似度'] or '-'}, {r['類似候補距離m'] or '-'}m) / " + f"最寄り {r['候補1']} {r['距離1m']}m" + ) + + +if __name__ == "__main__": + main() |