第11章：网络分析步骤详解

本章详细介绍 GeoPipeAgent 的三个网络分析步骤——最短路径分析、服务区分析和地理编码。这些步骤基于 networkx 和 geopy 库实现，属于 [network] 可选依赖，适用于路网分析、物流规划和地址解析等场景。

11.1 网络分析步骤概述

11.1.1 步骤清单

GeoPipeAgent 提供了三个网络分析相关步骤：

步骤 ID	功能	核心依赖	输入类型	输出类型
`network.shortest_path`	最短路径分析	networkx	路网 GeoDataFrame + 起终点	路径 GeoDataFrame
`network.service_area`	服务区分析	networkx	路网 GeoDataFrame + 中心点	服务区多边形
`network.geocode`	地理编码	geopy	地址字符串	点 GeoDataFrame

11.1.2 可选依赖安装

网络分析步骤需要安装 [network] 可选依赖：

pip install geopipeagent[network]

该依赖组包含：

包名	版本	用途
`networkx`	≥ 2.8	图论算法库，提供最短路径和图遍历
`geopy`	≥ 2.3	地理编码库，支持多种编码服务

如果未安装 [network] 依赖，调用网络分析步骤时会抛出导入错误并提示安装。与其他可选依赖一致，GeoPipeAgent 采用延迟导入策略，不会影响未使用网络功能的流水线。

11.1.3 网络分析数据流

路网数据 (GeoDataFrame)          地址字符串
      │                              │
      ▼                              ▼
┌─────────────────┐        ┌──────────────────┐
│ networkx 图构建  │        │  geopy 地理编码   │
│  线要素 → 图     │        │  地址 → 坐标      │
└────────┬────────┘        └────────┬─────────┘
         │                          │
    ┌────┴────┐                     │
    ▼         ▼                     ▼
┌────────┐ ┌────────────┐  ┌───────────────┐
│最短路径 │ │  服务区分析  │  │ 点 GeoDataFrame│
│Dijkstra│ │  ego_graph  │  └───────────────┘
└───┬────┘ └──────┬─────┘
    ▼              ▼
┌────────┐ ┌────────────┐
│路径 GDF│ │服务区多边形 │
└────────┘ └────────────┘

11.2 network.shortest_path 最短路径分析

11.2.1 功能说明

network.shortest_path 在路网 GeoDataFrame 上执行最短路径分析。它将路网线要素转化为 networkx 图结构，然后使用 Dijkstra 算法寻找两点之间的最短路径。

典型应用场景：

车辆导航路线规划
物流配送路径优化
应急救援最短到达路径
管网最短连通路径

11.2.2 参数定义

参数	类型	必需	默认值	说明
`input`	`GeoDataFrame`	✅	—	路网线要素数据
`origin`	`Point`	✅	—	起点坐标（Shapely Point 或 `$steps` 引用）
`destination`	`Point`	✅	—	终点坐标（Shapely Point 或 `$steps` 引用）
`weight_field`	`str`	❌	`None`	权重字段名（如长度、通行时间）

11.2.3 networkx 图构建

路网 GeoDataFrame 到 networkx 图的转换是最短路径分析的核心前置步骤。每条线要素的起点和终点成为图节点，线要素本身成为图边。

import networkx as nx

def _build_graph(road_gdf, weight_field=None):
    """将路网 GeoDataFrame 转换为 networkx 无向图"""
    G = nx.Graph()

    for idx, row in road_gdf.iterrows():
        line = row.geometry
        # 提取线要素的起点和终点
        start = (line.coords[0][0], line.coords[0][1])
        end = (line.coords[-1][0], line.coords[-1][1])

        # 计算权重
        if weight_field and weight_field in row.index:
            weight = row[weight_field]
        else:
            weight = line.length  # 默认使用几何长度

        # 添加边，携带权重和几何信息
        G.add_edge(start, end, weight=weight, geometry=line)

    return G

图构建过程的关键细节：

要素	图对象	说明
线要素起点坐标	图节点	`(x, y)` 元组作为节点标识
线要素终点坐标	图节点	`(x, y)` 元组作为节点标识
线要素	图边	连接起点到终点的边
`weight_field` 值	边权重	如未指定则使用几何长度
线几何	边属性	用于最终路径几何重建

11.2.4 Dijkstra 最短路径算法

networkx 使用 Dijkstra 算法查找最短路径：

Dijkstra 算法工作流程：

