OnLineTrajectoryKF

实时卡尔曼滤波类OnLineTrajectoryKF：

• 初始化

Parameters:

Name	Type	Description	Default
trajectory_df	DataFrame	轨迹数据	None
time_format	str	GPS轨迹数据中时间列的格式化字符串模板	'%Y-%m-%d %H:%M:%S'
time_unit	str	轨迹数据中时间列的时间单位	's'
x_field	str	轨迹数据表中表示经度的字段名称	'lng'
y_field	str	轨迹数据表中表示纬度的字段名称	'lat'

Source code in src/gotrackit/tools/kf.py

def __init__(self, trajectory_df: pd.DataFrame = None, time_format: str = '%Y-%m-%d %H:%M:%S', time_unit: str = 's',
             x_field: str = 'lng', y_field: str = 'lat'):
    """实时卡尔曼滤波类OnLineTrajectoryKF：

    - 初始化

    Args:
        trajectory_df: 轨迹数据
        time_format: GPS轨迹数据中时间列的格式化字符串模板
        time_unit: 轨迹数据中时间列的时间单位
        x_field: 轨迹数据表中表示经度的字段名称
        y_field: 轨迹数据表中表示纬度的字段名称
    """

    TrajectoryKalmanFilter.__init__(self, trajectory_df)

    self.kf_group: dict[object, KalmanFilter] = dict()
    self.his_o: dict[object, [np.ndarray, np.ndarray]] = dict()
    self.his_t: dict = dict()
    self.x_field, self.y_field = x_field, y_field
    self.time_format, self.time_unit = time_format, time_unit
    if trajectory_df is not None:
        build_time_col(df=self.trajectory_df, time_format=time_format, time_unit=time_unit)

OnLineTrajectoryKF类方法 - kf_smooth：

• 平滑：实时滤波平滑

Parameters:

Name	Type	Description	Default
p_deviation	list or float	过程噪声的标准差	0.01
o_deviation	list or float	观测噪声的标准差, 该值越小, 平滑后的结果就越接近原轨迹	0.1
time_gap_threshold	float	时间阈值，如果某agent的当前批GPS数据的最早定位时间和上批GPS数据的最晚定位时间的差值超过该值，则不参考历史概率链进行匹配计算	1800.0

Returns:

Type	Description
DataFrame or GeoDataFrame	平滑后的轨迹表

Source code in src/gotrackit/tools/kf.py

def kf_smooth(self, p_deviation: list or float = 0.01, o_deviation: list or float = 0.1,
              time_gap_threshold: float = 1800.0) -> \
        pd.DataFrame or gpd.GeoDataFrame:
    """OnLineTrajectoryKF类方法 - kf_smooth：

    - 平滑：实时滤波平滑

    Args:
        p_deviation: 过程噪声的标准差
        o_deviation: 观测噪声的标准差, 该值越小, 平滑后的结果就越接近原轨迹
        time_gap_threshold: 时间阈值，如果某agent的当前批GPS数据的最早定位时间和上批GPS数据的最晚定位时间的差值超过该值，则不参考历史概率链进行匹配计算

    Returns:
        平滑后的轨迹表
    """
    res_df = pd.DataFrame()

    for agent_id, tj_df in self.trajectory_df.groupby(agent_field):

        observations = tj_df[[self.x_field, self.y_field]].values
        t = tj_df[time_field]

        start_index = 0
        if agent_id in self.kf_group.keys() and \
                (t.iloc[0] - self.his_t[agent_id]).total_seconds() <= time_gap_threshold:
            kf = self.kf_group[agent_id]
            smoothed_states = np.zeros((len(observations), 4))
        else:
            kf = self.init_kf(initial_x=observations[0, 0],
                              initial_y=observations[0, 1],
                              o_deviation=o_deviation, p_deviation=p_deviation)
            self.kf_group[agent_id] = kf
            self.his_o[agent_id] = [kf.initial_state_mean, kf.initial_state_covariance]
            self.his_t[agent_id] = t.iloc[0]
            smoothed_states = np.zeros((len(observations), 4))

            # save initial state
            smoothed_states[0, :] = [observations[0, 0], observations[0, 1], 0, 0]

            start_index = 1

        for i in range(start_index, len(observations)):
            now_state, now_covariance = self.his_o[agent_id]
            now_t = t.iloc[i]
            dt = (now_t - self.his_t[agent_id]).total_seconds()  # calculate time interval

            now_state, now_covariance = self.single_step_process(kf=kf, dt=dt, previous_state=now_state,
                                                                 previous_covariance=now_covariance,
                                                                 now_state=observations[i])
            smoothed_states[i, :] = now_state
            self.his_o[agent_id] = [now_state, now_covariance]
            self.his_t[agent_id] = now_t

        tj_df[self.x_field] = smoothed_states[:, 0]
        tj_df[self.y_field] = smoothed_states[:, 1]
        tj_df[x_speed_field] = smoothed_states[:, 2]
        tj_df[y_speed_field] = smoothed_states[:, 3]
        res_df = pd.concat([res_df, tj_df])
    res_df.reset_index(inplace=True, drop=True)
    return res_df

OnLineTrajectoryKF类方法 - renew_trajectory

• 更新数据：更新待处理的轨迹点数据

Parameters:

Name	Type	Description	Default
trajectory_df	DataFrame	轨迹数据	None

Returns:

Type	Description
	None

Source code in src/gotrackit/tools/kf.py

def renew_trajectory(self, trajectory_df: pd.DataFrame = None):
    """OnLineTrajectoryKF类方法 - renew_trajectory

    - 更新数据：更新待处理的轨迹点数据

    Args:
        trajectory_df: 轨迹数据

    Returns:
        None
    """
    self.trajectory_df = trajectory_df
    build_time_col(df=self.trajectory_df, time_format=self.time_format, time_unit=self.time_unit)

Comments