Trajectory

轨迹类TrajectoryPoints：

• 初始化

Parameters:

Name	Type	Description	Default
gps_points_df	DataFrame	定位数据表	required
time_format	str	时间列字符串格式模板	'%Y-%m-%d %H:%M:%S'
time_unit	str	时间单位	's'
plain_crs	str	平面投影坐标系	'EPSG:3857'

Source code in src/gotrackit/gps/Trajectory.py

def __init__(self, gps_points_df: pd.DataFrame, time_format: str = '%Y-%m-%d %H:%M:%S', time_unit: str = 's',
             plain_crs: str = 'EPSG:3857', already_plain: bool = False):
    """轨迹类TrajectoryPoints：

    - 初始化

    Args:
        gps_points_df: 定位数据表
        time_format: 时间列字符串格式模板
        time_unit: 时间单位
        plain_crs: 平面投影坐标系

    """
    self.time_format = time_format
    self.time_unit = time_unit
    user_field_list = list(set(gps_points_df.columns) - {agent_field, lng_field, lat_field, time_field,
                                                         gps_field.POINT_SEQ_FIELD})
    assert '4326' not in plain_crs, 'incorrectly specifying plain_crs as a geographic coordinate system'
    user_field_list = self.check(gps_points_df=gps_points_df, user_field_list=user_field_list)
    GpsPointsGdf.__init__(self, gps_points_df=gps_points_df, time_format=time_format, time_unit=time_unit,
                          plane_crs=plain_crs, already_plain=already_plain, multi_agents=True,
                          user_filed_list=user_field_list)

类方法 - dense

• 增密：对定位点进行线性增密

Parameters:

Name	Type	Description	Default
dense_interval	float	增密阈值(米), 当相邻GPS点的球面距离L超过dense_interval即进行增密, 进行 int(L / dense_interval) + 1 等分加密	120.0

Returns:

Type	Description
	self

Source code in src/gotrackit/gps/LocGps.py

def dense(self, dense_interval: float = 120.0):
    """类方法 - dense

    - 增密：对定位点进行线性增密

    Args:
        dense_interval: 增密阈值(米), 当相邻GPS点的球面距离L超过dense_interval即进行增密, 进行 int(L / dense_interval) + 1 等分加密

    Returns:
        self
    """
    if len(self.__gps_points_gdf) <= 1:
        return self
    # 时间差和距离差
    self.calc_adj_dis_gap()
    self.calc_adj_time_gap()
    not_same_agent_idx = self.not_same_agent_idx()
    choose_idx = (self.__gps_points_gdf[dis_gap_field] > dense_interval) & ~not_same_agent_idx
    self.__gps_points_gdf['all_idx'] = [i for i in range(len(self.__gps_points_gdf))]
    self.__gps_points_gdf[n_seg_field] = 0
    should_be_dense_gdf = self.__gps_points_gdf[choose_idx].copy()
    self.__gps_points_gdf.drop(columns=[next_time_field, next_p_field, time_gap_field, dis_gap_field],
                               inplace=True)
    if should_be_dense_gdf.empty:
        return self

    should_be_dense_gdf[n_seg_field] = (0.001 + should_be_dense_gdf[dis_gap_field] / dense_interval).astype(
        int) + 1
    should_be_dense_gdf[n_seg_field] = should_be_dense_gdf[n_seg_field].apply(lambda n: [x for x in range(1, n)])
    del should_be_dense_gdf[geometry_field]
    should_be_dense_gdf = should_be_dense_gdf[[agent_field, n_seg_field, 'all_idx']].copy()
    should_be_dense_gdf = should_be_dense_gdf.explode(column=n_seg_field, ignore_index=True)

    self.__gps_points_gdf = pd.concat([self.__gps_points_gdf, should_be_dense_gdf])
    self.__gps_points_gdf.sort_values(by=[agent_field, 'all_idx', n_seg_field], inplace=True)
    self.__gps_points_gdf.reset_index(inplace=True, drop=True)
    self.__gps_points_gdf[gps_field.POINT_SEQ_FIELD] = \
        self.__gps_points_gdf[gps_field.POINT_SEQ_FIELD].fillna(-1).astype(np.int64)

    self.__gps_points_gdf[[time_field, 'prj_x', 'prj_y']] = self.__gps_points_gdf[
        [time_field, 'prj_x', 'prj_y']].interpolate(method='linear')
    del self.__gps_points_gdf['all_idx'], self.__gps_points_gdf[n_seg_field]

    # must be plain
    dense_idx = self.__gps_points_gdf[gps_field.POINT_SEQ_FIELD] == -1
    self.__gps_points_gdf.loc[dense_idx, geometry_field] = gpd.points_from_xy(
        self.__gps_points_gdf.loc[dense_idx, 'prj_x'],
        self.__gps_points_gdf.loc[dense_idx, 'prj_y'], crs=self.plane_crs)

    self.__gps_points_gdf[ori_seq_field] = self.__gps_points_gdf[gps_field.POINT_SEQ_FIELD]

    # self.__gps_points_gdf = pd.concat([self.__gps_points_gdf, should_be_dense_gdf])
    # self.__gps_points_gdf.sort_values(by=[agent_field, time_field], ascending=[True, True], inplace=True)
    # self.__gps_points_gdf.reset_index(inplace=True, drop=True)
    self.add_seq_field(gps_points_gdf=self.__gps_points_gdf, multi_agents=self.multi_agents)
    #
    if not self.__user_gps_info.empty:
        self.__user_gps_info[ori_seq_field] = self.__user_gps_info[gps_field.POINT_SEQ_FIELD]
        del self.__user_gps_info[gps_field.POINT_SEQ_FIELD]
        self.__user_gps_info = pd.merge(self.__user_gps_info, self.__gps_points_gdf[
            [ori_seq_field, gps_field.POINT_SEQ_FIELD, agent_field]], on=[ori_seq_field, agent_field],
                                        how='left')
        self.__user_gps_info[gps_field.POINT_SEQ_FIELD] = self.__user_gps_info[gps_field.POINT_SEQ_FIELD].fillna(
            -1).astype(np.int64)
        del self.__user_gps_info[ori_seq_field]
    del self.__gps_points_gdf[ori_seq_field]
    return self

