路网看起来只是一张带节点和边的图，但真正难的地方不是画线，而是想清楚它和地图的关系、它怎么生成、怎么编辑、怎么和云端同步。这个项目把路网当成独立于地图的语义资产，核心判断只有一句话：地图回答“这里长什么样”，路网回答“可以怎么走”。

路网为什么不能塞进地图

同一张图，巡逻路线可能今天改、明天加；同一条路网，底图也可能换一张重新扫描。如果把路网做成地图的一个字段，任何一次路线调整都会牵连地图资产。

所以项目把两者拆开：

地图资产 MapAsset 负责点云和定位。
路网资产 NetworkAsset 通过 map_refs 单向引用地图。
地图不知道谁在引用它。

消费路网时，坐标系必须取 primary 底图的 global_frame。如果拿 A 图的路网在 B 图上跑，节点坐标看起来正常，实际却是跨帧规划。因此 navigate 里加了帧守卫，导航图和定位当前图不一致时直接拒绝出发。

路网的数据形状

一个路网版本里，route_graph 表达几何，routes 表达任务。

route_graph 是导航真正消费的结构：

{
  "graph_id": "site-net",
  "global_frame": "map",
  "nodes": [
    {"node_id": "N1", "pose": {"x": 1.0, "y": 2.0, "z": 0.0, "yaw": 0.0}, "is_goal": true}
  ],
  "edges": [
    {
      "edge_id": "E1",
      "source_node_id": "N1",
      "target_node_id": "N2",
      "bidirectional": true,
      "length_m": 5.2,
      "road_width_m": 1.2,
      "waypoints": [
        {"x": 1.0, "y": 2.0, "z": 0.0},
        {"x": 1.2, "y": 2.3, "z": 0.0}
      ]
    }
  ]
}

节点是结构，边是几何。边的 waypoints 特别关键：它不是两个节点之间的直线，而是狗实际可走的那条中心线。导航沿着 waypoints 走，如果这里存直线，就可能穿墙。

routes 则是一组有序 node_ids，例如闭环巡逻 [N1, N2, N3, N1]。一份 route_graph 上可以挂多条路线，这样同一张路网既能跑这条巡逻，也能跑那条巡逻。

从建图轨迹自动生成

手画路网最大的问题是“你以为这里能走，实际狗过不去”。项目的推荐做法是先把狗带过去建图，再用那条真实轨迹生成路网。

轨迹是唯一被真机证明可通行的路径：宽度够、地面够平、没有过不去的坎。转换时先抽稀，再挑节点：

DEFAULTS = {
    "every_m": 3.0,        # 直路上每隔多远必放一个节点
    "corner_m": 0.35,      # 拐点容差:节点折线偏离实际轨迹超过它就加节点
    "simplify_m": 0.15,    # 航点抽稀间距
    "close_loop_m": 2.0,   # 首尾相距小于它就补一条闭合边
}

pts = _thin(raw, o["simplify_m"])
idx = _dp(pts, o["corner_m"])
idx = _enforce_spacing(pts, idx, o["every_m"])

节点选择分两步：

用 Douglas–Peucker 挑出拐点，容差 corner_m 表示节点折线允许偏离实际轨迹多少米。
再按 every_m 补点，保证长直路中间也有可选的巡逻点。

为什么不能用转角阈值？实测 20 米的室内环，如果用转角阈值，会被步态摆动喂出 22 个节点。那些不是结构，是狗走路时的噪声。

节点要稀疏，航点要密集。waypoints 取的是抽稀后的原始轨迹点，默认 simplify_m=0.15，既去掉重复点，又保证折线贴近原路径。

闭环也很重要。如果首尾足够近，会补一条闭环边，让巡逻可以循环。但这条边狗没有走过，所以 waypoints 只能给两个端点之间的直线，同时打上来源标记，让人知道它和其他边不同。

编辑器：复用云端模块，不自己重写

路网编辑器不是从零写出来的。项目把云端已有的路网编辑器模块同步到端侧，只写一个适配层。

适配层做三件事：

数据进：从端侧接口拿 editor-snapshot。
数据出：编辑器的保存回调把图写到端侧 route-graph 接口。
外壳：把编辑器标记和样式注入 Shadow Root，避免污染工作台页面。

已发布的版本默认只读。需要改已经发布的路网时，正确做法是新增一个版本，而不是把旧版本改回草稿，否则历史版本就没了。

资产台账和同步

端侧用文件版资产台账管理路网版本，不依赖 ROS。每个版本下保存一份归一化后的 route_graph。

同步最容易犯的错是拿端侧哈希和云端哈希直接比较。端侧存的是归一化后的图，云端存的是编辑器状态，两者的字节表示本来就不同。直接比较会把每个版本都判成冲突。

项目用两个基线解决：

端侧变了 = edge_hash  != base_edge
云端变了 = cloud_hash != base_cloud

都没变 -> in_sync
只端侧变 -> edge_ahead
只云端变 -> cloud_ahead
都变 -> conflict

云端下行时只做快进，不覆盖端侧的活。如果端侧没动过，就直接更新为云端版本；如果端侧动过，就把云端版本存到单独的 incoming 文件，标记冲突等人选择。

这里有一个真实踩坑：曾经把“从没同步过”判断成“本地没动过”，导致端侧刚生成的 11 个节点被云端空图覆盖。修正后的规则是，没有基线时只有本地也是空的才允许快进。

