#! python3 # -*- coding: utf-8 -*- """ 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