Spaces:
Sleeping
Sleeping
File size: 2,536 Bytes
4e22ad8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | """All-pairs travel-time + distance matrix between snapped OSM road nodes.
Snaps each (lat, lng) to its nearest road node, then runs one Dijkstra per
unique source on the *directed* graph (one-ways respected). The matrix is
asymmetric and dense — every fitness evaluation downstream is now an O(1)
numpy lookup.
"""
from __future__ import annotations
import csv
import json
import time
from dataclasses import dataclass
from pathlib import Path
from dataclasses import dataclass
import networkx as nx
import numpy as np
import osmnx as ox
import pyproj
from graph_manager import DEFAULT_COORDS_PATH, get_or_build_graph
DEFAULT_MATRIX_PATH = Path(__file__).with_name("travel_time.npz")
@dataclass
class TravelMatrix:
point_ids: list[str]
node_ids: list[int]
time_s: np.ndarray
dist_m: np.ndarray
def index_of(self, point_id: str) -> int:
return self.point_ids.index(point_id)
def _snap(graph: nx.MultiDiGraph, points: list[tuple[float, float]]) -> list[int]:
proj = ox.projection.project_graph(graph)
transformer = pyproj.Transformer.from_crs(
graph.graph["crs"], proj.graph["crs"], always_xy=True
)
lats = [p[0] for p in points]
lngs = [p[1] for p in points]
proj_x, proj_y = transformer.transform(lngs, lats)
return list(ox.distance.nearest_nodes(proj, X=list(proj_x), Y=list(proj_y)))
def build_matrix(
graph: nx.MultiDiGraph,
points: list[tuple[str, float, float]],
) -> TravelMatrix:
point_ids = [p[0] for p in points]
node_ids = _snap(graph, [(p[1], p[2]) for p in points])
n = len(points)
time_s = np.full((n, n), np.inf, dtype=np.float32)
dist_m = np.full((n, n), np.inf, dtype=np.float32)
by_node: dict[int, list[int]] = {}
for i, nd in enumerate(node_ids):
by_node.setdefault(nd, []).append(i)
targets = set(node_ids)
for src_node, src_rows in by_node.items():
t_to = nx.single_source_dijkstra_path_length(graph, src_node, weight="travel_time")
d_to = nx.single_source_dijkstra_path_length(graph, src_node, weight="length")
for tgt in targets:
t = t_to.get(tgt, np.inf)
d = d_to.get(tgt, np.inf)
for j in by_node[tgt]:
for i in src_rows:
time_s[i, j] = t
dist_m[i, j] = d
np.fill_diagonal(time_s, 0.0)
np.fill_diagonal(dist_m, 0.0)
return TravelMatrix(point_ids, node_ids, time_s, dist_m)
|