#!/usr/bin/env python3 """ Extract Jordan OSM Road Network into SQLite Database for 100% On-Device Offline Routing. Creates 'jordan_roads.db' containing all nodes, edges, geometries, street names, and spatial indexing. """ import sys import os import sqlite3 import json import math HIGHWAY_SPEEDS = { 'motorway': 110, 'motorway_link': 70, 'trunk': 90, 'trunk_link': 60, 'primary': 75, 'primary_link': 50, 'secondary': 60, 'secondary_link': 45, 'tertiary': 50, 'tertiary_link': 35, 'unclassified': 40, 'residential': 35, 'living_street': 20, 'track': 25, 'service': 25, } def haversine_dist(lat1, lon1, lat2, lon2): R = 6371000.0 phi1 = math.radians(lat1) phi2 = math.radians(lat2) dphi = math.radians(lat2 - lat1) dlambda = math.radians(lon2 - lon1) a = math.sin(dphi / 2)**2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2)**2 return 2 * R * math.atan2(math.sqrt(a), math.sqrt(1 - a)) def build_roads_db(pbf_path, out_db_path): print(f"🛣️ Extracting roads from {pbf_path} into {out_db_path}...") try: import osmium except ImportError: print("⚠️ pyosmium not available, installing or using alternative...") import subprocess subprocess.check_call([sys.executable, "-m", "pip", "install", "osmium"]) import osmium if os.path.exists(out_db_path): os.remove(out_db_path) conn = sqlite3.connect(out_db_path) cur = conn.cursor() cur.execute("PRAGMA journal_mode = WAL;") cur.execute("PRAGMA synchronous = NORMAL;") cur.execute(""" CREATE TABLE nodes ( id INTEGER PRIMARY KEY, lat REAL NOT NULL, lng REAL NOT NULL ); """) cur.execute(""" CREATE TABLE edges ( id INTEGER PRIMARY KEY AUTOINCREMENT, osm_way_id INTEGER, from_node INTEGER NOT NULL, to_node INTEGER NOT NULL, name TEXT, highway TEXT NOT NULL, speed_kmh INTEGER NOT NULL, oneway INTEGER NOT NULL, length_m REAL NOT NULL, geom_json TEXT NOT NULL, min_lat REAL, min_lng REAL, max_lat REAL, max_lng REAL ); """) cur.execute("CREATE INDEX idx_edges_from ON edges(from_node);") cur.execute("CREATE INDEX idx_edges_to ON edges(to_node);") cur.execute("CREATE INDEX idx_edges_bbox ON edges(min_lat, max_lat, min_lng, max_lng);") # Pass 1: Collect road ways and needed node IDs class RoadWayHandler(osmium.SimpleHandler): def __init__(self): super().__init__() self.needed_nodes = set() self.ways = [] def way(self, w): highway = w.tags.get('highway') if not highway or highway not in HIGHWAY_SPEEDS: return # Skip pedestrian-only if highway in ('footway', 'pedestrian', 'path', 'steps', 'cycleway'): return name = w.tags.get('name:ar') or w.tags.get('name') or '' oneway_tag = w.tags.get('oneway', 'no') oneway = 1 if oneway_tag in ('yes', '1', 'true') else (-1 if oneway_tag == '-1' else 0) node_refs = [n.ref for n in w.nodes] if len(node_refs) < 2: return for ref in node_refs: self.needed_nodes.add(ref) self.ways.append({ 'id': w.id, 'highway': highway, 'name': name, 'oneway': oneway, 'nodes': node_refs }) print("📖 Pass 1: Scanning road ways...") way_handler = RoadWayHandler() way_handler.apply_file(pbf_path) print(f" Found {len(way_handler.ways):,} road ways and {len(way_handler.needed_nodes):,} road nodes.") # Pass 2: Extract node coordinates node_coords = {} class NodeHandler(osmium.SimpleHandler): def __init__(self, needed): super().__init__() self.needed = needed def node(self, n): if n.id in self.needed: node_coords[n.id] = (round(n.location.lat, 6), round(n.location.lon, 6)) print("📖 Pass 2: Resolving node coordinates...") node_handler = NodeHandler(way_handler.needed_nodes) node_handler.apply_file(pbf_path, locations=False) print(f" Cached coordinates for {len(node_coords):,} nodes.") # Insert nodes into DB print("💾 Inserting nodes into SQLite...") cur.executemany("INSERT OR IGNORE INTO nodes (id, lat, lng) VALUES (?, ?, ?);", [(nid, lat, lng) for nid, (lat, lng) in node_coords.items()]) # Split ways into edges between intersections print("✂️ Segmenting ways into routable edges...") # Count node degrees to identify intersection nodes node_degree = {} for w in way_handler.ways: for nid in w['nodes']: node_degree[nid] = node_degree.get(nid, 0) + 1 edges_to_insert = [] for w in way_handler.ways: w_nodes = w['nodes'] highway = w['highway'] speed = HIGHWAY_SPEEDS.get(highway, 40) name = w['name'] oneway = w['oneway'] current_segment = [] for i, nid in enumerate(w_nodes): if nid not in node_coords: continue current_segment.append(nid) # Split at intersections or end of way is_endpoint = (i == 0 or i == len(w_nodes) - 1) is_intersection = node_degree.get(nid, 0) > 1 if len(current_segment) >= 2 and (is_intersection or is_endpoint): u = current_segment[0] v = current_segment[-1] if u != v: # Calculate geometry and length coords = [node_coords[x] for x in current_segment if x in node_coords] if len(coords) >= 2: length_m = 0.0 min_lat = min(c[0] for c in coords) max_lat = max(c[0] for c in coords) min_lng = min(c[1] for c in coords) max_lng = max(c[1] for c in coords) for j in range(len(coords) - 1): length_m += haversine_dist(coords[j][0], coords[j][1], coords[j+1][0], coords[j+1][1]) geom_json = json.dumps([[c[0], c[1]] for c in coords], separators=(',', ':')) # Forward edge if oneway >= 0: edges_to_insert.append((w['id'], u, v, name, highway, speed, oneway, round(length_m, 1), geom_json, min_lat, min_lng, max_lat, max_lng)) # Reverse edge if oneway <= 0: rev_geom = json.dumps([[c[0], c[1]] for c in reversed(coords)], separators=(',', ':')) edges_to_insert.append((w['id'], v, u, name, highway, speed, oneway, round(length_m, 1), rev_geom, min_lat, min_lng, max_lat, max_lng)) current_segment = [nid] print(f"💾 Inserting {len(edges_to_insert):,} routable edges into SQLite...") cur.executemany(""" INSERT INTO edges (osm_way_id, from_node, to_node, name, highway, speed_kmh, oneway, length_m, geom_json, min_lat, min_lng, max_lat, max_lng) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?); """, edges_to_insert) conn.commit() # Indexing and vacuum print("⚡ Optimizing database & building indices...") cur.execute("ANALYZE;") cur.execute("VACUUM;") conn.close() size_mb = os.path.getsize(out_db_path) / (1024 * 1024) print(f"✅ Successfully built {out_db_path} ({size_mb:.1f} MB) with {len(edges_to_insert):,} edges!") if __name__ == "__main__": pbf = sys.argv[1] if len(sys.argv) > 1 else "/data/valhalla-work/jordan_routing.osm.pbf" out = sys.argv[2] if len(sys.argv) > 2 else "/data/routing-packages/jordan_roads.db" build_roads_db(pbf, out)