OnLineMapMatch

实时路径(地图)匹配类OnLineMapMatch - 初始化：

• 轨迹数据实时匹配类, 提供了单核匹配方法

Parameters:

Name	Type	Description	Default
net	Net	路网Net对象	required
flag_name	str	标记字符名称, 会用于标记输出的可视化文件	'test'
use_sub_net	bool	是否在子网络上进行计算	False
dup_threshold	float	重复点检测阈值, 利用GPS轨迹计算sub_net时先对GPS点原始轨迹做简化	10.0
time_format	str	[1]时间列构建参数 - GPS数据中时间列的格式化字符串模板	'%Y-%m-%d %H:%M:%S'
time_unit	str	[1]时间列构建参数 - 时间单位, gotrackit会先尝试使用time_format进行时间列构建, 如果失败会再次尝试使用time_unit进行时间列构建	's'
gps_buffer	float	[2]候选规则参数 - GPS的搜索半径, 单位米, 意为只选取每个gps点附近gps_buffer米范围内的路段作为初步候选路段	200.0
top_k	int	[2]候选规则参数 - 选取每个GPS点gps_buffer范围内最近的top_k个路段	20
gps_route_buffer_gap	float	[2]候选规则参数 - 半径增量, gps_buffer + gps_route_buffer_gap 的半径范围用于计算子网络	15.0
beta	float	[3]概率参数 - 该值越大, 状态转移概率对于距离越不敏感	6.0
gps_sigma	float	[3]概率参数 - 该值越大, 发射概率对距离越不敏感	30.0
dis_para	float	[3]概率参数 - 距离的折减系数，不建议修改该参数	0.1
use_heading_inf	bool	[4]概率修正参数 - 是否利用GPS的差分方向向量修正发射概率, 适用于: 低定位误差 GPS数据 或者低频定位数据(配合加密参数)	False
heading_para_array	ndarray	[4]概率修正参数 - 差分方向修正参数数组	None
omitted_l	float	[4]概率修正参数 - 当某GPS点与前后GPS点的平均距离小于omitted_l(m)时, 该GPS点的方向限制作用被取消	6.0
del_dwell	bool	[5]停留点处理 - 是否进行停留点识别并且删除停留点	True
dwell_l_length	float	[5]停留点处理 - 停留点识别距离阈值	5.0
dwell_n	int	[5]停留点处理 - 超过连续dwell_n个相邻GPS点的距离小于dwell_l_length，那么这一组点就会被识别为停留点	2
is_lower_f	bool	[6]轨迹降频 - 是否对GPS数据进行数据降频率, 适用于: 高频-高定位误差 GPS数据	False
lower_n	int	[6]轨迹降频 - 频率倍率	2
is_rolling_average	bool	[7]滑动窗口平滑 - 是否启用滑动窗口平均对GPS数据进行降噪	False
window	int	[7]滑动窗口平滑 - 滑动窗口大小	2
dense_gps	bool	[8]轨迹增密 - 是否对GPS数据进行增密	True
dense_interval	float	[8]轨迹增密 - 当前后GPS点的直线距离l超过dense_interval即进行增密, 进行 int(l / dense_interval) + 1 等分增密	100.0
speed_threshold	float	[9]车辆行驶速度阈值, km/h, 用于判定在cut_off之外的路径是否满足速度合理性(速度超过阈值认为不合理)	200
use_st	bool	[10]st-match参数 - 是否启用行程速度限制	False
st_main_coe	float	[10]st-match参数 - 主系数	1.0
st_min_factor	float	[10]st-match参数 - 最小折减系数	0.1
out_fldr	str	[11]输出设置 - 保存匹配结果的文件目录	None
instant_output	bool	[11]输出设置 - 是否每匹配完一条轨迹就存储csv匹配结果	False
user_field_list	list[str]	[11]输出设置 - gps数据中, 用户想要附带在匹配结果表中输出的额外字段列表	None
export_html	bool	[12]HTML输出设置 - 是否输出匹配结果的网页可视化html文件	False
use_gps_source	bool	[12]HTML输出设置 - 是否在可视化结果中使用GPS源数据进行展示	False
gps_radius	float	[12]HTML输出设置 - HTML可视化中GPS点的半径大小，单位米	6.0
export_all_agents	bool	[12]HTML输出设置 - 是否将所有agent的可视化存储于一个html文件中	False
visualization_cache_times	int	[12]HTML输出设置 - 每匹配完visualization_cache_times辆车再进行(html or geojson文件)结果的统一存储。该参数设置的过大可能导致内存溢出	5
export_geo_res	bool	[13]GeoJSON输出设置 - 是否输出匹配结果的几何可视化文件	False
heading_vec_len	float	[13]GeoJSON输出设置 - 匹配航向向量的长度(控制geojson文件中的可视化)	15.0