类方法 - lower_frequency

• 降频：对定位数据进行降频采样

Parameters:

Name	Type	Description	Default
lower_n	int	降频倍数	2
multi_agents	bool		True

Returns:

Type	Description
	self

Source code in src/gotrackit/gps/LocGps.py

def lower_frequency(self, lower_n: int = 2, multi_agents: bool = True):
    """类方法 - lower_frequency

    - 降频：对定位数据进行降频采样

    Args:
        lower_n: 降频倍数
        multi_agents:

    Returns:
        self
    """
    if multi_agents:
        self.__gps_points_gdf['label'] = self.__gps_points_gdf.groupby(agent_field).cumcount() % lower_n
        self.__gps_points_gdf = self.__gps_points_gdf[self.__gps_points_gdf['label'].eq(0)].copy()
    else:
        self.__gps_points_gdf['label'] = pd.Series([i for i in range(len(self.__gps_points_gdf))]) % lower_n
        self.__gps_points_gdf = self.__gps_points_gdf[self.__gps_points_gdf['label'].eq(0)].copy()
    del self.__gps_points_gdf['label']
    return self

类方法 - kf_smooth：

• 使用卡尔曼滤波对轨迹数据进行平滑

Parameters:

Name	Type	Description	Default
p_deviation	list or float	过程噪声的标准差	0.01
o_deviation	list or float	观测噪声的标准差, 该值越小, 平滑后的结果就越接近原轨迹	0.1

Returns:

Type	Description
	self

Source code in src/gotrackit/gps/LocGps.py

def kf_smooth(self, p_deviation: list or float = 0.01, o_deviation: list or float = 0.1):
    """类方法 - kf_smooth：

    - 使用卡尔曼滤波对轨迹数据进行平滑

    Args:
        p_deviation: 过程噪声的标准差
        o_deviation: 观测噪声的标准差, 该值越小, 平滑后的结果就越接近原轨迹

    Returns:
        self
    """
    tks = OffLineTrajectoryKF(trajectory_df=self.__gps_points_gdf,
                              x_field=gps_field.PLAIN_X, y_field=gps_field.PLAIN_Y)
    self.__gps_points_gdf = tks.execute(p_deviation=p_deviation, o_deviation=o_deviation)
    # self.__gps_points_gdf[geometry_field] = self.__gps_points_gdf[[gps_field.PLAIN_X, gps_field.PLAIN_Y]].apply(
    #     lambda x: Point(x), axis=1)
    # print(self.__gps_points_gdf.crs)
    self.__gps_points_gdf[geometry_field] = gpd.points_from_xy(self.__gps_points_gdf[gps_field.PLAIN_X],
                                                               self.__gps_points_gdf[gps_field.PLAIN_Y],
                                                               crs=self.__gps_points_gdf.crs)
    # print(self.__gps_points_gdf.crs)
    return self