起点 A ──(3)──→ B ──(2)──→ D（终点）
  │              │            ↑
  └──(1)──→ C ──(5)──→──────┘

1. 初始化：dist[A]=0, 其余=∞
2. 访问 A：更新邻居 dist[B]=3, dist[C]=1
3. 访问 C（最小未访问）：更新 dist[D]=1+5=6
4. 访问 B：更新 dist[D]=min(6, 3+2)=5
5. 访问 D：最短路径 A→B→D，总距离=5

源码中的关键调用：

@step(id="network.shortest_path", ...)
def shortest_path(ctx):
    import networkx as nx

    gdf = ctx.input("input")
    origin = ctx.param("origin")        # Shapely Point
    destination = ctx.param("destination")  # Shapely Point
    weight_field = ctx.param("weight_field")

    # 1. 构建 networkx 图
    G = _build_graph(gdf, weight_field)

    # 2. 找到最近的图节点（snap 到路网）
    origin_node = _nearest_node(G, (origin.x, origin.y))
    dest_node = _nearest_node(G, (destination.x, destination.y))

    # 3. Dijkstra 最短路径
    path_nodes = nx.shortest_path(
        G, source=origin_node, target=dest_node, weight="weight"
    )

    # 4. 提取路径上的边几何
    path_edges = []
    total_distance = 0.0
    for i in range(len(path_nodes) - 1):
        edge_data = G.edges[path_nodes[i], path_nodes[i + 1]]
        path_edges.append(edge_data["geometry"])
        total_distance += edge_data["weight"]

    # 5. 构建输出 GeoDataFrame
    result = gpd.GeoDataFrame(
        {"path_segment": range(len(path_edges))},
        geometry=path_edges,
        crs=gdf.crs,
    )

    stats = {"total_distance": total_distance}
    return StepResult(output=result, stats=stats)

11.2.5 权重字段的作用

weight_field 参数决定了”最短”的含义：

weight_field	含义	典型列名
不指定	几何长度最短	—
长度字段	物理距离最短	`length_m`, `distance`
时间字段	通行时间最短	`travel_time`, `minutes`
成本字段	通行成本最低	`cost`, `toll`

11.2.6 使用示例

name: 最短路径分析
steps:
  # 1. 读取路网数据
  - id: read_roads
    step: io.read_vector
    params:
      path: "data/road_network.gpkg"

  # 2. 地理编码起点
  - id: geocode_origin
    step: network.geocode
    params:
      address: "北京市海淀区中关村大街1号"

  # 3. 地理编码终点
  - id: geocode_dest
    step: network.geocode
    params:
      address: "北京市朝阳区建国门外大街1号"

  # 4. 计算最短路径
  - id: find_path
    step: network.shortest_path
    params:
      input: "$steps.read_roads"
      origin: "$steps.geocode_origin"
      destination: "$steps.geocode_dest"
      weight_field: "length_m"

  # 5. 保存路径结果
  - id: save_path
    step: io.write_vector
    params:
      input: "$steps.find_path"
      path: "output/shortest_path.gpkg"
      format: gpkg

11.3 network.service_area 服务区分析

11.3.1 功能说明

network.service_area 计算从给定中心点出发，在指定距离或时间范围内可到达的区域。结果为一组服务区多边形，通常用于：

医院 / 消防站服务覆盖范围分析
商业选址可达性分析
公共交通站点覆盖评估
等时圈（isochrone）和等距圈（isodistance）生成

11.3.2 参数定义

参数	类型	必需	默认值	说明
`input`	`GeoDataFrame`	✅	—	路网线要素数据
`origin`	`Point`	✅	—	服务中心点坐标
`cutoffs`	`list[float]`	✅	—	距离/时间截断值列表
`weight_field`	`str`	❌	`None`	权重字段名

11.3.3 等时圈与等距圈

等时圈和等距圈是服务区分析的两种主要输出形式：

等距圈 (Isodistance)              等时圈 (Isochrone)
weight_field = "length_m"         weight_field = "travel_time"

cutoffs: [500, 1000, 1500]        cutoffs: [5, 10, 15]  (分钟)

    ╭───────────────╮                 ╭─────────────╮
    │  ╭─────────╮  │                 │ ╭─────────╮ │
    │  │ ╭─────╮ │  │                 │ │ ╭─────╮ │ │
    │  │ │  ★  │ │  │                 │ │ │  ★  │ │ │
    │  │ ╰─────╯ │  │                 │ │ ╰─────╯ │ │
    │  │  1000m  │  │                 │ │  10min  │ │
    │  ╰─────────╯  │                 │ ╰─────────╯ │
    │    1500m       │                 │   15min     │
    ╰───────────────╯                 ╰─────────────╯

   同心圆（受路网拓扑影响            形状不规则（受路网密度
    实际为不规则形状）                 和限速影响）

