-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathendpoints.py
More file actions
103 lines (80 loc) · 3.5 KB
/
Copy pathendpoints.py
File metadata and controls
103 lines (80 loc) · 3.5 KB
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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
"""
endpoints.py — 最远端点对计算工具
在城市路网中找到"实际路径距离最远"的两个节点，作为算法的极限压测场景。
方法（双向 BFS 近似图直径）：
1. 从随机节点 s₀ 跑完整 Dijkstra，找距离最远的节点 u
2. 从 u 再跑完整 Dijkstra，找距离最远的节点 v
结果 (u, v) 是近似最远端点对，误差通常 < 5%
"""
import heapq
import json
import os
def find_farthest_pair(G, seed_node=None):
"""
用双向 BFS 近似法找最远端点对。
参数：
G : NetworkX 有向图
seed_node : 第一轮 Dijkstra 的起始节点，默认取图中第一个节点
返回：
(source, target, dist_m)
source, target : 两个节点 ID
dist_m : 两点间最短路径距离（米）
"""
nodes = list(G.nodes)
if seed_node is None:
# 取图中坐标最西南角的节点作为种子（边界点更容易是直径端点）
seed_node = min(nodes, key=lambda n: (G.nodes[n]['x'], G.nodes[n]['y']))
print(f" [端点] 第一轮 Dijkstra（从种子节点出发）...", flush=True)
dist1, _ = _dijkstra_full(G, seed_node)
# 找离种子最远的节点 u
u = max(dist1, key=dist1.get)
print(f" [端点] 找到 u = {u}，距种子 {dist1[u]/1000:.1f} km", flush=True)
print(f" [端点] 第二轮 Dijkstra（从 u 出发）...", flush=True)
dist2, _ = _dijkstra_full(G, u)
# 找离 u 最远的节点 v
v = max(dist2, key=dist2.get)
print(f" [端点] 找到 v = {v}，u→v 路径距离 {dist2[v]/1000:.1f} km", flush=True)
return u, v, dist2[v]
def _dijkstra_full(G, source):
"""
完整 Dijkstra，返回从 source 到所有可达节点的最短距离和父节点字典。
"""
dist = {source: 0.0}
parent = {source: None}
heap = [(0.0, source)]
while heap:
d, node = heapq.heappop(heap)
if d > dist.get(node, float('inf')):
continue
for nb, edge_dict in G[node].items():
min_len = min(float(data.get('length', 0) or 0) for data in edge_dict.values())
nd = d + min_len
if nd < dist.get(nb, float('inf')):
dist[nb] = nd
parent[nb] = node
heapq.heappush(heap, (nd, nb))
return dist, parent
def save_endpoints(city_filename, source, target, dist_m, G, results_dir):
"""将端点信息保存到 results/<城市>/endpoints.json。"""
city_dir = os.path.join(results_dir, city_filename)
os.makedirs(city_dir, exist_ok=True)
info = {
'source': source,
'target': target,
'source_xy': [G.nodes[source]['x'], G.nodes[source]['y']],
'target_xy': [G.nodes[target]['x'], G.nodes[target]['y']],
'source_lonlat': [G.nodes[source].get('lon', 0), G.nodes[source].get('lat', 0)],
'target_lonlat': [G.nodes[target].get('lon', 0), G.nodes[target].get('lat', 0)],
'path_distance_km': round(dist_m / 1000, 3),
}
with open(os.path.join(city_dir, 'endpoints.json'), 'w') as f:
json.dump(info, f, indent=2, ensure_ascii=False)
return info
def load_endpoints(city_filename, results_dir):
"""从已保存的 endpoints.json 读取端点，避免重复计算。"""
path = os.path.join(results_dir, city_filename, 'endpoints.json')
if not os.path.exists(path):
return None
with open(path) as f:
d = json.load(f)
return d['source'], d['target']