类方法 - rolling_average：

• 滑动窗口：使用滑动窗口对定位数据进行平滑, 只要启用了该操作, 那么user_field_list自动失效

Parameters:

Name	Type	Description	Default
rolling_window	int	滑动窗口大小	2
multi_agents	bool		True

Returns:

Type	Description
	self

Source code in src/gotrackit/gps/LocGps.py

def rolling_average(self, multi_agents: bool = True, rolling_window: int = 2):
    """类方法 - rolling_average：

    - 滑动窗口：使用滑动窗口对定位数据进行平滑, 只要启用了该操作, 那么user_field_list自动失效

    Args:
        rolling_window: 滑动窗口大小
        multi_agents:

    Returns:
        self
    """
    # 滑动窗口执行后会重置所有的gps的seq字段
    if len(self.__gps_points_gdf) <= rolling_window:
        return self

    self.__gps_points_gdf[gps_field.TIME_FIELD] = self.__gps_points_gdf[gps_field.TIME_FIELD].apply(
        lambda t: t.timestamp())
    self.__gps_points_gdf.reset_index(inplace=True, drop=True)

    self.__gps_points_gdf = \
        self.__gps_points_gdf.groupby(agent_field)[
            [gps_field.PLAIN_X, gps_field.PLAIN_Y,
             gps_field.TIME_FIELD]].rolling(window=rolling_window, min_periods=1).mean().reset_index(drop=False)
    try:
        del self.__gps_points_gdf['level_1']
    except Exception as e:
        print(repr(e))

    self.__gps_points_gdf.dropna(subset=[gps_field.PLAIN_X, gps_field.PLAIN_Y,
                                         gps_field.TIME_FIELD], inplace=True, how='any')
    self.__gps_points_gdf.reset_index(inplace=True, drop=True)

    self.add_seq_field(gps_points_gdf=self.__gps_points_gdf, multi_agents=multi_agents)

    self.__gps_points_gdf[net_field.GEOMETRY_FIELD] = gpd.points_from_xy(self.__gps_points_gdf[gps_field.PLAIN_X],
                                                                         self.__gps_points_gdf[gps_field.PLAIN_Y],
                                                                         crs=self.crs)
    self.__gps_points_gdf[gps_field.TIME_FIELD] = \
        pd.to_datetime(self.__gps_points_gdf[gps_field.TIME_FIELD], unit='s')
    self.__gps_points_gdf = gpd.GeoDataFrame(self.__gps_points_gdf, geometry=gps_field.GEOMETRY_FIELD, crs=self.crs)
    self.__user_gps_info = self.__gps_points_gdf[[gps_field.AGENT_ID_FIELD, gps_field.POINT_SEQ_FIELD]].copy()
    self.__user_gps_info[gps_field.LOC_TYPE] = 's'
    self.user_filed_list = []
    return self

类方法 - del_dwell_points：

• 停留点识别：基于距离阈值对停留点进行识别并且进行删除

Parameters:

Name	Type	Description	Default
dwell_l_length	float	停留点识别距离阈值(米)	5.0
dwell_n	int	大于等于0的整数, 超过连续dwell_n + 1个相邻GPS点的距离小于dwell_l_length，那么这一组点就会被识别为停留点	2

Returns:

Type	Description
	self

Source code in src/gotrackit/gps/LocGps.py