11.3.4 ego_graph 实现原理

服务区分析的核心是 networkx 的 ego_graph 函数，它提取以指定节点为中心、在给定距离/权重范围内的子图：

def _compute_service_area(G, center_node, cutoff, weight="weight"):
    """使用 ego_graph 计算单个服务区"""
    # ego_graph: 提取距离中心节点 cutoff 范围内的所有节点和边
    subgraph = nx.ego_graph(
        G, center_node, radius=cutoff, distance=weight
    )

    # 收集子图中所有边的几何
    edges_geom = []
    for u, v, data in subgraph.edges(data=True):
        if "geometry" in data:
            edges_geom.append(data["geometry"])

    # 将边几何合并并生成凸包或缓冲区作为服务区多边形
    from shapely.ops import unary_union
    merged = unary_union(edges_geom)
    service_polygon = merged.buffer(50)  # 缓冲区宽度

    return service_polygon

ego_graph 的工作原理：

完整路网图 G：                    ego_graph(G, ★, radius=1000)：

  A──300──B──400──C──500──D       A──300──B──400──C
  │       │       │                │       │
  200     600     300              200     600
  │       │       │                │       │
  ★──500──E──200──F──800──G       ★──500──E──200──F
  │                                │
  400                              400
  │                                │
  H──300──I                        H──300──I

  图中数字表示边权重                只保留距离 ★ ≤ 1000 的节点和边

11.3.5 多级服务区生成

通过 cutoffs 参数可以一次生成多个嵌套的服务区：

@step(id="network.service_area", ...)
def service_area(ctx):
    import networkx as nx

    gdf = ctx.input("input")
    origin = ctx.param("origin")
    cutoffs = ctx.param("cutoffs")      # 如 [500, 1000, 1500]
    weight_field = ctx.param("weight_field")

    # 1. 构建图
    G = _build_graph(gdf, weight_field)
    center_node = _nearest_node(G, (origin.x, origin.y))

    # 2. 为每个 cutoff 生成服务区
    polygons = []
    cutoff_values = []
    for cutoff in sorted(cutoffs):
        poly = _compute_service_area(G, center_node, cutoff)
        polygons.append(poly)
        cutoff_values.append(cutoff)

    # 3. 构建输出 GeoDataFrame
    result = gpd.GeoDataFrame(
        {"cutoff": cutoff_values},
        geometry=polygons,
        crs=gdf.crs,
    )

    return StepResult(output=result, stats={
        "center": (origin.x, origin.y),
        "cutoffs": cutoff_values,
        "service_areas_count": len(polygons),
    })

11.3.6 使用示例

name: 医院服务区分析
steps:
  - id: read_roads
    step: io.read_vector
    params:
      path: "data/road_network.gpkg"

  - id: hospital_service
    step: network.service_area
    params:
      input: "$steps.read_roads"
      origin:
        type: Point
        coordinates: [116.3975, 39.9087]
      cutoffs: [1000, 3000, 5000]  # 1km, 3km, 5km 服务半径
      weight_field: "length_m"

  - id: save_service_area
    step: io.write_vector
    params:
      input: "$steps.hospital_service"
      path: "output/hospital_service_areas.gpkg"
      format: gpkg

11.4 network.geocode 地理编码

11.4.1 功能说明

network.geocode 使用 geopy 库将地址字符串转换为地理坐标点。它支持单个地址和批量地址编码，默认使用 Nominatim（OpenStreetMap）作为编码服务提供者。

应用场景：

地址数据转换为空间坐标
客户地址批量定位
POI（兴趣点）数据采集
与路网分析步骤配合提供起终点

11.4.2 参数定义

参数	类型	必需	默认值	说明
`address`	`str` 或 `list[str]`	✅	—	地址字符串或地址列表
`provider`	`str`	❌	`"nominatim"`	地理编码服务提供者
`exactly_one`	`bool`	❌	`True`	是否只返回单个最佳匹配

11.4.3 geopy Nominatim 编码

