en/latest/UserGuide/%E8%B7%AF%E5%BE%84%E5%8C%B9%E9%85%8D/

[giscus[bot]]

en/latest/UserGuide/%E8%B7%AF%E5%BE%84%E5%8C%B9%E9%85%8D/

在python中轻松进行路径匹配

https://gotrackit.readthedocs.io/en/latest/UserGuide/%E8%B7%AF%E5%BE%84%E5%8C%B9%E9%85%8D/

[zdsjjtTLG]

这里是评论区

为了避免重复提问，请大家在手册下方的评论区提问，历史问题都会保留在评论区，方便大家快速查阅，评论区的内容也会自动同步到项目主页的讨论区

[gjq-emperor]

有一个区域.osm文件，请问如何获取node.shp和link.shp文件，以及怎么得到路段的csv文件信息，其中包含所有的路段id，类型，道路数等信息。

[gjq-emperor]

有一个区域.osm文件，请问如何获取node.shp和link.shp文件，以及怎么得到路段的csv文件信息，其中包含所有的路段id，类型，道路数等信息。

[zdsjjtTLG]

看看这两个包：osmnx和osm2gmns，整合OSM路网的功能还在排期，后面会整合到gotrackit里面

[zdsjjtTLG]

可以试试@李金洋 李总写的这个从osmnx(1.9.1版本)转化为gotrackit路网的脚本
osmnxtogotrackit.zip

[gjq-emperor]

好滴，感谢

[promise2341]

之前那个有问题，用这个就没问题了@zdsjjtTLG

import osmnx as ox
import geopandas as gpd
import pandas as pd
import numpy as np

get the road network for a place

def base_road(place , dh=4326):
"""
This function is used to get the base road network for a place,
and the road network is represented by the nodes and edges

:param place_name: the name of the place
:param dh: the distance hreshold

:return: edges and nodes
"""
if type(place) == str:
    # get the street network for a place
    G = ox.graph_from_place(place, network_type="drive")
elif type(place) == list:
    G = ox.graph_from_bbox(north=place[0],south=place[1],east=place[2],west=place[3] ,network_type='drive',retain_all=True)
else:
    print('报错:graph_type输入参数有误,按照行政区请输入0,按照经纬度范围请输入1。')
# Get the nodes and edges
nodes , edges = ox.graph_to_gdfs(G)
# set the crs to UTM 51N
edges = edges.to_crs(epsg=dh)
# reset the index
edges.reset_index(inplace=True)
# rename the columns
edges.rename(columns={'u':'from_node','v':'to_node','name':'road_name','osmid':'link_id'}, inplace=True)


# 判断 dir 属性
# 如果不是布尔类型，则进行转换
edges['oneway'] = edges['oneway'].astype(bool)
edges['reversed'] = edges['reversed'].astype(bool)

# 定义条件
conditions = [
    ~edges['oneway'],  # oneway == False
    # (edges['oneway']) & (~edges['reversed']),  # oneway == True & reversed == False
    # (edges['oneway']) & (edges['reversed'])    # oneway == True & reversed == True
    edges['oneway']   # oneway == True
]

# 定义对应的值（这一部分待验证）
choices = [
    0,   # 如果 oneway 为 False
    # -1,   # 如果 oneway 为 True 且 reversed 为 False
    1   # 如果 oneway 为 True 且 reversed 为 True
]

# 使用 np.select 创建新的 'dir' 列
edges['dir'] = np.select(conditions, choices, default=0)    



colist = ['from_node','to_node','link_id','dir','road_name','length','geometry']
edges = edges[colist]
# add the link id
edges['link_id']  = [x for x in range(0,len(edges))]
# add the road name type
edges['road_name'] = edges['road_name'].astype(str)


# reset the index
nodes.reset_index(inplace=True)
# rename the columns
nodes.rename(columns={'osmid':'node_id'},inplace=True)
# input the node type
nodes.to_crs(epsg=dh,inplace=True)
nodes = nodes[['node_id','geometry']]


# 重构节点映射
# 提取所有唯一节点id
unique_nodes = pd.unique(edges[['from_node', 'to_node']].values.ravel())
# 创建一个节点id和索引的映射
# 创建映射字典
node_mapping = {old_id: new_id for new_id, old_id in enumerate(unique_nodes, start=1)}
# 将节点id映射到新的id
edges['from_node'] = edges['from_node'].map(node_mapping)
edges['to_node'] = edges['to_node'].map(node_mapping)
# 如果需要更新 nodes 表
nodes['node_id'] = nodes['node_id'].map(node_mapping)

