多个shapely.geometry.LineString路线重叠检测及算法优化问询
我有一长串存在部分重叠的shapely.geometry.LineString对象,具体为德国高速公路网络上的行车路线。每个LineString都有关联数值,我这里对应的是路线延误时长和通行车辆数。我的目标是找到这些不同LineString之间的重叠路段,对重叠段对应的关联数值进行求和,最终效果类似下图可视化结果:

我目前的处理方法大致如下(代码附在末尾):
- 维护一个已完成校验的路线列表
- 遍历每条新路线:校验新路线与已校验路线之间是否存在重叠
- 如果存在重叠,将重叠段的数值合并赋值,仅对剩余非重叠部分继续后续校验
该方法可以运行并得到预期结果,但问题是当待校验路线数量极大时,算法运行时间会变得过长,原因是需要计算相交的LineString数量会逐步增加。请问是否有现成的算法、工具包可以解决这个问题?或者有什么好的建议可以加速该算法?
当前代码实现
def intersects_rectangles(bounds, other_bounds): return not (bounds[2] < other_bounds[0] or bounds[0] > other_bounds[2] or bounds[3] < other_bounds[1] or bounds[1] > other_bounds[3]) def main_function(route_df) -> Dict[tuple, List[Union[LineString, float]]]: # route_df的索引指向存储LineString的文件,列存储每条路线的关联数值 table_merged: Dict[tuple, List[Union[LineString, float]]] = {} routes_covered = 0 tot_routes = len(route_df) print(f'Creating a merged view of all routes. Nr. routes: {tot_routes}') for (route_name, car_name), line in route_df.iterrows(): with open('buffer_data/' + route_name + '/' + BufferFileHierarchy.route.value, 'rb') as f: linestring: LineString linestring, _ = pickle.load(f) current_cum_delay = line['sum'] current_count = line['count'] if len(table_merged) == 0: table_merged[(route_name, )] = [linestring, linestring.buffer(0.001).buffer(0), current_cum_delay, current_count] # 做buffer是为了包含对向车道的LineString continue for other_contained_routes in list(table_merged.keys()): [other_linestring, other_linestring_buffer, other_cum_delay, other_cum_count] = table_merged[other_contained_routes] if route_name in other_contained_routes: continue seg_intersects = linestring.intersection(other_linestring_buffer) if seg_intersects.is_empty: continue if not intersects_rectangles(linestring.bounds, other_linestring.bounds): continue if seg_intersects.length > 0.0002: seg_intersects_buffer = seg_intersects.buffer(0.001).buffer(0) linestring_remainder = linestring.difference(seg_intersects_buffer) other_linestring_remainder = other_linestring.difference(seg_intersects_buffer) table_merged[(*other_contained_routes, route_name)] = [seg_intersects, seg_intersects_buffer, current_cum_delay + other_cum_delay, current_count + other_cum_count] if other_linestring_remainder.is_empty: del table_merged[other_contained_routes] else: table_merged[other_contained_routes] = [other_linestring_remainder, other_linestring_remainder.buffer(0.001).buffer(0), other_cum_delay, other_cum_count] linestring = linestring_remainder if linestring.is_empty: break if not linestring.is_empty: table_merged[(route_name, )] = [linestring, linestring.buffer(0.001).buffer(0), current_cum_delay, current_count] routes_covered += 1 if routes_covered % 100 == 0: progress_bar.print_progress_bar(routes_covered, tot_routes) progress_bar.print_progress_bar(routes_covered, tot_routes) return table_merged
调整边界检查顺序
你现有代码先计算几何相交再判断外接矩形是否相交,完全搞反了优先级。外接矩形相交判断是纯数值计算,耗时只有几何相交计算的几百分之一,应该先做矩形过滤,排除完全不相交的线段后再做几何相交计算,能直接砍掉绝大多数无效计算。引入空间索引降低相交筛选复杂度
你现在每次遍历所有已处理的线段,时间复杂度是O(n²),数据量大的时候必然变慢。可以用R树空间索引,每次只筛选出外接矩形和当前线段相交的候选线段,把单次筛选的复杂度降到O(logn),整体性能会有数量级提升。使用现成的矢量叠加工具避免手写循环
不用自己实现重叠合并逻辑,可以把所有LineString转成GeoDataFrame格式,直接调用现成的overlay函数做交集计算,或者用针对路网优化的线段合并工具,底层是优化过的C++实现,比纯Python手写循环快得多。路网匹配优化(最适合你的场景)
你处理的是德国高速公路的路线,本身是有标准拓扑结构的。可以先把所有LineString匹配到公开的德国高速路网拓扑上,每条路线拆分成交互的路网路段(edge)列表,之后直接按路段ID对延误和车流量求和就行,完全不需要做几何相交计算,性能会提升至少10倍以上。小的代码优化点
- 提前计算好所有LineString的buffer,不要每次处理的时候重复计算
buffer(0.001).buffer(0)的操作可以简化,直接用平端参数做单次buffer就可以得到没有圆角的缓冲区,同时自动修复拓扑错误- 路线加载和预处理可以用多进程并行处理,这部分逻辑没有依赖,并行改造难度很低
内容的提问来源于stack exchange,提问作者C Hecht

