osm neatnet

neatnet.py

#!/usr/bin/env python3
import sys
import os
import osmnx as ox
import networkx as nx
import neatnet
import geopandas as gpd
from shapely.geometry import Point

# Configure OSMnx settings
ox.settings.all_oneway = True

def main():
    if len(sys.argv) != 3:
        print(f"Usage: {sys.argv[0]} input.osm.pbf output.gpkg")
        sys.exit(1)

    # place = "Monaco"
    local_crs = 2154

    osm = sys.argv[1]
    osm_drive = osm.replace('latest', 'latest-drive').replace('.osm.pbf', '.osm')
    osm_building = osm.replace('latest', 'latest-building').replace('.osm.pbf', '.osm')

    print(f"Prepare OSM data {osm}...")
    os.system(f"osmium tags-filter --overwrite -o 0.{osm_drive} {osm} 'w/highway!=abandoned,bridleway,bus_guideway,construction,corridor,cycleway,elevator,escalator,footway,no,path,pedestrian,planned,platform,proposed,raceway,razed,service,steps,track'")
    os.system(f"osmium tags-filter --overwrite -o {osm_drive} 0.{osm_drive} --invert-match 'w/area=yes' 'w/motor_vehicle=no' 'w/motorcar=no' 'w/service=alley,driveway,emergency_access,parking,parking_aisle,private'")
    os.system(f"osmium tags-filter --overwrite -o {osm_building} {osm} 'a/building!=roog,no'")

    # Read drive graph from XML file
    print(f"Loading drive graph from {osm_drive}...")
    osm_graph = ox.graph_from_xml(osm_drive, bidirectional=True, simplify=False, retain_all=False)
    # osm_graph = ox.graph_from_place(place, network_type='drive', simplify=False)

    osm_graph = ox.projection.project_graph(osm_graph, to_crs=local_crs)
    streets = ox.graph_to_gdfs(
        ox.convert.to_undirected(osm_graph),
        nodes=False,
        edges=True,
        node_geometry=False,
        fill_edge_geometry=True,
    ).reset_index(drop=True)

    # Read building graph from XML file
    print(f"Loading building graph from {osm_building}...")
    buildings_graph = ox.graph_from_xml(osm_building, bidirectional=False, simplify=True, retain_all=True)
    buildings_graph = ox.projection.project_graph(buildings_graph, to_crs=local_crs)
    buildings = ox.graph_to_gdfs(
        ox.convert.to_undirected(buildings_graph),
        nodes=False,
        edges=True,
        node_geometry=False,
        fill_edge_geometry=True,
    ).reset_index(drop=True)

    # Apply neatnet processing
    print("Neatify the graph...")
    # streets = neatnet.neatify(streets)
    streets = neatnet.neatify(streets, exclusion_mask=buildings.geometry)

    # Make it compatible with graph_from_gdfs
    streets.rename(columns={'from': 'u', 'to': 'v'}, inplace=True)

    # Extract unique nodes from the processed streets' geometries
    unique_nodes = set()
    node_geoms = {}

    streets.u[streets['u'].isna()] = -streets.index
    n = streets.size
    streets.v[streets['v'].isna()] = -streets.index - n

    # Get start and end nodes from each street segment
    for idx, row in streets.iterrows():
        u, v = row['u'], row['v']
        unique_nodes.add(u)
        unique_nodes.add(v)

        # Store node geometries from start/end points of LineStrings
        if u not in node_geoms and row.geometry is not None:
            node_geoms[u] = row.geometry.coords[0]
        if v not in node_geoms and row.geometry is not None:
            node_geoms[v] = row.geometry.coords[-1]

    node_data = []
    for node_id in unique_nodes:
        if node_id in node_geoms:
            x, y = node_geoms[node_id]
            node_data.append({'osmid': node_id, 'geometry': Point(x, y)})

    gdf_nodes = gpd.GeoDataFrame(node_data, crs=streets.crs)
    gdf_nodes.set_index('osmid', inplace=True)
    gdf_nodes.to_crs(4326, inplace=True)
    gdf_nodes['x'] = gdf_nodes.geometry.x
    gdf_nodes['y'] = gdf_nodes.geometry.y

    streets["id"] = streets.index + 1
    streets.set_index(['u', 'v', 'id'], inplace=True)
    streets.to_crs(4326, inplace=True)

    osm_graph = ox.graph_from_gdfs(gdf_nodes, streets)

    # Save the processed graph
    print(f"Saving processed graph to {sys.argv[2]}...")
    ox.save_graph_geopackage(osm_graph, sys.argv[2])
    # ox.save_graph_xml(osm_graph, sys.argv[2])
    print("Done!")

if __name__ == "__main__":
    main()