Osmium ist ein CLI-Tool für die Arbeit mit [[OpenStreetMap|OSM]]-Daten und basiert auf der C-Bibliothek libosmium. Es kann mit mehreren OSM-Datenformaten arbeiten, vor allem: * .osm.pbf → binäres [[PBF]]-Format * .osm → [[XML]] * .osc / .osc.gz → [[osm_change_datei|OSM-Change-Dateien]] (siehe [[OpenStreetMap]]) * oft auch komprimierte Varianten wie .gz oder .bz2, je nach Tool/Funktion Osmium ist eher zum Anzeigen und Exportieren gedacht als zum Bearbeiten Siehe auch osmosis, osmconvert =====Buslinien extrahieren===== Buslinien aus [[PBF]] extrahieren. ====1==== # relationen in eine neue pbf schreiben osmium tags-filter wien.osm.pbf \ n/highway=bus_stop \ n/public_transport=platform \ r/route=bus \ -R \ -o bus-relevant.osm.pbf -O Ein Python script #!/usr/bin/env python3 import json import subprocess import sys import xml.etree.ElementTree as ET from collections import defaultdict def fail(msg, code=1): print(msg, file=sys.stderr) sys.exit(code) if len(sys.argv) != 3: fail("Aufruf: python3 osm_bus_stops_to_geojson.py input.osm.pbf output.geojson") input_pbf = sys.argv[1] output_geojson = sys.argv[2] try: proc = subprocess.run( ["osmium", "cat", input_pbf, "-f", "osm"], capture_output=True, text=True, check=False, ) except FileNotFoundError: fail("Fehler: 'osmium' wurde nicht gefunden.") if proc.returncode != 0: fail(f"Fehler bei 'osmium cat':\n{proc.stderr}") if not proc.stdout.strip(): fail("Fehler: 'osmium cat' hat keine Ausgabe geliefert.") try: root = ET.fromstring(proc.stdout) except ET.ParseError as e: fail(f"XML Parse Error: {e}") # Haltestellen-Nodes sammeln stops = {} for node in root.findall("node"): node_id = node.get("id") lat = node.get("lat") lon = node.get("lon") if node_id is None or lat is None or lon is None: continue tags = {tag.get("k"): tag.get("v") for tag in node.findall("tag")} is_bus_stop = tags.get("highway") == "bus_stop" is_platform = tags.get("public_transport") == "platform" if is_bus_stop or is_platform: stops[node_id] = { "id": node_id, "lat": float(lat), "lon": float(lon), "tags": tags, } # Routenrelationen -> Haltestellen mappen routes_by_stop = defaultdict(list) for rel in root.findall("relation"): rel_tags = {tag.get("k"): tag.get("v") for tag in rel.findall("tag")} if rel_tags.get("route") != "bus": continue route_info = { "relation_id": rel.get("id"), "ref": rel_tags.get("ref"), "name": rel_tags.get("name"), "from": rel_tags.get("from"), "to": rel_tags.get("to"), "operator": rel_tags.get("operator"), "network": rel_tags.get("network"), } for member in rel.findall("member"): if member.get("type") != "node": continue ref = member.get("ref") role = member.get("role", "") if ref in stops: entry = dict(route_info) entry["member_role"] = role routes_by_stop[ref].append(entry) features = [] for stop_id, stop in stops.items(): routes = routes_by_stop.get(stop_id, []) # doppelte Routen vermeiden seen = set() unique_routes = [] for r in routes: key = ( r.get("relation_id"), r.get("ref"), r.get("name"), r.get("member_role"), ) if key not in seen: seen.add(key) unique_routes.append(r) route_refs = sorted({r["ref"] for r in unique_routes if r.get("ref")}) route_names = sorted({r["name"] for r in unique_routes if r.get("name")}) properties = dict(stop["tags"]) properties["osm_id"] = stop["id"] properties["osm_type"] = "node" properties["route_refs"] = route_refs properties["route_names"] = route_names properties["routes"] = unique_routes features.append({ "type": "Feature", "geometry": { "type": "Point", "coordinates": [stop["lon"], stop["lat"]], }, "properties": properties, }) geojson = { "type": "FeatureCollection", "features": features, } try: with open(output_geojson, "w", encoding="utf-8") as f: json.dump(geojson, f, ensure_ascii=False, indent=2) except OSError as e: fail(f"Fehler beim Schreiben von '{output_geojson}': {e}") print(f"GeoJSON geschrieben: {output_geojson}") print(f"Haltestellen: {len(features)}") print(f"Haltestellen mit Routeninfo: {sum(1 for f in features if f['properties']['routes'])}") Aufrufen python3 osm_bus_stops_to_geojson.py bus-relevant.osm.pbf bus-stops.geojson Die geojson hat dann so eine struktur { "type": "Feature", "geometry": { "type": "Point", "coordinates": [16.3738, 48.2082] }, "properties": { "name": "Karlsplatz", "highway": "bus_stop", "osm_id": "123456789", "route_refs": ["13A", "59A"], "route_names": [ "13A: Alser Straße - Hauptbahnhof", "59A: Bhf. Meidling - Oper" ], "routes": [ { "relation_id": "111", "ref": "13A", "name": "13A: Alser Straße - Hauptbahnhof", "from": "Alser Straße", "to": "Hauptbahnhof", "operator": "Wiener Linien", "network": "Wiener Linien", "member_role": "platform" } ] } } ====2==== osmium tags-filter wien.osm.pbf \ n/highway=bus_stop \ r/route=bus \ -o step1.osm.pbf -O osmium tags-filter -i step1.osm.pbf \ n/public_transport=platform \ -o bus-relevant.osm.pbf -O #!/usr/bin/env python3 import json import math import subprocess import sys import xml.etree.ElementTree as ET from collections import defaultdict if len(sys.argv) != 3: print("Aufruf: python3 osm_bus_geojson.py input.osm.pbf output.geojson") sys.exit(1) input_pbf = sys.argv[1] output_geojson = sys.argv[2] proc = subprocess.Popen( ["osmium", "cat", input_pbf, "-f", "osm"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) nodes = {} # node_id -> (lon, lat) stop_nodes = {} # node_id -> stop feature candidate route_ways = {} # way_id -> dict(tags, node_refs) route_memberships_by_stop = defaultdict(list) route_relations = [] def tags_from_elem(elem): return {t.attrib["k"]: t.attrib["v"] for t in elem.findall("tag")} def is_stop_node(tags): return ( tags.get("highway") == "bus_stop" or tags.get("public_transport") in {"platform", "stop_position"} ) def normalize_name(value): if not value: return "" return " ".join(value.strip().lower().split()) def rounded_coord(value, digits=5): return round(float(value), digits) def make_stop_candidate(osm_id, lon, lat, tags): return { "osm_id": int(osm_id), "lon": float(lon), "lat": float(lat), "name": tags.get("name"), "highway": tags.get("highway"), "public_transport": tags.get("public_transport"), "ref": tags.get("ref"), "local_ref": tags.get("local_ref"), "network": tags.get("network"), "operator": tags.get("operator"), "shelter": tags.get("shelter"), "bench": tags.get("bench"), "bin": tags.get("bin"), "tactile_paving": tags.get("tactile_paving"), "wheelchair": tags.get("wheelchair"), "raw_tags": tags, } def stop_priority(stop): """ Bevorzugung für den repräsentativen Punkt je Haltestelle. Niedriger = besser. """ pt = stop.get("public_transport") hw = stop.get("highway") if hw == "bus_stop": return 0 if pt == "platform": return 1 if pt == "stop_position": return 2 return 9 context = ET.iterparse(proc.stdout, events=("end",)) for event, elem in context: if elem.tag == "node": node_id = elem.attrib["id"] lon = float(elem.attrib["lon"]) lat = float(elem.attrib["lat"]) nodes[node_id] = (lon, lat) tags = tags_from_elem(elem) if is_stop_node(tags): stop_nodes[node_id] = make_stop_candidate(node_id, lon, lat, tags) elem.clear() elif elem.tag == "way": way_id = elem.attrib["id"] tags = tags_from_elem(elem) node_refs = [nd.attrib["ref"] for nd in elem.findall("nd")] route_ways[way_id] = {"tags": tags, "node_refs": node_refs} elem.clear() elif elem.tag == "relation": tags = tags_from_elem(elem) if tags.get("route") == "bus": rel_id = int(elem.attrib["id"]) members = elem.findall("member") route_way_ids = [] for member in members: mtype = member.attrib.get("type") mref = member.attrib.get("ref") role = member.attrib.get("role", "") if mtype == "node" and role in {"stop", "platform", "stop_entry_only", "stop_exit_only", ""}: route_memberships_by_stop[mref].append({ "route_id": rel_id, "line_ref": tags.get("ref"), "route_name": tags.get("name"), "role": role, "network": tags.get("network"), "operator": tags.get("operator"), "from": tags.get("from"), "to": tags.get("to"), }) elif mtype == "way" and role in {"", "forward", "backward", "route"}: route_way_ids.append(mref) route_relations.append({ "relation_id": rel_id, "tags": tags, "way_ids": route_way_ids, }) elem.clear() stderr = proc.stderr.read() ret = proc.wait() if ret != 0: print(stderr, file=sys.stderr) sys.exit(ret) # ------------------------------------------------- # Stop-Kandidaten mit Routen anreichern # ------------------------------------------------- enriched_stops = [] for node_id, stop in stop_nodes.items(): memberships = route_memberships_by_stop.get(node_id, []) seen = set() bus_routes = [] bus_lines = [] bus_route_names = [] for r in memberships: key = (r["route_id"], r["line_ref"], r["route_name"]) if key in seen: continue seen.add(key) bus_routes.append(r) if r["line_ref"] and r["line_ref"] not in bus_lines: bus_lines.append(r["line_ref"]) if r["route_name"] and r["route_name"] not in bus_route_names: bus_route_names.append(r["route_name"]) stop["bus_routes"] = bus_routes stop["bus_lines"] = bus_lines stop["bus_route_names"] = bus_route_names enriched_stops.append(stop) # ------------------------------------------------- # Haltestellen zusammenfassen # ------------------------------------------------- grouped = defaultdict(list) for stop in enriched_stops: name_key = normalize_name(stop.get("name")) local_ref_key = (stop.get("local_ref") or "").strip().lower() coord_key = (rounded_coord(stop["lon"], 5), rounded_coord(stop["lat"], 5)) # Falls name fehlt, immerhin local_ref + Nähe verwenden # Falls local_ref auch fehlt, name + Nähe group_key = ( name_key, local_ref_key, coord_key, ) grouped[group_key].append(stop) features = [] for group_key, group_stops in grouped.items(): # repräsentativen Punkt wählen group_stops_sorted = sorted( group_stops, key=lambda s: ( stop_priority(s), 0 if s.get("name") else 1, s["osm_id"] ) ) rep = group_stops_sorted[0] # Linien / Routen deduplizieren gruppenweit seen_route_ids = set() merged_routes = [] merged_lines = [] merged_route_names = [] for stop in group_stops: for r in stop.get("bus_routes", []): rid = r.get("route_id") if rid in seen_route_ids: continue seen_route_ids.add(rid) merged_routes.append(r) line_ref = r.get("line_ref") if line_ref and line_ref not in merged_lines: merged_lines.append(line_ref) route_name = r.get("route_name") if route_name and route_name not in merged_route_names: merged_route_names.append(route_name) merged_lines.sort(key=lambda x: str(x)) merged_route_names.sort(key=lambda x: str(x)) feature = { "type": "Feature", "geometry": { "type": "Point", "coordinates": [rep["lon"], rep["lat"]], }, "properties": { "feature_kind": "stop", "osm_type": "node", "osm_id": rep["osm_id"], "name": rep.get("name"), "highway": rep.get("highway"), "public_transport": rep.get("public_transport"), "ref": rep.get("ref"), "local_ref": rep.get("local_ref"), "network": rep.get("network"), "operator": rep.get("operator"), "shelter": rep.get("shelter"), "bench": rep.get("bench"), "bin": rep.get("bin"), "tactile_paving": rep.get("tactile_paving"), "wheelchair": rep.get("wheelchair"), "bus_lines": merged_lines, "bus_route_names": merged_route_names, "bus_routes": merged_routes, "group_size": len(group_stops), "merged_osm_ids": [s["osm_id"] for s in group_stops], "raw_tags": rep.get("raw_tags", {}), }, } features.append(feature) # ------------------------------------------------- # Bus-Routen als GeoJSON drinlassen # ------------------------------------------------- for rel in route_relations: segments = [] for way_id in rel["way_ids"]: way = route_ways.get(way_id) if not way: continue coords = [nodes[nid] for nid in way["node_refs"] if nid in nodes] if len(coords) >= 2: segments.append([list(c) for c in coords]) if not segments: continue if len(segments) == 1: geometry = { "type": "LineString", "coordinates": segments[0], } else: geometry = { "type": "MultiLineString", "coordinates": segments, } tags = rel["tags"] features.append({ "type": "Feature", "geometry": geometry, "properties": { "feature_kind": "route", "osm_type": "relation", "osm_id": rel["relation_id"], "route": tags.get("route"), "ref": tags.get("ref"), "name": tags.get("name"), "network": tags.get("network"), "operator": tags.get("operator"), "from": tags.get("from"), "to": tags.get("to"), "colour": tags.get("colour"), "raw_tags": tags, }, }) geojson = { "type": "FeatureCollection", "features": features, } with open(output_geojson, "w", encoding="utf-8") as f: json.dump(geojson, f, ensure_ascii=False) n_stops = sum(1 for f in features if f["properties"].get("feature_kind") == "stop") n_routes = sum(1 for f in features if f["properties"].get("feature_kind") == "route") print(f"Geschrieben: {output_geojson}") print(f"Stops: {n_stops}") print(f"Routen: {n_routes}") ====HTML==== let busGeojson = await fetch("bus-stops.geojson").then(r => { if (!r.ok) throw new Error(`bus-stops.geojson konnte nicht geladen werden (${r.status})`); return r.json(); }); console.log("busGeojson:", busGeojson); map.addSource("bus", { type: "geojson", data: busGeojson }); // Quelle für die ausgewählte Route map.addSource("selected-bus-route", { type: "geojson", data: { type: "FeatureCollection", features: [] } }); // Nur die ausgewählte Linie zeichnen map.addLayer({ id: "selected-bus-route-outline", type: "line", source: "selected-bus-route", layout: { "line-cap": "round", "line-join": "round" }, paint: { "line-color": "#ffffff", "line-width": 8, "line-opacity": 0.95 } }); map.addLayer({ id: "selected-bus-route-line", type: "line", source: "selected-bus-route", layout: { "line-cap": "round", "line-join": "round" }, paint: { "line-color": "#d08a00", "line-width": 5, "line-opacity": 0.95 } }); // NUR Stops anzeigen map.addLayer({ id: "bus-stops", type: "circle", source: "bus", minzoom: 12, filter: [ "all", ["==", ["geometry-type"], "Point"], ["==", ["get", "feature_kind"], "stop"], ["==", ["get", "public_transport"], "stop_position"] ], paint: { "circle-radius": 4, "circle-color": "#15803d", "circle-stroke-color": "#ffffff", "circle-stroke-width": 1 } }); function parseJsonArray(value) { if (Array.isArray(value)) return value; if (typeof value !== "string" || !value.trim()) return []; try { const parsed = JSON.parse(value); return Array.isArray(parsed) ? parsed : []; } catch (err) { console.warn("parseJsonArray failed:", value, err); return []; } } function parseJsonObjectArray(value) { if (Array.isArray(value)) return value; if (typeof value !== "string" || !value.trim()) return []; try { const parsed = JSON.parse(value); return Array.isArray(parsed) ? parsed : []; } catch (err) { console.warn("parseJsonObjectArray failed:", value, err); return []; } } function getBusRouteFeatureByRelationId(routeId) { if (!busGeojson?.features) return null; return busGeojson.features.find((f) => { const p = f.properties || {}; return p.feature_kind === "route" && String(p.osm_id) === String(routeId); }) || null; } function clearSelectedBusRoute() { const src = map.getSource("selected-bus-route"); if (!src) return; src.setData({ type: "FeatureCollection", features: [] }); } function showSelectedBusRoute(routeId) { const feature = getBusRouteFeatureByRelationId(routeId); if (!feature) { console.warn("Keine Route gefunden für relation id:", routeId); return; } const src = map.getSource("selected-bus-route"); if (!src) return; src.setData({ type: "FeatureCollection", features: [feature] }); const bounds = new maplibregl.LngLatBounds(); if (feature.geometry?.type === "LineString") { feature.geometry.coordinates.forEach((coord) => bounds.extend(coord)); } else if (feature.geometry?.type === "MultiLineString") { feature.geometry.coordinates.forEach((line) => { line.forEach((coord) => bounds.extend(coord)); }); } if (!bounds.isEmpty()) { map.fitBounds(bounds, { padding: 50, maxZoom: 16 }); } } function buildBusStopPopupHtml(props, lng, lat) { const name = props.name || "Haltestelle"; const network = props.network || "-"; const osmId = props.osm_id || "-"; const busRoutes = parseJsonObjectArray(props.bus_routes); const busLines = parseJsonArray(props.bus_lines); let connectionsHtml = "
Keine Linien gefunden
"; if (busRoutes.length) { connectionsHtml = `
${busRoutes.map((route) => ` `).join("")}
`; } else if (busLines.length) { connectionsHtml = `
${busLines.map(line => `
${escapeHtml(String(line))}
`).join("")}
`; } return `
${escapeHtml(name)}
Netzwerk: ${escapeHtml(network)}
OSMID: ${escapeHtml(String(osmId))}
Koordinaten: ${lat}, ${lng}
Linien
${connectionsHtml}
`; } map.on("click", "bus-stops", (e) => { const f = e.features?.[0]; if (!f) return; const props = f.properties || {}; const lng = e.lngLat.lng.toFixed(6); const lat = e.lngLat.lat.toFixed(6); const popup = new maplibregl.Popup() .setLngLat(e.lngLat) .setHTML(buildBusStopPopupHtml(props, lng, lat)) .addTo(map); const popupEl = popup.getElement(); if (!popupEl) return; popupEl.querySelectorAll(".bus-route-btn").forEach((btn) => { btn.addEventListener("click", () => { const routeId = btn.dataset.routeId; if (!routeId) return; showSelectedBusRoute(routeId); }); }); popupEl.querySelector("#clear-bus-route-btn")?.addEventListener("click", () => { clearSelectedBusRoute(); }); }); map.on("mouseenter", "bus-stops", () => { map.getCanvas().style.cursor = "pointer"; }); map.on("mouseleave", "bus-stops", () => { map.getCanvas().style.cursor = ""; });
=====Linux===== sudo apt update sudo apt install osmium-tool osmium extract --bbox 16.20,48.10,16.50,48.35 --strategy smart -o wien.osm.pbf europe-latest.osm.pbf osmium tags-filter wien.osm.pbf n/highway=bus_stop -o bus_stops.osm.pbf osmium sort edited.osm.pbf -o edited-sorted.osm.pbf osmium renumber edited-sorted.osm.pbf -o edited-renumbered.osm.pbf osmium fileinfo -e edited-renumbered.osm.pbf osmium extract -b 16.36,48.20,16.38,48.21 input.osm.pbf -o output.osm.pbf --output-header=generator="MeinTool/1.0" Siehe [[https://docs.osmcode.org/osmium/latest/osmium-extract.html|osmium-extract]] # bbox at - 9.54063,46.37853,17.17397,49.01266 vienna - 16.20,48.10,16.50,48.35 =====Python===== import osmium import sys MIN_LON = 16.20 MIN_LAT = 48.10 MAX_LON = 16.50 MAX_LAT = 48.35 INPUT_FILE = "europe-latest.osm.pbf" OUTPUT_FILE = "wien.osm.pbf" def in_bbox(lon: float, lat: float) -> bool: return MIN_LON <= lon <= MAX_LON and MIN_LAT <= lat <= MAX_LAT class BBoxWriter(osmium.SimpleHandler): def __init__(self, writer): super().__init__() self.writer = writer def node(self, n): if n.location.valid() and in_bbox(n.location.lon, n.location.lat): self.writer.add_node(n) def way(self, w): for nr in w.nodes: if nr.location.valid() and in_bbox(nr.location.lon, nr.location.lat): self.writer.add_way(w) return def relation(self, r): # Sehr einfache Variante: # Relations werden hier nicht sauber räumlich geprüft. # Für vollständige OSM-Extrakte ist osmium-tool besser. pass def main(): writer = osmium.SimpleWriter(OUTPUT_FILE) handler = BBoxWriter(writer) try: # mit_locations() ist nötig, damit Ways Node-Koordinaten bekommen handler.apply_file(INPUT_FILE, locations=True) finally: writer.close() print(f"Fertig: {OUTPUT_FILE}") if __name__ == "__main__": main() python.exe -m venv .venv .venv\Scripts\activate python -m pip install --upgrade pip pip install osmium python script.py =====Search===== Search-Source extrahieren wget https://download.geofabrik.de/europe/austria-latest.osm.pbf osmium tags-filter austria-latest.osm.pbf \ nwr/place \ nwr/amenity \ nwr/shop \ nwr/tourism \ nwr/leisure \ nwr/railway \ nwr/public_transport \ nwr/highway \ nwr/addr:* \ -o search-source.osm.pbf osmium export search-source.osm.pbf -o search-source.geojson search.json bauen import json INPUT = "search-source.geojson" OUTPUT = "search.json" def center_of_geometry(geom): t = geom.get("type") coords = geom.get("coordinates") if t == "Point": return coords[0], coords[1] if t == "LineString" and coords: mid = coords[len(coords) // 2] return mid[0], mid[1] if t == "Polygon" and coords and coords[0]: ring = coords[0] mid = ring[len(ring) // 2] return mid[0], mid[1] if t == "MultiPolygon" and coords and coords[0] and coords[0][0]: ring = coords[0][0] mid = ring[len(ring) // 2] return mid[0], mid[1] return None, None def norm(props, *keys): for k in keys: v = props.get(k) if v: return str(v).strip() return "" with open(INPUT, "r", encoding="utf-8") as f: geo = json.load(f) out = [] seen = set() for feat in geo.get("features", []): props = feat.get("properties", {}) geom = feat.get("geometry", {}) lng, lat = center_of_geometry(geom) if lng is None or lat is None: continue name = norm(props, "name:de", "name_de", "name") place = norm(props, "place") housenumber = norm(props, "addr:housenumber") street = norm(props, "addr:street") city = norm(props, "addr:city", "addr:place") postcode = norm(props, "addr:postcode") amenity = norm(props, "amenity") shop = norm(props, "shop") tourism = norm(props, "tourism") leisure = norm(props, "leisure") railway = norm(props, "railway") public_transport = norm(props, "public_transport") highway = norm(props, "highway") item = None if housenumber and street: item = { "t": "house", "name": f"{street} {housenumber}", "street": street, "housenumber": housenumber, "city": city, "postcode": postcode, "lat": round(lat, 6), "lng": round(lng, 6), } elif highway and name: item = { "t": "street", "name": name, "city": city, "postcode": postcode, "lat": round(lat, 6), "lng": round(lng, 6), } elif place and name: item = { "t": "place", "name": name, "place": place, "postcode": postcode, "lat": round(lat, 6), "lng": round(lng, 6), } elif name and (amenity or shop or tourism or leisure or railway or public_transport): item = { "t": "poi", "name": name, "class": amenity or shop or tourism or leisure or railway or public_transport, "lat": round(lat, 6), "lng": round(lng, 6), } if not item: continue key = (item["t"], item["name"], item["lat"], item["lng"]) if key in seen: continue seen.add(key) out.append(item) with open(OUTPUT, "w", encoding="utf-8") as f: json.dump(out, f, ensure_ascii=False, separators=(",", ":")) print(f"{len(out)} Einträge nach {OUTPUT} geschrieben") Im Browser suchen let SEARCH_INDEX = []; async function loadSearchIndex() { const res = await fetch("search.json"); SEARCH_INDEX = await res.json(); } function searchLocal(query) { const q = query.trim().toLowerCase(); if (!q) return []; return SEARCH_INDEX .filter(item => { const hay = [ item.name || "", item.street || "", item.city || "", item.postcode || "" ].join(" ").toLowerCase(); return hay.includes(q); }) .slice(0, 20); } =====Links===== * [[https://osmcode.org/osmium-tool/|Osmium Tool]] * [[https://github.com/osmcode/osmium-tool|Osmium Tool - Github]] * [[https://docs.osmcode.org/osmium|Docs]]