def del_dwell_points(self, dwell_l_length: float = 5.0, dwell_n: int = 2):
    """类方法 - del_dwell_points：

    - 停留点识别：基于距离阈值对停留点进行识别并且进行删除

    Args:
        dwell_l_length: 停留点识别距离阈值(米)
        dwell_n: 大于等于0的整数, 超过连续dwell_n + 1个相邻GPS点的距离小于dwell_l_length，那么这一组点就会被识别为停留点

    Returns:
        self
    """
    # add field = dis_gap_field
    self.calc_adj_dis_gap()
    del self.__gps_points_gdf[next_p_field]
    self.__gps_points_gdf['dwell_label'] = \
        (self.__gps_points_gdf[dis_gap_field] > dwell_l_length).astype(int)
    del self.__gps_points_gdf[dis_gap_field]

    not_same_agent_idx = self.not_same_agent_idx()
    self.__gps_points_gdf.loc[not_same_agent_idx, 'dwell_label'] = 1
    self.__gps_points_gdf.loc[self.__gps_points_gdf.tail(1).index, 'dwell_label'] = 1
    self.__gps_points_gdf = self.del_consecutive_zero(df=self.__gps_points_gdf, col='dwell_label', n=dwell_n)
    del self.__gps_points_gdf['dwell_label']
    try:
        self.__gps_points_gdf.drop(columns=[sub_group_field], inplace=True)
    except KeyError:
        pass
    return self

类方法 - simplify_trajectory：

• 简化轨迹：利用道格拉斯-普克算法对轨迹进行抽稀简化

Parameters:

Name	Type	Description	Default
l_threshold	float	抽稀阈值(米)	5.0

Returns:

Type	Description
	self

Source code in src/gotrackit/gps/LocGps.py

def simplify_trajectory(self, l_threshold: float = 5.0):
    """类方法 - simplify_trajectory：

    - 简化轨迹：利用道格拉斯-普克算法对轨迹进行抽稀简化

    Args:
        l_threshold: 抽稀阈值(米)

    Returns:
        self
    """
    agent_count = self.__gps_points_gdf.groupby(agent_field)[[time_field]].count().rename(columns={time_field: 'c'})
    one_agent = list(agent_count[agent_count['c'] <= 1].index)
    process_trajectory = self.__gps_points_gdf[~self.__gps_points_gdf[agent_field].isin(one_agent)].copy()

    if not process_trajectory.empty:
        process_trajectory.reset_index(inplace=True, drop=True)

        no_process_gps = self.__gps_points_gdf[self.__gps_points_gdf[agent_field].isin(one_agent)].copy()

        origin_crs = self.__gps_points_gdf.crs
        del self.__gps_points_gdf

        line_gdf = process_trajectory.groupby(agent_field)[[geometry_field]].agg(
            {geometry_field: list}).reset_index(drop=False)
        line_gdf[geometry_field] = line_gdf[geometry_field].apply(lambda p: LineString(p))
        line_gdf = gpd.GeoDataFrame(line_gdf, geometry=geometry_field, crs=origin_crs)
        line_gdf[geometry_field] = line_gdf[geometry_field].simplify(l_threshold)

        p_simplify_line = pd.merge(process_trajectory[[gps_field.AGENT_ID_FIELD]],
                                   line_gdf, on=gps_field.AGENT_ID_FIELD, how='left')
        p_simplify_line = gpd.GeoSeries(p_simplify_line[geometry_field])

        # origin point prj to simplify
        prj_p_array = p_simplify_line.project(process_trajectory[geometry_field])

        process_trajectory[geometry_field] = p_simplify_line.interpolate(prj_p_array)

        if not no_process_gps.empty:
            self.__gps_points_gdf = pd.concat([process_trajectory, no_process_gps])
        self.__gps_points_gdf = process_trajectory
        self.__gps_points_gdf.reset_index(inplace=True, drop=True)
    return self

类方法 - trajectory_data

• 获取轨迹：获取处理后的轨迹数据(支持DataFrame和GeoDataFrame)

Parameters:

Name	Type	Description	Default
export_crs	str	输出结果的坐标系	'EPSG:4326'
_type	str	输出结果的类型, gdf或者df	'gdf'

Returns:

Type	Description
GeoDataFrame or DataFrame	轨迹数据

Source code in src/gotrackit/gps/LocGps.py