Source code in src/gotrackit/MapMatch.py

def __init__(self, net: Net, flag_name: str = 'test', use_sub_net: bool = False,
             time_format: str = "%Y-%m-%d %H:%M:%S", time_unit: str = 's',
             gps_buffer: float = 200.0, gps_route_buffer_gap: float = 15.0,
             beta: float = 6.0, gps_sigma: float = 30.0, dis_para: float = 0.1,
             is_lower_f: bool = False, lower_n: int = 2,
             use_heading_inf: bool = False, heading_para_array: np.ndarray = None,
             dense_gps: bool = True, dense_interval: float = 100.0,
             dwell_l_length: float = 5.0, dwell_n: int = 2, del_dwell: bool = True,
             dup_threshold: float = 10.0,
             is_rolling_average: bool = False, window: int = 2,
             speed_threshold: float = 200,
             use_st: bool = False, st_main_coe: float = 1.0, st_min_factor: float = 0.1,
             export_html: bool = False, use_gps_source: bool = False, out_fldr: str = None,
             export_geo_res: bool = False, top_k: int = 20, omitted_l: float = 6.0,
             link_width: float = 1.5, node_radius: float = 1.5,
             match_link_width: float = 5.0, gps_radius: float = 6.0, export_all_agents: bool = False,
             visualization_cache_times: int = 5, multi_core_save: bool = False, instant_output: bool = False,
             user_field_list: list[str] = None, heading_vec_len: float = 15.0):
    """实时路径(地图)匹配类OnLineMapMatch - 初始化：

    - 轨迹数据实时匹配类, 提供了单核匹配方法

    Args:
        net: 路网Net对象
        flag_name: 标记字符名称, 会用于标记输出的可视化文件
        use_sub_net: 是否在子网络上进行计算
        dup_threshold: 重复点检测阈值, 利用GPS轨迹计算sub_net时先对GPS点原始轨迹做简化
        time_format: [1]时间列构建参数 - GPS数据中时间列的格式化字符串模板
        time_unit: [1]时间列构建参数 - 时间单位, gotrackit会先尝试使用time_format进行时间列构建, 如果失败会再次尝试使用time_unit进行时间列构建
        gps_buffer: [2]候选规则参数 - GPS的搜索半径, 单位米, 意为只选取每个gps点附近gps_buffer米范围内的路段作为初步候选路段
        top_k: [2]候选规则参数 - 选取每个GPS点gps_buffer范围内最近的top_k个路段
        gps_route_buffer_gap: [2]候选规则参数 - 半径增量, gps_buffer + gps_route_buffer_gap 的半径范围用于计算子网络
        beta: [3]概率参数 - 该值越大, 状态转移概率对于距离越不敏感
        gps_sigma: [3]概率参数 - 该值越大, 发射概率对距离越不敏感
        dis_para: [3]概率参数 - 距离的折减系数，不建议修改该参数
        use_heading_inf: [4]概率修正参数 - 是否利用GPS的差分方向向量修正发射概率, 适用于: 低定位误差 GPS数据 或者低频定位数据(配合加密参数)
        heading_para_array: [4]概率修正参数 - 差分方向修正参数数组
        omitted_l: [4]概率修正参数 - 当某GPS点与前后GPS点的平均距离小于omitted_l(m)时, 该GPS点的方向限制作用被取消
        del_dwell: [5]停留点处理 - 是否进行停留点识别并且删除停留点
        dwell_l_length: [5]停留点处理 - 停留点识别距离阈值
        dwell_n: [5]停留点处理 - 超过连续dwell_n个相邻GPS点的距离小于dwell_l_length，那么这一组点就会被识别为停留点
        is_lower_f: [6]轨迹降频 - 是否对GPS数据进行数据降频率, 适用于: 高频-高定位误差 GPS数据
        lower_n: [6]轨迹降频 - 频率倍率
        is_rolling_average: [7]滑动窗口平滑 - 是否启用滑动窗口平均对GPS数据进行降噪
        window: [7]滑动窗口平滑 - 滑动窗口大小
        dense_gps: [8]轨迹增密 - 是否对GPS数据进行增密
        dense_interval: [8]轨迹增密 - 当前后GPS点的直线距离l超过dense_interval即进行增密, 进行 int(l / dense_interval) + 1 等分增密
        speed_threshold: [9]车辆行驶速度阈值, km/h, 用于判定在cut_off之外的路径是否满足速度合理性(速度超过阈值认为不合理)
        use_st: [10]st-match参数 - 是否启用行程速度限制
        st_main_coe: [10]st-match参数 - 主系数
        st_min_factor: [10]st-match参数 - 最小折减系数
        out_fldr: [11]输出设置 - 保存匹配结果的文件目录
        instant_output: [11]输出设置 - 是否每匹配完一条轨迹就存储csv匹配结果
        user_field_list: [11]输出设置 - gps数据中, 用户想要附带在匹配结果表中输出的额外字段列表
        export_html: [12]HTML输出设置 - 是否输出匹配结果的网页可视化html文件
        use_gps_source: [12]HTML输出设置 - 是否在可视化结果中使用GPS源数据进行展示
        gps_radius: [12]HTML输出设置 - HTML可视化中GPS点的半径大小，单位米
        export_all_agents: [12]HTML输出设置 - 是否将所有agent的可视化存储于一个html文件中
        visualization_cache_times: [12]HTML输出设置 - 每匹配完visualization_cache_times辆车再进行(html or geojson文件)结果的统一存储。该参数设置的过大可能导致内存溢出
        export_geo_res: [13]GeoJSON输出设置 - 是否输出匹配结果的几何可视化文件
        heading_vec_len: [13]GeoJSON输出设置 - 匹配航向向量的长度(控制geojson文件中的可视化)
    """
    MapMatch.__init__(self, flag_name=flag_name, net=net, use_sub_net=use_sub_net, time_format=time_format,
                      time_unit=time_unit, gps_buffer=gps_buffer, gps_route_buffer_gap=gps_route_buffer_gap,
                      beta=beta, gps_sigma=gps_sigma, dis_para=dis_para, is_lower_f=is_lower_f,
                      lower_n=lower_n, use_heading_inf=use_heading_inf, heading_para_array=heading_para_array,
                      dense_gps=dense_gps, dense_interval=dense_interval,
                      dwell_l_length=dwell_l_length, dwell_n=dwell_n, del_dwell=del_dwell,
                      dup_threshold=dup_threshold, is_rolling_average=is_rolling_average, window=window,
                      speed_threshold=speed_threshold,
                      export_html=export_html, use_gps_source=use_gps_source, out_fldr=out_fldr,
                      export_geo_res=export_geo_res, top_k=top_k, omitted_l=omitted_l, gps_radius=gps_radius,
                      link_width=link_width, node_radius=node_radius, match_link_width=match_link_width,
                      export_all_agents=export_all_agents, visualization_cache_times=visualization_cache_times,
                      multi_core_save=multi_core_save, instant_output=instant_output,
                      user_field_list=user_field_list, heading_vec_len=heading_vec_len,
                      use_st=use_st, st_main_coe=st_main_coe, st_min_factor=st_min_factor)
    self.his_hmm_dict = dict()
    self.his_gps = dict()
    self.add_single_ft = [True]

