如何基于最小生成树(MST)分析图结构并提取所有主轨迹

核心逻辑 #

最小生成树是无环连通图，主轨迹拆分遵循以下规则：

• 轨迹的端点都是图中度为1的叶子节点（仅和1个其他节点相连）
• 轨迹的拆分交汇点是度大于2的分叉节点（和3个及以上其他节点相连）
• 从每个叶子节点出发沿边遍历，遇到分叉节点或其他叶子节点时终止，即可得到独立的轨迹段，自动去重重复边即可避免重复输出
• 特殊兼容无分叉的单链MST场景（所有节点串成一条线，此时无度大于2的节点，直接返回整条连通路径）

完整实现代码 #

from scipy.sparse.csgraph import minimum_spanning_tree
from scipy.spatial.distance import cdist
import pandas as pd
import numpy as np
from collections import defaultdict

def extract_main_tracks(df: pd.DataFrame) -> list[list]:
    """
    输入：索引为节点名称、列为坐标值的DataFrame
    输出：拆分后的主轨迹列表，每个子列表为一条独立轨迹的节点顺序
    """
    node_list = df.index.tolist()
    # 计算欧氏距离矩阵与MST
    dist_mat = cdist(df.values, df.values)
    mst_res = minimum_spanning_tree(dist_mat)
    mst_arr = mst_res.toarray()

    # 构建无向邻接表
    adj_table = defaultdict(list)
    edge_coords = np.argwhere(mst_arr > 0)
    for i, j in edge_coords:
        n1, n2 = node_list[i], node_list[j]
        adj_table[n1].append(n2)
        adj_table[n2].append(n1)
    
    # 统计节点度，识别叶子节点、分叉节点
    node_degree = {node: len(neighbors) for node, neighbors in adj_table.items()}
    leaf_nodes = [node for node, deg in node_degree.items() if deg == 1]
    junction_nodes = {node for node, deg in node_degree.items() if deg > 2}

    visited_edges = set()
    raw_tracks = []

    # 从每个叶子节点出发遍历提取轨迹段
    for leaf in leaf_nodes:
        current_path = [leaf]
        cur_node = leaf
        pre_node = None
        while True:
            # 排除上一个节点，找下一个可走的相邻节点
            next_candidates = [n for n in adj_table[cur_node] if n != pre_node]
            if not next_candidates:
                break
            next_node = next_candidates[0]
            # 用不可变集合标记无向边，避免重复遍历
            cur_edge = frozenset((cur_node, next_node))
            if cur_edge in visited_edges:
                break
            visited_edges.add(cur_edge)
            current_path.append(next_node)
            # 遇到分叉点或其他叶子节点则终止当前轨迹
            if next_node in junction_nodes or node_degree[next_node] == 1:
                break
            pre_node = cur_node
            cur_node = next_node
        # 过滤长度不足2的无效路径
        if len(current_path) >= 2:
            raw_tracks.append(current_path)
    
    # 处理无分叉的单链MST场景
    if not junction_nodes and len(raw_tracks) == 2:
        full_track = raw_tracks[0] + raw_tracks[1][::-1][1:]
        return [full_track]
    
    return raw_tracks

示例运行效果 #

使用题目给出的测试数据运行：

# 加载示例坐标数据
df = pd.DataFrame(
    [[2, 2], [30, 2], [2, 30], [25, 25], [14,10]],
    columns=['xcord', 'ycord'],
    index=['source', 'dest1', 'dest2', 'dest3', 'dest4']
)
# 提取主轨迹
track_result = extract_main_tracks(df)
print(track_result)

基础拆分输出为：

[['source', 'dest4'], ['dest1', 'dest4'], ['dest2', 'dest4'], ['dest3', 'dest4']]

如果需要得到示例中跨中心节点的长轨迹（如['dest2', 'dest4', 'dest1']、['source', 'dest4', 'dest3']），只需在分叉节点处按照业务规则（如边的方向、距离权重、轨迹走向）对拆分出的短轨迹做两两配对拼接即可，无需修改核心遍历逻辑。

该实现兼容任意坐标点输入，无需提前指定中心节点，自动适配多分叉、单链等不同结构的MST计算结果。

内容的提问来源于stack exchange，提问作者Ruthger Righart