geopy 是 Python 地理编码客户端库，支持多种在线编码服务。GeoPipeAgent 默认使用免费的 Nominatim 服务：

from geopy.geocoders import Nominatim

# Nominatim 基于 OpenStreetMap 数据
geolocator = Nominatim(user_agent="geopipeagent")
location = geolocator.geocode("北京市天安门")

print(location.latitude)   # 39.9054
print(location.longitude)  # 116.3976
print(location.address)    # 天安门, 东城区, 北京市, ...

支持的编码服务提供者：

Provider	说明	是否免费	限流
`nominatim`	OpenStreetMap Nominatim	✅ 免费	1 请求/秒
`google`	Google Maps Geocoding API	❌ 付费	按配额
`arcgis`	ArcGIS World Geocoding	部分免费	按配额
`baidu`	百度地图 API	部分免费	按配额

11.4.4 批量地理编码

当 address 参数为列表时，network.geocode 自动执行批量编码，并对每个地址依次请求：

@step(id="network.geocode", ...)
def geocode(ctx):
    from geopy.geocoders import Nominatim

    address = ctx.param("address")
    provider = ctx.param("provider") or "nominatim"
    exactly_one = ctx.param("exactly_one")
    if exactly_one is None:
        exactly_one = True

    # 初始化编码器
    geolocator = Nominatim(user_agent="geopipeagent")

    # 统一为列表处理
    addresses = [address] if isinstance(address, str) else address

    results = []
    for addr in addresses:
        location = geolocator.geocode(addr, exactly_one=exactly_one)
        if location:
            results.append({
                "address": addr,
                "resolved_address": location.address,
                "geometry": Point(location.longitude, location.latitude),
            })
        else:
            results.append({
                "address": addr,
                "resolved_address": None,
                "geometry": None,
            })

    # 过滤成功结果，构建 GeoDataFrame
    valid = [r for r in results if r["geometry"] is not None]
    result = gpd.GeoDataFrame(valid, geometry="geometry", crs="EPSG:4326")

    stats = {
        "total_addresses": len(addresses),
        "geocoded_count": len(valid),
        "failed_count": len(addresses) - len(valid),
    }

    return StepResult(output=result, stats=stats)

11.4.5 编码结果数据结构

输出 GeoDataFrame 包含以下列：

列名	类型	说明
`address`	`str`	原始输入地址
`resolved_address`	`str`	编码服务返回的标准化地址
`geometry`	`Point`	编码后的坐标点（WGS 84）

11.4.6 使用示例

单地址编码：

- id: geocode_location
  step: network.geocode
  params:
    address: "上海市浦东新区世纪大道100号"

批量地址编码：

- id: geocode_branches
  step: network.geocode
  params:
    address:
      - "北京市朝阳区望京SOHO"
      - "上海市黄浦区南京东路步行街"
      - "广州市天河区珠江新城"
      - "深圳市南山区科技园"
    provider: nominatim
    exactly_one: true

11.4.7 注意事项

使用地理编码步骤时需注意：

限流控制：Nominatim 限制 1 请求/秒，批量编码时自动添加延时
结果精度：编码精度取决于服务提供者的数据覆盖范围
网络依赖：编码需要访问外部网络服务，离线环境不可用
CRS 说明：编码结果始终为 WGS 84（EPSG:4326），如需其他投影需后续转换

11.5 网络分析组合实战

11.5.1 城市物流配送路径规划

以下流水线演示了一个完整的物流配送场景：将仓库和客户地址编码为坐标，计算最短配送路径，并生成仓库的服务覆盖范围。

name: 城市物流配送路径规划
description: >
  地理编码仓库和客户地址，计算最短配送路径，
  分析仓库服务覆盖范围

variables:
  warehouse_address: "北京市大兴区物流园区A座"
  customer_address: "北京市海淀区清华科技园"