绑定和入网

“设为当前”会同时更新两个“当前”：

路网绑定：路网资产、版本、底图和路线。
当前地图：costmap、navigate 和 localization 真正读取的地图。

如果只更新其中一个，就会发生“导航栈用室外图、路网给室内节点”的对不上情况。两者看起来都正常，直到巡逻开始按错误坐标系走路。

上层按路网排线时，不是找最近节点，而是把当前位姿投影到最近的边。原因很具体：找最近节点会让狗掉头。投影到边后，再跑 Dijkstra，拼出沿边中心线的折线，交给 navigate 逐段走。

关键代码示例

轨迹生成路网：节点和航点不是一回事

先对原始轨迹抽稀，再用 Douglas–Peucker 挑结构点，最后补长直路节点。DP 用迭代实现，因为轨迹长度由用户走了多久决定，递归写长了会爆栈：

def _dp(pts, eps: float) -> List[int]:
    """Douglas–Peucker,返回保留点的下标。迭代实现:递归在长轨迹上会爆栈。"""
    n = len(pts)
    if n <= 2 or eps <= 0:
        return list(range(n))
    keep = {0, n - 1}
    stack = [(0, n - 1)]
    while stack:
        a, b = stack.pop()
        if b - a < 2:
            continue
        best, bi = -1.0, -1
        for i in range(a + 1, b):
            d = _perp(pts[i], pts[a], pts[b])
            if d > best:
                best, bi = d, i
        if best > eps:
            keep.add(bi)
            stack.append((a, bi))
            stack.append((bi, b))
    return sorted(keep)


def _enforce_spacing(pts, idx: List[int], every_m: float) -> List[int]:
    """DP 只在"形状变了"时给点 —— 长直走廊上一个都不给。
    这里按累计弧长补,保证巡逻能在直路中间挑到抓手。"""
    if every_m <= 0 or len(idx) < 2:
        return idx
    out = [idx[0]]
    for k in range(1, len(idx)):
        a, b = idx[k - 1], idx[k]
        acc = 0.0
        for i in range(a + 1, b):
            acc += _dist(pts[i - 1], pts[i])
            if acc >= every_m:
                out.append(i)
                acc = 0.0
        out.append(b)
    return sorted(set(out))

边的 waypoints 必须取抽稀后的原始轨迹点，不能取两个节点之间的直线：

for k in range(len(idx) - 1):
    a, b = idx[k], idx[k + 1]
    wp = pts[a:b + 1]
    length = sum(_dist(wp[i - 1], wp[i]) for i in range(1, len(wp)))
    edges.append({
        "edge_id": f"E_TRK_{k + 1}",
        "source_node_id": nodes[k]["node_id"],
        "target_node_id": nodes[k + 1]["node_id"],
        "bidirectional": True,
        "waypoints": [{"x": round(p[0], 3), "y": round(p[1], 3), "z": round(p[2], 3)}
                      for p in wp],
        "length_m": round(length, 2),
        "road_width_m": o["width_m"],
    })

节点稀、航点密，这是两个不同的东西。节点是给人选巡逻点用的，航点是给 navigate 真正驱动折线用的。混成一个粒度，要么节点多得没法挑，要么狗切角撞墙。

同步判据：两端各自和自己的基线比

同步状态不会拿端侧哈希和云端哈希直接比，因为两边的字节表示本来就不同：

def sync_state(edge_hash, cloud_hash, base_edge, base_cloud) -> str:
    """四态判定:各自判自己变没变,绝不拿两边的哈希互比。
    端侧存的是归一化后的图,云端存的是 editorState.routeGraph,
    同一份路网在两边的字节表示本来就不同。
    """
    if not cloud_hash:
        return SYNC_LOCAL_ONLY if edge_hash else SYNC_UNKNOWN
    if not base_edge and not base_cloud:
        # 从没同步过:两边都有内容,谁也不比谁新 —— 保守判冲突
        return SYNC_CONFLICT
    edge_changed = bool(base_edge) and edge_hash != base_edge
    cloud_changed = bool(base_cloud) and cloud_hash != base_cloud
    if edge_changed and cloud_changed:
        return SYNC_CONFLICT
    if edge_changed:
        return SYNC_EDGE_AHEAD
    if cloud_changed:
        return SYNC_CLOUD_AHEAD
    return SYNC_IN_SYNC

真正危险的落盘边界在云端下行。从没同步过 不等于 本地没动过，端侧刚由轨迹生成的路网就是“没有基线 + 本地有内容”：

never_synced = not s.get("base_edge_hash")
local_empty = not ((local.get("nodes") or []) or (local.get("edges") or []))

if never_synced:
    # 没有基线时,只有本地也是空的才允许快进
    edge_changed = (not local_empty) and lh != ch
else:
    edge_changed = lh != s.get("base_edge_hash")

if not edge_changed:
    # 端侧没动过 → 快进
    self._write_json(graph, "network", code, "versions", ver, "route_graph.json")
    action, state = ("in_sync" if lh == content_hash(graph) else "fast_forward"), SYNC_IN_SYNC
else:
    # 端侧动过 —— 不写 route_graph。云端那份单独存着供对比/选边。
    self._write_json(graph, "network", code, "versions", ver, "cloud_incoming.json")
    action, state = "conflict", SYNC_CONFLICT