4、轨迹数据预处理:数据清洗、插值与压缩
大家好,我是老张。做轨迹回放这些年,我踩过最多的坑,其实不在算法本身,而在数据预处理阶段。
你想想看,拿到的原始轨迹数据,那叫一个「脏」。GPS漂移、重复点、丢帧……直接扔进算法里,结果肯定一塌糊涂。我刚开始带项目时,就吃过这个亏——回放出来的轨迹,车辆直接穿墙而过,客户当场黑脸。
所以今天咱们聊聊数据预处理。说白了,就是给原始数据「洗个澡」,让它变得干净、连续、轻量。我习惯把预处理分成三步:清洗、插值、压缩。下面这张图能帮你快速建立整体认知:
4.1 数据清洗:去重与去噪
数据清洗,我习惯先做去重,再做去噪。顺序别搞反了——先去掉重复点,再去处理异常点,逻辑上更顺畅。
4.1.1 去重
重复点怎么来的?最常见的情况是:设备在同一个位置连续上报了多次。比如车辆等红灯时,GPS没动,但上报了5条几乎一样的数据。
我个人习惯的做法是:按时间戳排序后,判断相邻点的经纬度差。如果差值小于某个阈值(比如0.00001度,约1米),就认为是重复点,只保留第一个。
def remove_duplicates(points, threshold=0.00001):
"""
points: [(lat, lng, timestamp), ...]
threshold: 经纬度差值阈值
"""
cleaned = [points[0]]
for i in range(1, len(points)):
lat_diff = abs(points[i][0] - points[i-1][0])
lng_diff = abs(points[i][1] - points[i-1][1])
if lat_diff > threshold or lng_diff > threshold:
cleaned.append(points[i])
return cleaned
4.1.2 去噪
去噪,说白了就是干掉那些「离谱」的点。GPS信号受高楼、隧道、天气影响,偶尔会蹦出一个离群点。比如车辆明明在A点,突然跳到了500米外的B点,下一秒又跳回来。
我曾经在做一个物流项目时,遇到一辆货车轨迹里有个点直接漂到了海里。回放时,货车「游泳」了3秒钟……客户截图发过来,我那个尴尬啊。
去噪我常用两种方法:
- 速度阈值法: 计算相邻点之间的速度,如果超过合理范围(比如城市道路超过120km/h),就判定为噪声点。
- 距离突变法: 如果某点与前后两点的距离都远大于正常步长,直接剔除。
def remove_noise_by_speed(points, max_speed=33.3):
"""
max_speed: 最大合理速度,单位m/s(约120km/h)
"""
cleaned = [points[0]]
for i in range(1, len(points)):
lat1, lng1, t1 = points[i-1]
lat2, lng2, t2 = points[i]
# 简化计算:1度经纬度约111km
dist = ((lat2-lat1)**2 + (lng2-lng1)**2)**0.5 * 111000
dt = abs(t2 - t1)
if dt > 0:
speed = dist / dt
if speed <= max_speed:
cleaned.append(points[i])
return cleaned
4.2 数据插值:线性插值与样条插值
清洗完数据,你会发现轨迹里有很多「空洞」。设备丢帧、信号中断,都会导致时间不连续。插值就是把这些洞补上。
嗯,这里要注意:插值不是万能的。如果丢帧超过10秒,我建议直接标记为「数据缺失」,而不是强行补一个假点。
4.2.1 线性插值
线性插值最简单,也最常用。它假设物体在两个已知点之间做匀速直线运动。说白了,就是「两点之间画条直线,中间的点按比例取」。我80%的场景都用它,够用且快。
def linear_interpolate(p1, p2, t):
"""
p1, p2: (lat, lng, timestamp)
t: 目标时间戳
"""
lat1, lng1, t1 = p1
lat2, lng2, t2 = p2
ratio = (t - t1) / (t2 - t1)
lat = lat1 + (lat2 - lat1) * ratio
lng = lng1 + (lng2 - lng1) * ratio
return (lat, lng, t)
4.2.2 样条插值
线性插值有个硬伤——拐弯处不平滑。你想想看,车辆在路口转弯,你用直线去连,轨迹就变成了「折线」,看着很生硬。
样条插值就不一样了。它用多项式曲线去拟合数据点,保证曲线在连接处一阶、二阶导数连续。说白了,就是让轨迹「弯」得更自然。
我记得有一次做赛车轨迹回放,客户要求画面必须丝滑。线性插值做出来,赛车过弯像「漂移」——其实是折线导致的视觉跳跃。换成三次样条插值后,效果立马好了。
from scipy.interpolate import CubicSpline
import numpy as np
def spline_interpolate(points, target_times):
"""
points: [(lat, lng, timestamp), ...]
target_times: 需要插值的时间点列表
"""
times = np.array([p[2] for p in points])
lats = np.array([p[0] for p in points])
lngs = np.array([p[1] for p in points])
cs_lat = CubicSpline(times, lats)
cs_lng = CubicSpline(times, lngs)
result = []
for t in target_times:
result.append((cs_lat(t), cs_lng(t), t))
return result
4.3 数据压缩:道格拉斯-普克算法
数据压缩,说白了就是「删掉不重要的点,保留轨迹形状」。你想想看,一条直线上的100个点,其实只需要首尾2个点就够了。
道格拉斯-普克算法(简称DP算法)是轨迹压缩领域的经典方法。它的核心思想很简单:
- 连接轨迹首尾两点,形成一条基线。
- 找到离基线最远的点,计算距离。
- 如果距离大于阈值,保留该点,并递归处理左右两段。
- 如果距离小于阈值,删除中间所有点。
我刚开始用这个算法时,阈值设得太小,压缩率只有10%,等于没压。后来调大了,又压过头了,轨迹变成了「折线」。嗯,阈值这东西,得根据场景调。
def douglas_peucker(points, epsilon):
"""
points: [(lat, lng), ...]
epsilon: 距离阈值,单位度
"""
if len(points) <= 2:
return points
# 找到离基线最远的点
dmax = 0
index = 0
for i in range(1, len(points) - 1):
d = perpendicular_distance(points[i], points[0], points[-1])
if d > dmax:
dmax = d
index = i
# 递归处理
if dmax > epsilon:
left = douglas_peucker(points[:index+1], epsilon)
right = douglas_peucker(points[index:], epsilon)
return left[:-1] + right
else:
return [points[0], points[-1]]
def perpendicular_distance(point, line_start, line_end):
"""计算点到直线的垂直距离"""
# 简化实现,实际项目中用经纬度球面距离
x0, y0 = point
x1, y1 = line_start
x2, y2 = line_end
return abs((y2-y1)*x0 - (x2-x1)*y0 + x2*y1 - y2*x1) / ((y2-y1)**2 + (x2-x1)**2)**0.5
| 场景 | 推荐阈值(度) | 压缩率 |
|---|---|---|
| 城市道路 | 0.0001 ~ 0.0005 | 60% ~ 80% |
| 高速公路 | 0.0005 ~ 0.001 | 80% ~ 95% |
| 步行轨迹 | 0.00005 ~ 0.0001 | 50% ~ 70% |
好了,数据预处理这块就聊到这儿。清洗、插值、压缩,三步走完,你的轨迹数据基本就「能打」了。下一节咱们开始讲轨迹匹配,那才是真正考验算法功底的地方。
可可精品资料库