steps:
  # ─── 数据准备 ───
  - id: read_roads
    step: io.read_vector
    params:
      path: "data/beijing_road_network.gpkg"

  # ─── 地理编码 ───
  - id: geocode_warehouse
    step: network.geocode
    params:
      address: "$var.warehouse_address"

  - id: geocode_customer
    step: network.geocode
    params:
      address: "$var.customer_address"

  # ─── 最短路径分析 ───
  - id: delivery_route
    step: network.shortest_path
    params:
      input: "$steps.read_roads"
      origin: "$steps.geocode_warehouse"
      destination: "$steps.geocode_customer"
      weight_field: "travel_time"

  # ─── 仓库服务区分析 ───
  - id: warehouse_service
    step: network.service_area
    params:
      input: "$steps.read_roads"
      origin: "$steps.geocode_warehouse"
      cutoffs: [5, 10, 15, 30]  # 分钟
      weight_field: "travel_time"

  # ─── 保存结果 ───
  - id: save_route
    step: io.write_vector
    params:
      input: "$steps.delivery_route"
      path: "output/delivery_route.gpkg"
      format: gpkg

  - id: save_service_area
    step: io.write_vector
    params:
      input: "$steps.warehouse_service"
      path: "output/warehouse_service_area.gpkg"
      format: gpkg

11.5.2 应急响应分析流水线

name: 消防站应急响应分析
description: 计算消防站覆盖范围和到达事故点的最短路径

steps:
  - id: read_roads
    step: io.read_vector
    params:
      path: "data/city_roads.gpkg"

  # 消防站位置编码
  - id: geocode_station
    step: network.geocode
    params:
      address: "杭州市西湖区消防救援站"

  # 事故点位置编码
  - id: geocode_incident
    step: network.geocode
    params:
      address: "杭州市西湖区文三路与教工路交叉口"

  # 消防站服务覆盖范围（5分钟、10分钟响应圈）
  - id: response_coverage
    step: network.service_area
    params:
      input: "$steps.read_roads"
      origin: "$steps.geocode_station"
      cutoffs: [5, 10]
      weight_field: "travel_time"

  # 到事故点最短路径
  - id: response_route
    step: network.shortest_path
    params:
      input: "$steps.read_roads"
      origin: "$steps.geocode_station"
      destination: "$steps.geocode_incident"
      weight_field: "travel_time"

  # 保存结果
  - id: save_coverage
    step: io.write_vector
    params:
      input: "$steps.response_coverage"
      path: "output/fire_station_coverage.gpkg"
      format: gpkg

  - id: save_response_route
    step: io.write_vector
    params:
      input: "$steps.response_route"
      path: "output/response_route.gpkg"
      format: gpkg

11.5.3 批量客户地址分析

name: 批量客户地址编码与服务区判断
description: 批量编码客户地址并判断其是否在仓库服务区内

steps:
  - id: read_roads
    step: io.read_vector
    params:
      path: "data/road_network.gpkg"

  # 批量编码客户地址
  - id: geocode_customers
    step: network.geocode
    params:
      address:
        - "成都市武侯区科华北路65号"
        - "成都市锦江区春熙路"
        - "成都市高新区天府大道北段"
        - "成都市青羊区宽窄巷子"
      provider: nominatim

  # 仓库服务区
  - id: warehouse_area
    step: network.service_area
    params:
      input: "$steps.read_roads"
      origin:
        type: Point
        coordinates: [104.0668, 30.5728]
      cutoffs: [5000, 10000]
      weight_field: "length_m"

  # 空间连接——判断客户是否在服务区内
  - id: customers_in_area
    step: vector.spatial_join
    params:
      input: "$steps.geocode_customers"
      join: "$steps.warehouse_area"
      how: "inner"
      predicate: "within"

  - id: save_result
    step: io.write_vector
    params:
      input: "$steps.customers_in_area"
      path: "output/customers_in_service_area.gpkg"
      format: gpkg

11.6 本章小结

本章详细介绍了 GeoPipeAgent 的三个网络分析步骤：

network.shortest_path：将路网 GeoDataFrame 转换为 networkx 图结构，使用 Dijkstra 算法计算最短路径。通过 weight_field 参数可灵活定义”最短”的语义——距离最短、时间最短或成本最低。输出为路径线段 GeoDataFrame 并携带 total_distance 统计。
network.service_area：基于 networkx 的 ego_graph 函数提取可达子图，为指定中心点生成多级服务区多边形。cutoffs 参数支持同时生成多个嵌套等距圈或等时圈，广泛应用于设施选址和覆盖评估。
network.geocode：通过 geopy 库调用 Nominatim 等地理编码服务，将地址字符串转换为点坐标。支持单地址和批量地址编码，输出 WGS 84 坐标系下的点 GeoDataFrame。

这三个步骤属于 [network] 可选依赖组，需要额外安装 networkx 和 geopy。它们可与 IO 步骤、矢量分析步骤自由组合，构建完整的路网分析、物流规划和地址解析工作流。

下一章：数据质检（QC）步骤详解 →