def trajectory_data(self, export_crs: str = 'EPSG:4326', _type: str = "gdf") -> gpd.GeoDataFrame or pd.DataFrame:
    """类方法 - trajectory_data

    - 获取轨迹：获取处理后的轨迹数据(支持DataFrame和GeoDataFrame)

    Args:
        export_crs: 输出结果的坐标系
        _type: 输出结果的类型, gdf或者df

    Returns:
        轨迹数据
    """
    export_trajectory = self.__gps_points_gdf.copy()
    try:
        del export_trajectory[gps_field.PLAIN_X], export_trajectory[gps_field.PLAIN_Y]
    except Exception as e:
        print(repr(e))
    export_trajectory = export_trajectory.to_crs(export_crs)
    export_trajectory = pd.merge(export_trajectory, self.__user_gps_info,
                                 on=[agent_field, gps_field.POINT_SEQ_FIELD], how='left')
    export_trajectory[gps_field.LOC_TYPE] = export_trajectory[gps_field.LOC_TYPE].fillna('d')
    export_trajectory[lng_field] = export_trajectory[geometry_field].x
    export_trajectory[lat_field] = export_trajectory[geometry_field].y
    if _type == 'df':
        del export_trajectory[geometry_field]
        return export_trajectory
    return export_trajectory

TrajectoryPoints类方法 - export_html

导出HTML：将处理后的轨迹导出为HTML，可动态展示处理前后的轨迹

Parameters:

Name	Type	Description	Default
out_fldr	str	存储目录	'./'
file_name	str	文件名称	'trajectory'
radius	float	点的半径大小	10.0

Returns:

Type	Description
	None

Source code in src/gotrackit/gps/Trajectory.py

def export_html(self, out_fldr: str = r'./', file_name: str = 'trajectory', radius: float = 10.0):
    """TrajectoryPoints类方法 - export_html

    导出HTML：将处理后的轨迹导出为HTML，可动态展示处理前后的轨迹

    Args:
        out_fldr: 存储目录
        file_name: 文件名称
        radius: 点的半径大小

    Returns:
        None
    """
    if self.already_plain:
        origin_tj_df = self.source_gps.to_crs(prj_const.PRJ_CRS)
        origin_tj_df = origin_tj_df[[agent_field, time_field, geometry_field]].copy()
        origin_tj_df[lng_field] = origin_tj_df[geometry_field].x
        origin_tj_df[lat_field] = origin_tj_df[geometry_field].y
        del origin_tj_df[geometry_field]
    else:
        origin_tj_df = self.source_gps[[agent_field, lng_field, lat_field, time_field]].copy()

    tj_gdf = self.trajectory_data(export_crs=prj_const.PRJ_CRS, _type='df')
    if {gps_field.X_SPEED_FIELD, gps_field.Y_SPEED_FIELD}.issubset(set(tj_gdf.columns)):
        tj_df = tj_gdf[[agent_field, lng_field, lat_field,
                        gps_field.X_SPEED_FIELD, gps_field.Y_SPEED_FIELD, time_field]].copy()
    else:
        tj_df = tj_gdf[[agent_field, lng_field, lat_field, time_field]].copy()
    del tj_gdf
    tj_df['type'] = 'process'
    origin_tj_df['type'] = 'source'
    tj_df[time_field] = tj_df[time_field].astype(origin_tj_df[time_field].dtype)
    df = pd.concat([tj_df, origin_tj_df]).reset_index(drop=True, inplace=False)
    for agent_id, _df in df.groupby(agent_field):
        vis_df = pd.DataFrame(_df)
        vis_df.sort_values(by='type', ascending=False, inplace=True)
        cen_x, cen_y = vis_df[gps_field.LNG_FIELD].mean(), vis_df[gps_field.LAT_FIELD].mean()
        # vis_df[gps_field.TIME_FIELD] = vis_df[gps_field.TIME_FIELD].astype(str)
        try:
            kv = KeplerVis(cen_loc=[cen_x, cen_y])
            kv.add_point_layer(vis_df, lng_field=lng_field, lat_field=lat_field, time_format=self.time_format,
                               time_unit=self.time_unit, set_avg_zoom=False, radius=radius,
                               time_field=time_field, layer_id='trajectory', color=[65, 72, 88],
                               color_field='type',
                               color_list=['#438ECD', '#FFC300'])
            kv.export_html(out_fldr=out_fldr, file_name=rf'{agent_id}_' + file_name)
        except Exception as e:
            print(repr(e))

Comments