# 可视化
ox.plot_graph(G)

return nodes,edges

[zdsjjtTLG]

好的 谢谢您 🎉

[gjq-emperor]

请问，咱们路径匹配用的node和link文件除了shp文件，可以用csv文件吗，如果可以怎么使用

[zdsjjtTLG]

可以的，你pandas读取进来，将他转化为GeoDataFrame就可以了

[buaa-csc]

你好，请问我已经对输入的文件做了数据检查，但是运行时仍然报错，显示AssertionError: the value of lng must in [-180, 180]，这会是什么问题导致的？
Validating Road Line Layer:
✅ All required fields are present.
✅ link_id values are valid.
✅ from_node values are valid.
✅ to_node values are valid.
✅ 'dir' values are valid.
✅ 'length' values are valid.
✅ 'geometry' contains only 'LineString' types.
✅ 'geometry' contains only 2D coordinates.
Validation completed.

Checking Node Layer:
✅ All geometries are valid.
✅ All longitudes are within the range [-180, 180].
✅ All latitudes are within the range [-90, 90].
✅ No duplicated IDs.
/opt/anaconda3/envs/data-mining_454_homework/lib/python3.9/site-packages/gotrackit/tools/geo_process.py:296: RuntimeWarning: Mean of empty slice.
mean_x = np.array([geo.x for geo in node_gdf['geometry']]).mean()
/opt/anaconda3/envs/data-mining_454_homework/lib/python3.9/site-packages/numpy/_core/_methods.py:138: RuntimeWarning: invalid value encountered in scalar divide
ret = ret.dtype.type(ret / rcount)
Traceback (most recent call last):
File "/Users/csc/Documents/data-mining_454_homework/MatchTheMap.py", line 149, in
my_net = Net(link_gdf=link, node_gdf=node, cut_off=1200.0) # 可根据需要调整cut_off参数
File "/opt/anaconda3/envs/data-mining_454_homework/lib/python3.9/site-packages/gotrackit/WrapsFunc.py", line 13, in inner
res = f(*args, **kwargs)
File "/opt/anaconda3/envs/data-mining_454_homework/lib/python3.9/site-packages/gotrackit/map/Net.py", line 123, in init
self.__node = Node(node_gdf=node_gdf, is_check=is_check, init_available_node=self.cache_id,
File "/opt/anaconda3/envs/data-mining_454_homework/lib/python3.9/site-packages/gotrackit/map/Node.py", line 35, in init
self.planar_crs = judge_plain_crs_based_on_node(node_gdf=self.__node_gdf)
File "/opt/anaconda3/envs/data-mining_454_homework/lib/python3.9/site-packages/gotrackit/tools/geo_process.py", line 297, in judge_plain_crs_based_on_node
assert -180.0 <= mean_x <= 180.0, 'the value of lng must in [-180, 180]'
AssertionError: the value of lng must in [-180, 180]

[buaa-csc]

这个是路网匹配：

import pandas as pd
import geopandas as gpd
from gotrackit.map.Net import Net
from gotrackit.MapMatch import MapMatch
import os


def validate_road_line_layer(gdf):
    try:
        print("Validating Road Line Layer:")

        # 1. 字段检查
        required_fields = ['link_id', 'from_node', 'to_node', 'dir', 'length', 'geometry']
        missing_fields = [field for field in required_fields if field not in gdf.columns]
        if missing_fields:
            print(f"❌ Missing required fields: {missing_fields}")
            return
        else:
            print("✅ All required fields are present.")

        # 2. 验证字段值
        # (1) link_id, from_node, to_node 是否为正整数
        for field in ['link_id', 'from_node', 'to_node']:
            if not (gdf[field].apply(lambda x: isinstance(x, int) and x > 0).all()):
                print(f"❌ {field} contains invalid values (must be positive integers).")
            else:
                print(f"✅ {field} values are valid.")

        # (2) dir 是否为 0 或 1
        if not gdf['dir'].isin([0, 1]).all():
            print("❌ 'dir' contains invalid values (must be 0 or 1).")
        else:
            print("✅ 'dir' values are valid.")

        # (3) length 是否为正数
        if not (gdf['length'] > 0).all():
            print("❌ 'length' contains invalid values (must be positive).")
        else:
            print("✅ 'length' values are valid.")

        # 3. 几何检查
        # (1) 验证是否为 LineString 类型
        if not gdf.geometry.geom_type.isin(['LineString']).all():
            print("❌ 'geometry' contains invalid types (only 'LineString' is allowed).")
        else:
            print("✅ 'geometry' contains only 'LineString' types.")

        # (2) 检查是否包含三维坐标
        if gdf.geometry.has_z.any():
            print("❌ 'geometry' contains 3D coordinates, which are not supported.")
        else:
            print("✅ 'geometry' contains only 2D coordinates.")

        print("Validation completed.")

    except Exception as e:
        print(f"An error occurred during validation: {e}")


def check_shapefile(gdf, name):
    try:
        print(f"\nChecking {name}:")

        # 检查是否有几何列
        if 'geometry' not in gdf.columns:
            print("❌ No geometry column found.")
            return

        # 检查是否有空值
        if gdf['geometry'].isnull().any():
            print(f"❌ {gdf['geometry'].isnull().sum()} geometries are null.")
        else:
            print("✅ All geometries are valid.")

        # 检查经纬度范围
        gdf['longitude'] = gdf.geometry.x
        gdf['latitude'] = gdf.geometry.y

        invalid_lng = gdf[(gdf['longitude'] < -180) | (gdf['longitude'] > 180)]
        invalid_lat = gdf[(gdf['latitude'] < -90) | (gdf['latitude'] > 90)]

        if not invalid_lng.empty:
            print(f"❌ {len(invalid_lng)} rows with longitude out of range [-180, 180].")
        else:
            print("✅ All longitudes are within the range [-180, 180].")

        if not invalid_lat.empty:
            print(f"❌ {len(invalid_lat)} rows with latitude out of range [-90, 90].")
        else:
            print("✅ All latitudes are within the range [-90, 90].")

        # 检查重复值（假设 `id` 是关键字段）
        if 'id' in gdf.columns and gdf['id'].duplicated().any():
            print(f"❌ {gdf['id'].duplicated().sum()} duplicated IDs found.")
        else:
            print("✅ No duplicated IDs.")

    except Exception as e:
        print(f"An error occurred while checking {name}: {e}")
def check_longitude_range(shapefile_path):
    if not os.path.exists(shapefile_path):
        print(f"File not found: {shapefile_path}")
        return

    try:
        # 加载 Shapefile
        gdf = gpd.read_file(shapefile_path)

        # 检查是否包含几何列
        if 'geometry' not in gdf.columns:
            print("No geometry column found in the shapefile.")
            return

        # 提取经纬度
        gdf['longitude'] = gdf.geometry.x
        gdf['latitude'] = gdf.geometry.y

        # 检查经度是否超出范围
        invalid_lng = gdf[(gdf['longitude'] < -180) | (gdf['longitude'] > 180)]
        if not invalid_lng.empty:
            print(f"Found {len(invalid_lng)} rows with longitude out of range [-180, 180]:")
            print(invalid_lng[['longitude', 'latitude']])
        else:
            print("All longitudes are within the range [-180, 180].")

    except Exception as e:
        print(f"An error occurred: {e}")

# 1. 读取路网线层和点层
link = gpd.read_file(r'./cleaned_road_lines.shp', encoding='utf-8')  # 替换为实际的线层文件路径
node = gpd.read_file(r'./data/output/create_node/GeneratedNodes.shp', encoding='utf-8')  # 替换为实际的点层文件路径
# 路径和文件加载
link_path = './cleaned_road_lines.shp'
node_path = './data/output/create_node/GeneratedNodes.shp'

if not os.path.exists(link_path):
    print(f"File not found: {link_path}")
else:
    link_gdf = gpd.read_file(link_path)
    validate_road_line_layer(link_gdf)

if not os.path.exists(node_path):
    print(f"File not found: {node_path}")
else:
    node_gdf = gpd.read_file(node_path)
    check_shapefile(node_gdf, "Node Layer")

# 2. 构建 Net 对象并初始化
my_net = Net(link_gdf=link, node_gdf=node, cut_off=1200.0)  # 可根据需要调整cut_off参数
my_net.init_net()  # 初始化网络

# 3. 读取并处理 GPS 数据
gps_df = pd.read_csv(r'./after_reprocess_gps.csv')  # 替换为实际的GPS数据路径

# 4. 基于 Net 初始化 MapMatch 类，并进行参数配置
mpm = MapMatch(net=my_net,
               flag_name='example_sample',  # 可以自定义标记名称
               gps_buffer=100,  # 设置 GPS 缓冲区
               dense_gps=False,  # 根据需要调整
               use_sub_net=True,  # 是否使用子网
               use_heading_inf=True,  # 是否使用方向信息
               omitted_l=6.0,  # 忽略的最短路径长度
               del_dwell=True,  # 是否删除停留点
               dwell_l_length=50.0,  # 停留长度
               dwell_n=0,  # 停留点的个数
               export_html=True,  # 是否导出 HTML 可视化结果
               export_geo_res=True,  # 是否导出 Geo 数据
               use_gps_source=True,  # 是否使用原始 GPS 数据
               gps_radius=15.0,  # GPS 半径容差
               export_all_agents=False,  # 是否导出所有匹配的代理
               out_fldr=r'./data/output/match_visualization/example_sample')  # 输出路径

# 5. 执行路径匹配
match_res, warn_info, error_info = mpm.execute(gps_df=gps_df)

# 6. 导出匹配结果
match_res.to_csv(r'./data/output/match_res.csv', encoding='utf_8_sig', index=False)

# 7. 可选：输出警告和错误信息
if warn_info:
    print("Warnings:", warn_info)
if error_info:
    print("Errors:", error_info)

[zdsjjtTLG]

你似乎将平面投影坐标系指定为了地理坐标系EPSG:4326

[zdsjjtTLG]

地理坐标系下，0.7的Buffer大概是70Km

[buaa-csc]

谢谢解答，但我调整之后出现了新的问题，运行路径匹配后显示
Traceback (most recent call last):
File "/Users/csc/Documents/data-mining_454_homework/MatchTheMap.py", line 152, in
my_net = Net(link_gdf=link, node_gdf=node, cut_off=1200.0) # 可根据需要调整cut_off参数
File "/opt/anaconda3/envs/data-mining_454_homework/lib/python3.9/site-packages/gotrackit/WrapsFunc.py", line 13, in inner
res = f(*args, **kwargs)
File "/opt/anaconda3/envs/data-mining_454_homework/lib/python3.9/site-packages/gotrackit/map/Net.py", line 158, in init
self.check()
File "/opt/anaconda3/envs/data-mining_454_homework/lib/python3.9/site-packages/gotrackit/map/Net.py", line 191, in check
assert link_node_set.issubset(node_set), 'Link层中部分节点在Node层中没有记录'
AssertionError: Link层中部分节点在Node层中没有记录
这个该如何解决

[zdsjjtTLG]

为何不用创建点层那一步返回的new_link_gdf作为线层呢，你原来那个线层怎么可能会和新的点层有关联关系呢？

[gjq-emperor]

每个轨迹点之间的间隔为3秒，请问参数怎么设置合理？我按照这个代码参数没变匹配了一下，很多点都没匹配成功发出警告

[zdsjjtTLG]

讲清楚数据概况，而且可能是你的路网问题？

[zdsjjtTLG]

如果方便，直接发样例数据和路网

[zdsjjtTLG]

不过还是建议先：自行探索各参数含义，文档都写的很详细，多试一下，多思考，实在解决不了再来提问

[JunzeYang]

您好，想请教一下轨迹匹配模块是否能够区分轨迹在高架快速路还是地面辅路（包括上下匝道）？尤其是在城市路网中，这点挺影响结果的。谢谢！

[JunzeYang]

谢谢解答，您所说的拓扑关系区别具体指什么？我理解的如果轨迹通过地面平交道口驶入或驶离辅路是能够区分的（因为拓扑上与快速路不直接连通），但当轨迹有很长一段经过快速路/辅路时，依靠HMM要如何去判断这一段全程在快速路/辅路还是中间经过了匝道上/下了快速路呢？两者线形、方向太相似，而且距离很近，拓扑上都连通可行。

[zdsjjtTLG]

明白你的意思，这种情况需要考虑GPS的速度和路段限速，就是考虑时间限制，目前还不支持这个方式，后面会上这个，现在有提供一个节点限制的方法，就是限定某个GPS点一定会经过某个节点，这个新手册没有写的很详细，晚点我更新下手册

[zdsjjtTLG]

这种情况应该不多，所以之前没有优先更新这个速度限制的方法

[JunzeYang]

好的 感谢

[zdsjjtTLG]

https://gotrackit.readthedocs.io/en/latest/UserGuide/%E8%B7%AF%E5%BE%84%E5%8C%B9%E9%85%8D/#_13