OnLineMapMatch类方法 - execute：

• 对输入的gps数据执行实时路径匹配并且输出匹配结果

Parameters:

Name	Type	Description	Default
gps_df	DataFrame | GeoDataFrame	gps数据表对象, 必需参数	required
gps_already_plain	bool	传入的GPS数据是否是平面投影坐标系	False
time_gap_threshold	float	时间阈值，如果某agent的当前批GPS数据的最早定位时间，和上批次GPS数据的最晚定位时间的差值超过该值，则不参考历史概率链进行匹配计算	1800.0
dis_gap_threshold	float	距离阈值，如果某agent的当前批GPS数据的最早定位点，和上批次GPS数据的最晚定位点的距离超过该值，则不参考历史概率链进行匹配计算	600.0
overlapping_window	int	重叠窗口长度，和历史GPS数据的重叠窗口	3

Returns:

Type	Description
tuple[DataFrame, dict, list]	匹配结果表(pd.DataFrame), 警告信息(dict), 错误信息(list)

Source code in src/gotrackit/MapMatch.py

def execute(self, gps_df: pd.DataFrame | gpd.GeoDataFrame,
            gps_already_plain: bool = False, time_gap_threshold: float = 1800.0,
            dis_gap_threshold: float = 600.0,
            overlapping_window: int = 3) -> tuple[pd.DataFrame, dict, list]:
    """OnLineMapMatch类方法 - execute：

    - 对输入的gps数据执行实时路径匹配并且输出匹配结果

    Args:
        gps_df: gps数据表对象, 必需参数
        gps_already_plain: 传入的GPS数据是否是平面投影坐标系
        time_gap_threshold: 时间阈值，如果某agent的当前批GPS数据的最早定位时间，和上批次GPS数据的最晚定位时间的差值超过该值，则不参考历史概率链进行匹配计算
        dis_gap_threshold: 距离阈值，如果某agent的当前批GPS数据的最早定位点，和上批次GPS数据的最晚定位点的距离超过该值，则不参考历史概率链进行匹配计算
        overlapping_window: 重叠窗口长度，和历史GPS数据的重叠窗口

    Returns:
        匹配结果表(pd.DataFrame), 警告信息(dict), 错误信息(list)
    """
    # check and format
    self.is_rolling_average = False
    # self.del_dwell = False
    user_field_list = GpsPointsGdf.check(gps_points_df=gps_df, user_field_list=self.user_field_list)
    may_error_list = dict()
    error_list = list()
    all_agent_match_res = list()
    hmm_res_list = []  # save hmm_res
    gps_df.dropna(subset=[agent_id_field], inplace=True)
    agent_num = len(gps_df[gps_field.AGENT_ID_FIELD].unique())
    if agent_num == 0:
        print('after removing the rows with empty values in the agent_id column, the gps data is empty...')
        return pd.DataFrame(), may_error_list, error_list

    # 对每辆车的轨迹进行匹配
    agent_count = 0
    # self.add_single_ft = [True]

    for agent_id, _gps_df in gps_df.groupby(gps_field.AGENT_ID_FIELD):
        cor_with_his = False
        agent_count += 1
        if agent_id in self.his_hmm_dict.keys():
            self.gps_buffer = self.his_hmm_dict[agent_id].gps_points.buffer
        else:
            if len(_gps_df) <= 1 and agent_id not in self.his_gps.keys():
                self.his_gps[agent_id] = _gps_df.copy()
                continue
            else:
                if agent_id in self.his_gps.keys():
                    _gps_df = pd.concat([self.his_gps[agent_id], _gps_df])
                    del self.his_gps[agent_id]
                _gps_df.reset_index(inplace=True, drop=True)

        print(rf'- gotrackit ------> No.{agent_count}: agent: {agent_id} ')
        try:
            gps_obj = GpsPointsGdf(gps_points_df=_gps_df, time_format=self.time_format,
                                   buffer=self.gps_buffer, time_unit=self.time_unit,
                                   plane_crs=self.plain_crs,
                                   user_filed_list=user_field_list, already_plain=gps_already_plain)
        except Exception as e:
            print(f'error constructing GPS object: {repr(e)}')
            error_list.append(agent_id)
            continue
        del _gps_df
        # judge if the same chain with his agent
        last_seq_list = list()
        if agent_id in self.his_hmm_dict.keys():
            cor_with_his, last_seq_list = gps_obj.merge_gps(
                gps_obj=self.his_hmm_dict[agent_id].gps_points,
                depth=overlapping_window,
                dis_gap_threshold=dis_gap_threshold,
                time_gap_threshold=time_gap_threshold)

        # gps-process
        try:
            self.gps_reprocess(gps_obj=gps_obj,
                               is_lower_f=self.is_lower_f, lower_n=self.lower_n,
                               is_rolling_average=self.is_rolling_average, rolling_window=self.rolling_window,
                               dense_gps=self.dense_gps, dense_interval=self.dense_interval,
                               del_dwell=self.del_dwell, dwell_l_length=self.dwell_l_length, dwell_n=self.dwell_n)
        except Exception as e:
            print(rf'GPS data preprocessing error: {repr(e)}')
            error_list.append(agent_id)
            continue

        if len(gps_obj.gps_gdf) <= 1:
            print(r'after data preprocessing, there are less than 2 GPS observation points.')
            error_list.append(agent_id)
            continue

        # 依据当前的GPS数据(源数据)做一个子网络
        if self.use_sub_net:
            print(rf'using sub net')
            self.add_single_ft[0] = True
            used_net = self.my_net.create_computational_net(
                gps_array_buffer=gps_obj.get_gps_array_buffer(buffer=self.sub_net_buffer,
                                                              dup_threshold=self.dup_threshold),
                fmm_cache=self.my_net.fmm_cache, prj_cache=self.my_net.prj_cache,
                weight_field=self.my_net.weight_field,
                cache_path=self.my_net.cache_path, cache_id=self.my_net.cache_id,
                not_conn_cost=self.my_net.not_conn_cost)
            if used_net is None:
                error_list.append(agent_id)
                continue
        else:
            used_net = self.my_net
            print(rf'using whole net')

        hmm_obj = HiddenMarkov(net=used_net, gps_points=gps_obj, beta=self.beta,
                               gps_sigma=self.gps_sigma,
                               not_conn_cost=self.not_conn_cost,
                               use_heading_inf=self.use_heading_inf,
                               heading_para_array=self.heading_para_array,
                               dis_para=self.dis_para,
                               top_k=self.top_k, omitted_l=self.omitted_l,
                               para_grid=self.para_grid, speed_threshold=self.speed_threshold,
                               heading_vec_len=self.heading_vec_len, flag_name=self.flag_name,
                               out_fldr=self.out_fldr, use_st=self.use_st, st_main_coe=self.st_main_coe,
                               st_min_factor=self.st_min_factor)
        his_emission = dict()
        his_ft_idx_map = pd.DataFrame()
        if cor_with_his:
            his_emission = self.his_hmm_dict[agent_id].extract_emission(seq_list=last_seq_list)
            his_ft_idx_map = self.his_hmm_dict[agent_id].get_ft_idx_map

        is_success, _match_res_df = \
            hmm_obj.hmm_rt_execute(add_single_ft=self.add_single_ft, last_em_para=his_emission,
                                   last_seq_list=last_seq_list, his_ft_idx_map=his_ft_idx_map)
        if not is_success:
            error_list.append(agent_id)
            continue
        hmm_obj.del_ft_trans(del_prj_df=False)

        self.his_hmm_dict[agent_id] = hmm_obj

        if hmm_obj.is_warn:
            may_error_list[agent_id] = hmm_obj.format_warn_info

        if self.instant_output:
            _match_res_df.to_csv(os.path.join(self.out_fldr, rf'{agent_id}_match_res.csv'),
                                 encoding='utf_8_sig', index=False)
        else:
            all_agent_match_res.append(_match_res_df)

        # if export files
        if self.export_html or self.export_geo_res:
            hmm_res_list.append(hmm_obj)
            if len(hmm_res_list) >= self.visualization_cache_times or agent_count == agent_num:
                export_visualization(hmm_obj_list=hmm_res_list, use_gps_source=self.use_gps_source,
                                     export_geo=self.export_geo_res, export_html=self.export_html,
                                     gps_radius=self.gps_radius, export_all_agents=self.export_all_agents,
                                     out_fldr=self.out_fldr, flag_name=self.flag_name,
                                     multi_core_save=self.multi_core_save, sub_net_buffer=self.sub_net_buffer,
                                     dup_threshold=self.dup_threshold)
                del hmm_res_list
                hmm_res_list = []

    if hmm_res_list:
        export_visualization(hmm_obj_list=hmm_res_list, use_gps_source=self.use_gps_source,
                             export_geo=self.export_geo_res, export_html=self.export_html,
                             gps_radius=self.gps_radius, export_all_agents=self.export_all_agents,
                             out_fldr=self.out_fldr, flag_name=self.flag_name,
                             multi_core_save=self.multi_core_save, sub_net_buffer=self.sub_net_buffer,
                             dup_threshold=self.dup_threshold)
    match_res_df = pd.concat(all_agent_match_res, ignore_index=True) if all_agent_match_res else pd.DataFrame()
    return match_res_df, may_error_list, error_list

Comments