DOSSIER/rhino/osm.py

#! python3
# -*- coding: utf-8 -*-
# SPDX-License-Identifier: AGPL-3.0-or-later
# Copyright (C) 2026 Karim Gabriele Varano
"""
OSM-Importer fuer Dossier — holt OpenStreetMap-Daten via Overpass-API als
Polylinien (Strassen, Gebaeudeumrisse, Wasser, Gruenflaechen, Wege).

Pipeline:
  Adresse → bbox (LV95) → bbox (WGS84) → Overpass-Query →
  JSON-Response → OSM-Ways → Polylinien (in Doc-Units) → Rhino-Layer

Koord-Konversion WGS84↔LV95 nutzt swisstopo.wgs84_to_lv95 (LV95 ist die
gemeinsame Basis mit dem swisstopo-Importer).
"""

import os
import json
import urllib.request
import urllib.parse

import Rhino
import Rhino.Geometry as rg

import swisstopo  # fuer wgs84_to_lv95


OVERPASS_URL = "https://overpass-api.de/api/interpreter"


# --- Kategorien ------------------------------------------------------------
# Jede Kategorie liefert (Overpass-Selektor, Layer-Code, Layer-Name, Color).
# Codes 7100-7199 reserviert fuer OSM-Sub-Ebenen unter '70_osm'.
CATEGORIES = {
    "streets": {
        "selector": '[highway~"^(motorway|trunk|primary|secondary|tertiary|residential|unclassified|service|living_street|pedestrian)$"]',
        "code":  "7101", "name": "Strassen",         "color": "#a89070",
    },
    "buildings": {
        "selector": '[building]',
        "code":  "7102", "name": "Gebaeudeumrisse",  "color": "#888888",
        "include_relations": True,
    },
    "water": {
        "selector": '[natural=water]',
        "code":  "7103", "name": "Wasser",           "color": "#4080a0",
        "include_relations": True,
    },
    "waterways": {
        "selector": '[waterway~"^(river|stream|canal)$"]',
        "code":  "7104", "name": "Wasserlaeufe",     "color": "#4080a0",
    },
    "parks": {
        "selector": '[leisure~"^(park|garden)$"]',
        "code":  "7105", "name": "Parks",            "color": "#60a070",
        "include_relations": True,
    },
    "forest": {
        "selector": '[landuse~"^(forest|grass|meadow)$"]',
        "code":  "7106", "name": "Wald_Gruen",       "color": "#406050",
        "include_relations": True,
    },
    "footpaths": {
        "selector": '[highway~"^(footway|path|track|cycleway)$"]',
        "code":  "7107", "name": "Wege",             "color": "#806040",
    },
}


def build_overpass_query(bbox_wgs, categories):
    """Baut die Overpass-QL-Query fuer bbox + ausgewaehlte Kategorien.
    bbox_wgs: (min_lon, min_lat, max_lon, max_lat) — WGS84."""
    south = bbox_wgs[1]; west = bbox_wgs[0]
    north = bbox_wgs[3]; east = bbox_wgs[2]
    bbox_str = "{},{},{},{}".format(south, west, north, east)
    parts = []
    for cat in categories:
        spec = CATEGORIES.get(cat)
        if not spec: continue
        parts.append('way{}({});'.format(spec["selector"], bbox_str))
        if spec.get("include_relations"):
            parts.append('relation{}({});'.format(spec["selector"], bbox_str))
    body = ''.join(parts)
    return '[out:json][timeout:60];({});out body;>;out skel qt;'.format(body)


def fetch_overpass(bbox_wgs, categories, progress=None):
    """Schickt Overpass-Query, liefert JSON-Dict oder None."""
    q = build_overpass_query(bbox_wgs, categories)
    if progress: progress("Overpass-Query ({} Kategorien)...".format(len(categories)))
    data = urllib.parse.urlencode({"data": q}).encode("utf-8")
    req = urllib.request.Request(OVERPASS_URL, data=data, method="POST",
                                  headers={"User-Agent": "Dossier/OSM-Importer"})
    try:
        with urllib.request.urlopen(req, timeout=180) as resp:
            text = resp.read().decode("utf-8", errors="ignore")
        out = json.loads(text)
        if progress: progress("Antwort: {} Elemente".format(len(out.get("elements", []))))
        return out
    except Exception as ex:
        if progress: progress("Overpass fail: {}".format(ex))
        return None


