fix(stats): reject sub-threshold gps drift
This commit is contained in:
@@ -22,6 +22,7 @@ stat-writer 已支持在累计里程缺失时,用有速度证据的相邻 GPS
|
|||||||
- 重复累计里程仍保留为 `skipped_same_mileage` 证据;
|
- 重复累计里程仍保留为 `skipped_same_mileage` 证据;
|
||||||
- GB32960、宇通只有在设备速度大于 0 且坐标有效时才进入 GPS 累计;
|
- GB32960、宇通只有在设备速度大于 0 且坐标有效时才进入 GPS 累计;
|
||||||
- GPS 状态仍按 10 秒节流,并拒绝超过 10 分钟、时间倒序或隐含速度超过 220 km/h 的线段;
|
- GPS 状态仍按 10 秒节流,并拒绝超过 10 分钟、时间倒序或隐含速度超过 220 km/h 的线段;
|
||||||
|
- 低于 100 米的累计仅保留为内部状态,不生成业务日里程候选;
|
||||||
- 正向仪表累计里程优先于 GPS 估算,真实仪表值恢复后会自动覆盖估算;
|
- 正向仪表累计里程优先于 GPS 估算,真实仪表值恢复后会自动覆盖估算;
|
||||||
- 静止车辆不会因 GPS 漂移产生虚假里程。
|
- 静止车辆不会因 GPS 漂移产生虚假里程。
|
||||||
|
|
||||||
|
|||||||
@@ -511,7 +511,7 @@ BACKFILL_DATE_TO=2026-07-12 \
|
|||||||
|
|
||||||
宇通协议会把定位、车况和仪表里程拆成不同的稀疏帧。当天只有明确 `meter_speed=0` 的定位帧、但没有新的仪表总里程帧时,stat-writer 会用同一来源最近的历史总里程生成一条显式 `0 km` 记录,`quality_reason=stationary_carry_forward`;后续收到真实总里程后仍会正常覆盖增长。`vehicle_stat_samples_total{status="recovered_stationary"}` 用于观察该恢复路径。历史补录只能在轨迹核验确认为 `0 km` 后,临时为单次 `stats-backfill` 增加 `BACKFILL_STATIONARY_CARRY_FORWARD=true`;该开关默认关闭,禁止写入 timer,避免仅凭最新静止状态推断整日未行驶。
|
宇通协议会把定位、车况和仪表里程拆成不同的稀疏帧。当天只有明确 `meter_speed=0` 的定位帧、但没有新的仪表总里程帧时,stat-writer 会用同一来源最近的历史总里程生成一条显式 `0 km` 记录,`quality_reason=stationary_carry_forward`;后续收到真实总里程后仍会正常覆盖增长。`vehicle_stat_samples_total{status="recovered_stationary"}` 用于观察该恢复路径。历史补录只能在轨迹核验确认为 `0 km` 后,临时为单次 `stats-backfill` 增加 `BACKFILL_STATIONARY_CARRY_FORWARD=true`;该开关默认关闭,禁止写入 timer,避免仅凭最新静止状态推断整日未行驶。
|
||||||
|
|
||||||
三类协议都优先采用终端上报的累计里程字段。部分终端只有坐标、速度而累计里程稀疏或停止刷新时,stat-writer 会按同一 VIN、自然日、协议和物理来源累计相邻坐标距离,状态持久化在 `vehicle_daily_gps_mileage_state`,并以 `quality_reason=gps_coordinate_accumulation` 写入低优先级候选。该路径既覆盖总里程字段缺失,也覆盖字段存在但数值与上一帧相同;重复总里程仍计入 `skipped_same_mileage` 证据,只有同时具备有效移动点时才增加 GPS 候选。JT808 兼容没有速度字段的存量终端;GB32960、宇通 MQTT 必须同时满足设备速度大于 0,且实时状态写入按 10 秒节流,避免静止 GPS 漂移和 1 秒级宇通上报放大 MySQL 压力。相邻点超过 10 分钟、时间倒序或隐含速度超过 220 km/h 的线段不累计。选举顺序为“正向累计里程 > 正向 GPS 轨迹估算 > 累计里程零值占位”,因此稀疏累计里程造成的错误 0 km 会被已有移动轨迹纠正,真实正向累计里程恢复后仍会自动覆盖估算值;禁止把估算值混入平台累计里程归并。`vehicle_stat_samples_total{status="recovered_gps_coordinate"}` 和 `vehicle_stat_cache_entries{cache="gps_accumulation"}` 用于观察实时恢复量及节流缓存。`stats-backfill` 默认用 TDengine `vehicle_locations` 为缺少正向累计里程但存在可用移动线段的车辆补建同口径候选;可临时设置 `BACKFILL_GPS_FALLBACK=false` 禁用全部协议的降级路径,旧环境变量 `BACKFILL_JT808_GPS_FALLBACK` 继续兼容。
|
三类协议都优先采用终端上报的累计里程字段。部分终端只有坐标、速度而累计里程稀疏或停止刷新时,stat-writer 会按同一 VIN、自然日、协议和物理来源累计相邻坐标距离,状态持久化在 `vehicle_daily_gps_mileage_state`,并以 `quality_reason=gps_coordinate_accumulation` 写入低优先级候选。该路径既覆盖总里程字段缺失,也覆盖字段存在但数值与上一帧相同;重复总里程仍计入 `skipped_same_mileage` 证据,只有同时具备有效移动点时才增加 GPS 候选。JT808 兼容没有速度字段的存量终端;GB32960、宇通 MQTT 必须同时满足设备速度大于 0,且实时状态写入按 10 秒节流,避免静止 GPS 漂移和 1 秒级宇通上报放大 MySQL 压力。相邻点超过 10 分钟、时间倒序或隐含速度超过 220 km/h 的线段不累计;低于 100 米的自然日累计只保留在状态表中,不投影为业务候选,避免瞬时车速噪声和微小定位漂移生成虚假日里程。选举顺序为“正向累计里程 > 正向 GPS 轨迹估算 > 累计里程零值占位”,因此稀疏累计里程造成的错误 0 km 会被已有移动轨迹纠正,真实正向累计里程恢复后仍会自动覆盖估算值;禁止把估算值混入平台累计里程归并。`vehicle_stat_samples_total{status="recovered_gps_coordinate"}` 和 `vehicle_stat_cache_entries{cache="gps_accumulation"}` 用于观察实时恢复量及节流缓存。`stats-backfill` 默认用 TDengine `vehicle_locations` 为缺少正向累计里程但存在可用移动线段的车辆补建同口径候选;可临时设置 `BACKFILL_GPS_FALLBACK=false` 禁用全部协议的降级路径,旧环境变量 `BACKFILL_JT808_GPS_FALLBACK` 继续兼容。
|
||||||
|
|
||||||
生产定时补算使用 systemd timer,默认每天 01:30 补算昨天往前 3 天,覆盖上游晚到或断传后恢复的总里程字段:
|
生产定时补算使用 systemd timer,默认每天 01:30 补算昨天往前 3 天,覆盖上游晚到或断传后恢复的总里程字段:
|
||||||
|
|
||||||
|
|||||||
@@ -622,7 +622,7 @@ ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
|
|||||||
}
|
}
|
||||||
added := 0
|
added := 0
|
||||||
for target, state := range states {
|
for target, state := range states {
|
||||||
if state.usableSegments == 0 {
|
if !stats.GPSMileageCandidateEligible(state.distanceKM, state.usableSegments) {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
parts := strings.SplitN(target, "|", 2)
|
parts := strings.SplitN(target, "|", 2)
|
||||||
|
|||||||
@@ -2128,7 +2128,7 @@ func TestWriterDuplicateYutongOdometerStillEntersGPSFallback(t *testing.T) {
|
|||||||
EventTimeMS: eventTime.UnixMilli(),
|
EventTimeMS: eventTime.UnixMilli(),
|
||||||
Fields: map[string]any{
|
Fields: map[string]any{
|
||||||
"yutong_mqtt.data.total_mileage": 24245000,
|
"yutong_mqtt.data.total_mileage": 24245000,
|
||||||
"yutong_mqtt.data.longitude": 114.435672,
|
"yutong_mqtt.data.longitude": 114.436672,
|
||||||
"yutong_mqtt.data.latitude": 30.645438,
|
"yutong_mqtt.data.latitude": 30.645438,
|
||||||
"yutong_mqtt.data.meter_speed": 18,
|
"yutong_mqtt.data.meter_speed": 18,
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -17,6 +17,7 @@ const (
|
|||||||
gpsCoordinateSourceSuffix = "#GPS_COORDINATE"
|
gpsCoordinateSourceSuffix = "#GPS_COORDINATE"
|
||||||
gpsMaxSegmentGap = 10 * time.Minute
|
gpsMaxSegmentGap = 10 * time.Minute
|
||||||
gpsMaxImpliedSpeedKMH = 220.0
|
gpsMaxImpliedSpeedKMH = 220.0
|
||||||
|
gpsMinimumDailyDistanceKM = 0.1
|
||||||
)
|
)
|
||||||
|
|
||||||
const GPSMileageStateTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_gps_mileage_state (
|
const GPSMileageStateTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_gps_mileage_state (
|
||||||
@@ -238,7 +239,7 @@ func AccumulateGPSMileage(ctx context.Context, exec Execer, point GPSMileagePoin
|
|||||||
_ = tx.Rollback()
|
_ = tx.Rollback()
|
||||||
return GPSMileageState{}, false, err
|
return GPSMileageState{}, false, err
|
||||||
}
|
}
|
||||||
if state.UsableSegmentCount == 0 {
|
if !GPSMileageCandidateEligible(state.DailyMileageKM, state.UsableSegmentCount) {
|
||||||
if err := tx.Commit(); err != nil {
|
if err := tx.Commit(); err != nil {
|
||||||
return GPSMileageState{}, false, err
|
return GPSMileageState{}, false, err
|
||||||
}
|
}
|
||||||
@@ -258,6 +259,16 @@ func AccumulateGPSMileage(ctx context.Context, exec Execer, point GPSMileagePoin
|
|||||||
return state, true, nil
|
return state, true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// GPSMileageCandidateEligible keeps sub-100-metre stationary drift and
|
||||||
|
// momentary speed noise in the durable accumulation state without exposing it
|
||||||
|
// as a business daily-mileage candidate. Real movement continues accumulating
|
||||||
|
// and becomes eligible once the evidence crosses the minimum distance.
|
||||||
|
func GPSMileageCandidateEligible(distanceKM float64, usableSegmentCount int64) bool {
|
||||||
|
return usableSegmentCount > 0 &&
|
||||||
|
isFiniteNonNegative(distanceKM) &&
|
||||||
|
distanceKM >= gpsMinimumDailyDistanceKM
|
||||||
|
}
|
||||||
|
|
||||||
// AccumulateJT808GPSMileage remains as a compatibility wrapper for callers and
|
// AccumulateJT808GPSMileage remains as a compatibility wrapper for callers and
|
||||||
// operational tooling compiled against the original JT808-only fallback.
|
// operational tooling compiled against the original JT808-only fallback.
|
||||||
func AccumulateJT808GPSMileage(ctx context.Context, exec Execer, point GPSMileagePoint) (GPSMileageState, bool, error) {
|
func AccumulateJT808GPSMileage(ctx context.Context, exec Execer, point GPSMileagePoint) (GPSMileageState, bool, error) {
|
||||||
|
|||||||
@@ -151,6 +151,18 @@ func TestWriterThrottlesHighFrequencyGPSMileageStateWrites(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestGPSMileageCandidateEligibleRejectsStationaryDrift(t *testing.T) {
|
||||||
|
if GPSMileageCandidateEligible(0.099999, 10) {
|
||||||
|
t.Fatal("sub-100-metre GPS drift must not become a daily-mileage candidate")
|
||||||
|
}
|
||||||
|
if GPSMileageCandidateEligible(0.2, 0) {
|
||||||
|
t.Fatal("distance without a usable segment must not become a candidate")
|
||||||
|
}
|
||||||
|
if !GPSMileageCandidateEligible(0.1, 1) {
|
||||||
|
t.Fatal("100 metres with a usable segment should become eligible")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestAccumulateJT808GPSMileagePersistsEstimateAndProjects(t *testing.T) {
|
func TestAccumulateJT808GPSMileagePersistsEstimateAndProjects(t *testing.T) {
|
||||||
db, mock, err := sqlmock.New()
|
db, mock, err := sqlmock.New()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -250,3 +262,47 @@ func TestAccumulateJT808GPSMileageNeedsUsableSegmentBeforeProjection(t *testing.
|
|||||||
t.Fatalf("unmet SQL expectations: %v", err)
|
t.Fatalf("unmet SQL expectations: %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestAccumulateGPSMileageKeepsSubThresholdDistanceAsStateOnly(t *testing.T) {
|
||||||
|
db, mock, err := sqlmock.New()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("sqlmock.New() error = %v", err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
|
||||||
|
start := time.Date(2026, 7, 20, 0, 14, 56, 0, time.FixedZone("Asia/Shanghai", 8*3600))
|
||||||
|
point := GPSMileagePoint{
|
||||||
|
VIN: "LMRKH9ACXR1004130",
|
||||||
|
StatDate: "2026-07-20",
|
||||||
|
Protocol: envelope.ProtocolYutongMQTT,
|
||||||
|
SourceKey: "YUTONG_MQTT:LMRKH9ACXR1004130@PLATFORM:yutong",
|
||||||
|
SourceIP: "mqtt",
|
||||||
|
EventID: "evt-small-movement",
|
||||||
|
EventTime: start.Add(time.Second),
|
||||||
|
Longitude: 114.435573,
|
||||||
|
Latitude: 30.645503,
|
||||||
|
}
|
||||||
|
|
||||||
|
mock.ExpectBegin()
|
||||||
|
mock.ExpectQuery("SELECT first_event_time, latest_event_time").
|
||||||
|
WillReturnRows(sqlmock.NewRows([]string{
|
||||||
|
"first_event_time", "latest_event_time", "latest_event_id",
|
||||||
|
"latest_longitude", "latest_latitude", "daily_mileage_km",
|
||||||
|
"point_count", "usable_segment_count", "bad_jump_count",
|
||||||
|
"long_gap_count", "out_of_order_count",
|
||||||
|
}).AddRow(start, start, "evt-before", 114.435572, 30.645503, 0, 1, 0, 0, 0, 0))
|
||||||
|
mock.ExpectExec("UPDATE vehicle_daily_gps_mileage_state").
|
||||||
|
WillReturnResult(sqlmock.NewResult(0, 1))
|
||||||
|
mock.ExpectCommit()
|
||||||
|
|
||||||
|
state, recovered, err := AccumulateGPSMileage(context.Background(), db, point)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("AccumulateGPSMileage() error = %v", err)
|
||||||
|
}
|
||||||
|
if recovered || state.DailyMileageKM <= 0 || state.DailyMileageKM >= gpsMinimumDailyDistanceKM {
|
||||||
|
t.Fatalf("sub-threshold state = %+v recovered=%v", state, recovered)
|
||||||
|
}
|
||||||
|
if err := mock.ExpectationsWereMet(); err != nil {
|
||||||
|
t.Fatalf("unmet SQL expectations: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user