MatchResAna

匹配结果处理类-MatchResAna

• 初始化

Source code in src/gotrackit/MatchResAna.py

def __init__(self):
    """匹配结果处理类-MatchResAna

    - 初始化
    """
    pass

MatchResAna类静态方法 - del_dup_links

• 对匹配结果中的局部重复路段进行剔除

Parameters:

Name	Type	Description	Default
match_res_df	DataFrame | GeoDataFrame	匹配结果表	required
use_time	bool	是否启用时间字段	True
time_format	str	时间列字符串模板, use_time=True时起效	'%Y-%m-%d %H:%M:%S'
time_unit	str	时间列单位, use_time=True时起效	's'
keep	str	去重类型, first: 保留最早时刻进入某link的记录, last: 保留最晚时刻进入某link的记录	'first'

Returns:

Type	Description
DataFrame | GeoDataFrame	局部去重后的匹配结果表

Source code in src/gotrackit/MatchResAna.py

@staticmethod
def del_dup_links(match_res_df: pd.DataFrame | gpd.GeoDataFrame,
                  time_format: str = '%Y-%m-%d %H:%M:%S', time_unit: str = 's', keep: str = 'first',
                  use_time: bool = True) -> pd.DataFrame | gpd.GeoDataFrame:
    """MatchResAna类静态方法 - del_dup_links

    - 对匹配结果中的局部重复路段进行剔除

    Args:
        match_res_df: 匹配结果表
        use_time: 是否启用时间字段
        time_format: 时间列字符串模板, use_time=True时起效
        time_unit: 时间列单位, use_time=True时起效
        keep: 去重类型, first: 保留最早时刻进入某link的记录, last: 保留最晚时刻进入某link的记录

    Returns:
        局部去重后的匹配结果表
    """
    return del_dup_links(match_res_df=match_res_df, time_format=time_format, time_unit=time_unit, keep=keep,
                         use_time=use_time)

MatchResAna类静态方法 - dense_res_based_on_net：

• 对MapMatch后输出的匹配结果进行路径增密

Parameters:

Name	Type	Description	Default
net	Net	Net路网对象	required
match_res_df	DataFrame	匹配结果表	required
lng_field	str	经度字段名称	'prj_lng'
lat_field	str	纬度字段名称	'prj_lat'
dis_threshold	float	合并距离阈值(米)	3
time_format	str	匹配结果表时间列字符串格式化模板	'%Y-%m-%d %H:%M:%S'
time_unit	str	匹配结果表时间列单位	's'
plain_crs	str	平面投影坐标系	'EPSG:3857'

Returns:

Type	Description
DataFrame	增密后的匹配结果表

Source code in src/gotrackit/MatchResAna.py

@staticmethod
def dense_res_based_on_net(net: Net, match_res_df: pd.DataFrame, lng_field: str = 'prj_lng',
                           lat_field: str = 'prj_lat', dis_threshold: float = 3,
                           time_format: str = '%Y-%m-%d %H:%M:%S',
                           time_unit: str = 's', plain_crs: str = 'EPSG:3857') -> pd.DataFrame:
    """MatchResAna类静态方法 - dense_res_based_on_net：

    - 对MapMatch后输出的匹配结果进行路径增密

    Args:
        net: Net路网对象
        match_res_df: 匹配结果表
        lng_field: 经度字段名称
        lat_field: 纬度字段名称
        dis_threshold: 合并距离阈值(米)
        time_format: 匹配结果表时间列字符串格式化模板
        time_unit: 匹配结果表时间列单位
        plain_crs: 平面投影坐标系

    Returns:
        增密后的匹配结果表
    """
    return dense_res_based_on_net(net=net, match_res_df=match_res_df, lat_field=lat_field, lng_field=lng_field,
                                  dis_threshold=dis_threshold, time_unit=time_unit, time_format=time_format,
                                  plain_crs=plain_crs)

generate_check_file函数：

• 依据匹配警告信息生成空间书签文件

Parameters:

Name	Type	Description	Default
net	Net	Net路网对象	required
warn_info_dict	dict	MapMatch匹配警告信息	None
out_fldr	str	空间书签的存储目录	'./'
file_name	str	空间书签文件名称	'check'

Returns:

Type	Description
	None

Source code in src/gotrackit/MatchResAna.py

def generate_check_file(net: Net, warn_info_dict: dict = None, out_fldr: str = r'./', file_name: str = 'check'):
    """generate_check_file函数：

    - 依据匹配警告信息生成空间书签文件

    Args:
        net: Net路网对象
        warn_info_dict: MapMatch匹配警告信息
        out_fldr: 空间书签的存储目录
        file_name: 空间书签文件名称

    Returns:
        None
    """
    single_link_gdf = net.get_link_data()
    # single_link_gdf.reset_index(inplace=True, drop=True)
    may_error_list, book_mark_dict = list(), dict()
    for agent_id in warn_info_dict.keys():
        _may_error_gdf = format_warn_info_to_geo(warn_info=warn_info_dict[agent_id], single_link_gdf=single_link_gdf)
        _may_error_gdf['ft_gps'] = str(agent_id) + '-' + _may_error_gdf['ft_gps']
        may_error_list.append(_may_error_gdf)

    if may_error_list:
        may_error_gdf = pd.concat(may_error_list).to_crs('EPSG:4326')
        book_mark_df = may_error_gdf.groupby('ft_gps').apply(
            lambda df: df[['geometry']].values[0, 0].centroid).reset_index(drop=False)
        book_mark_dict.update({k: (v.x, v.y) for k, v in zip(book_mark_df['ft_gps'], book_mark_df[0])})
        generate_book_mark(name_loc_dict=book_mark_dict, input_fldr=out_fldr, prj_name=file_name, _mode='replace')
        save_file(data_item=may_error_gdf, file_name=file_name, out_fldr=out_fldr, file_type='shp')

Comments