def parse_osm_elements(osm_json):
    """Zerlegt OSM-JSON in {nodes: {id: (lon, lat)}, ways: [{id, nodes, tags}]}."""
    if not osm_json: return None
    nodes = {}
    ways  = []
    for el in osm_json.get("elements", []):
        t = el.get("type")
        if t == "node":
            nodes[el["id"]] = (el["lon"], el["lat"])
        elif t == "way":
            ways.append({
                "id":    el["id"],
                "nodes": el.get("nodes", []),
                "tags":  el.get("tags") or {},
            })
    return {"nodes": nodes, "ways": ways}


def classify_way(tags):
    """Mappt Way-Tags auf eine Kategorie-Key (oder None falls uninteressant)."""
    if not tags: return None
    hw = tags.get("highway")
    if hw in ("motorway","trunk","primary","secondary","tertiary",
              "residential","unclassified","service","living_street","pedestrian"):
        return "streets"
    if hw in ("footway","path","track","cycleway"): return "footpaths"
    if tags.get("building"): return "buildings"
    if tags.get("natural") == "water": return "water"
    ww = tags.get("waterway")
    if ww in ("river","stream","canal"): return "waterways"
    if tags.get("leisure") in ("park","garden"): return "parks"
    if tags.get("landuse") in ("forest","grass","meadow"): return "forest"
    return None


def way_to_polyline(way_node_ids, nodes, shift_lv95, m_to_unit, z=0.0):
    """OSM-Way → Rhino.Polyline in Doc-Units. shift_lv95 = (sx, sy, sz) Origin-
    Shift in LV95-Metern (gleicher Pipeline wie swisstopo)."""
    pts = []
    sx, sy, sz = shift_lv95
    for nid in way_node_ids:
        node = nodes.get(nid)
        if node is None: continue
        lon, lat = node
        e, n = swisstopo.wgs84_to_lv95(lon, lat)
        x = (e - sx) * m_to_unit
        y = (n - sy) * m_to_unit
        pts.append(rg.Point3d(x, y, z))
    if len(pts) < 2: return None
    poly = rg.Polyline(pts)
    return poly


def import_osm_to_doc(doc, bbox_wgs, categories, shift_lv95, m_to_unit,
                        z_doc=0.0, progress=None):
    """End-to-end-Import: Overpass-Query + Polylinien-Erzeugung. Liefert
    Liste von dicts: [{category, obj_id, way_tags}, ...] — Aufrufer macht
    Layer-Move + Tag selbst."""
    osm_json = fetch_overpass(bbox_wgs, categories, progress=progress)
    if osm_json is None: return []
    parsed = parse_osm_elements(osm_json)
    if not parsed: return []
    nodes = parsed["nodes"]
    ways  = parsed["ways"]
    if progress: progress("Parse {} Ways...".format(len(ways)))
    created = []
    for way in ways:
        cat = classify_way(way["tags"])
        if cat is None or cat not in categories: continue
        poly = way_to_polyline(way["nodes"], nodes, shift_lv95,
                                m_to_unit, z=z_doc)
        if poly is None or poly.Count < 2: continue
        # Wenn Polyline geschlossen ist (erster == letzter Punkt) → als Curve
        # mit Schluss-Edge, sonst offene Polyline.
        curve = poly.ToNurbsCurve()
        if curve is None: continue
        gid = doc.Objects.AddCurve(curve)
        if gid is None: continue
        obj = doc.Objects.Find(gid)
        if obj is None: continue
        created.append({
            "category": cat,
            "obj":      obj,
            "tags":     way["tags"],
        })
    if progress: progress("→ {} OSM-Linien erzeugt".format(len(created)))
    return created