From 4c64c4c738536b540e0d7b679048cfa301dc893d Mon Sep 17 00:00:00 2001 From: lingniu Date: Sun, 19 Jul 2026 22:57:31 +0800 Subject: [PATCH] fix(mileage): recover sparse source daily distance --- docs/ops/vehicle-ingest-runbook.md | 2 +- go/vehicle-gateway/cmd/stat-writer/main.go | 1 + go/vehicle-gateway/cmd/stats-backfill/main.go | 75 +++++++-- .../cmd/stats-backfill/main_test.go | 62 ++++++- .../internal/stats/daily_metric.go | 155 ++++++++++++------ .../internal/stats/gps_mileage.go | 39 ++++- .../internal/stats/gps_mileage_test.go | 64 ++++++++ .../internal/stats/source_mileage.go | 26 ++- .../internal/stats/source_mileage_test.go | 10 +- .../api/internal/platform/mysql_queries.go | 5 +- .../internal/platform/query_builders_test.go | 5 + 11 files changed, 364 insertions(+), 80 deletions(-) diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index 5e5e064e..e9908baf 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -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，避免仅凭最新静止状态推断整日未行驶。 -JT808 仍优先采用终端上报的 `jt808.location.total_mileage_km`。部分终端只上报坐标、速度而没有累计里程时，stat-writer 会按同一 VIN、自然日和物理来源累计相邻坐标距离，状态持久化在 `vehicle_daily_gps_mileage_state`，并以 `quality_reason=gps_coordinate_accumulation` 写入低优先级候选。相邻点超过 10 分钟、时间倒序或隐含速度超过 220 km/h 的线段不累计；真实累计里程候选一旦出现，平台来源优先级会自动覆盖坐标估算，禁止把估算值混入平台累计里程归并。`vehicle_stat_samples_total{status="recovered_gps_coordinate"}` 用于观察实时恢复量。`stats-backfill` 默认用 TDengine `vehicle_locations` 为缺少累计里程且至少有一个可用相邻线段的 JT808 车辆补建同口径候选；需要诊断对照时可临时设置 `BACKFILL_JT808_GPS_FALLBACK=false` 禁用该降级路径。 +三类协议都优先采用终端上报的累计里程字段。部分终端只有坐标、速度而累计里程稀疏或停止刷新时，stat-writer 会按同一 VIN、自然日、协议和物理来源累计相邻坐标距离，状态持久化在 `vehicle_daily_gps_mileage_state`，并以 `quality_reason=gps_coordinate_accumulation` 写入低优先级候选。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` 继续兼容。 生产定时补算使用 systemd timer，默认每天 01:30 补算昨天往前 3 天，覆盖上游晚到或断传后恢复的总里程字段： diff --git a/go/vehicle-gateway/cmd/stat-writer/main.go b/go/vehicle-gateway/cmd/stat-writer/main.go index 20cbc450..92be79f7 100644 --- a/go/vehicle-gateway/cmd/stat-writer/main.go +++ b/go/vehicle-gateway/cmd/stat-writer/main.go @@ -861,6 +861,7 @@ func recordStatCacheMetrics(registry *metrics.Registry, appender statAppender) { setStatCacheGauge(registry, "projection", stats.LastProjectionEntries, stats.MaxEntries, stats.LastCleanupProjection, stats.ProjectionEvictions) registry.SetGauge("vehicle_stat_pending_projection_entries", nil, float64(stats.PendingProjectionEntries)) setStatCacheGauge(registry, "baseline", stats.BaselineEntries, stats.MaxEntries, stats.LastCleanupBaseline, stats.BaselineEvictions) + setStatCacheGauge(registry, "gps_accumulation", stats.GPSAccumulationEntries, stats.MaxEntries, stats.LastCleanupGPS, stats.GPSEvictions) if !stats.LastCleanupAt.IsZero() { registry.SetGauge("vehicle_stat_cache_last_cleanup_unix_seconds", nil, float64(stats.LastCleanupAt.Unix())) } diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go index 899d9f36..fab4430b 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main.go @@ -36,7 +36,7 @@ type config struct { Debug bool EventTimeFullScan bool StationaryCarry bool - JT808GPSFallback bool + GPSFallback bool ProgressEvery int64 BaselineLookback int Location *time.Location @@ -151,9 +151,9 @@ func main() { if err != nil { fail("build realtime-location fallback aggregates", err) } - gpsFallbacks, err := addJT808GPSCoordinateFallbackAggregates(ctx, mysqlDB, td, cfg, aggregates) + gpsFallbacks, err := addGPSCoordinateFallbackAggregates(ctx, mysqlDB, td, cfg, aggregates) if err != nil { - fail("build JT808 GPS-coordinate fallback aggregates", err) + fail("build GPS-coordinate fallback aggregates", err) } var written int64 var normalized int @@ -516,29 +516,47 @@ type gpsCoordinateBackfillState struct { longGaps int64 } -func addJT808GPSCoordinateFallbackAggregates(ctx context.Context, mysqlDB *sql.DB, tdDB *sql.DB, cfg config, aggregates map[string]*metricAgg) (int, error) { - if !cfg.JT808GPSFallback || tdDB == nil || aggregates == nil || !containsProtocol(cfg.Protocols, envelope.ProtocolJT808) { +func addGPSCoordinateFallbackAggregates(ctx context.Context, mysqlDB *sql.DB, tdDB *sql.DB, cfg config, aggregates map[string]*metricAgg) (int, error) { + if !cfg.GPSFallback || tdDB == nil || aggregates == nil { return 0, nil } - existing, err := existingJT808OdometerTargets(ctx, mysqlDB, cfg) + totalAdded := 0 + for _, protocol := range cfg.Protocols { + added, err := addProtocolGPSCoordinateFallbackAggregates(ctx, mysqlDB, tdDB, cfg, aggregates, protocol) + if err != nil { + return totalAdded, err + } + totalAdded += added + } + return totalAdded, nil +} + +func addProtocolGPSCoordinateFallbackAggregates(ctx context.Context, mysqlDB *sql.DB, tdDB *sql.DB, cfg config, aggregates map[string]*metricAgg, protocol envelope.Protocol) (int, error) { + if !supportsBackfillGPSFallback(protocol) { + return 0, nil + } + existing, err := existingPositiveOdometerTargets(ctx, mysqlDB, cfg, protocol) if err != nil { return 0, err } for _, agg := range aggregates { - if agg != nil && agg.Protocol == envelope.ProtocolJT808 && agg.QualityReason != stats.QualityReasonGPSCoordinate { + if agg != nil && agg.Protocol == protocol && agg.QualityReason != stats.QualityReasonGPSCoordinate && agg.LatestKM > agg.FirstKM { existing[agg.VIN+"|"+agg.Date] = struct{}{} } } where := []string{ fmt.Sprintf("ts >= '%s 00:00:00'", quote(cfg.DateFrom)), fmt.Sprintf("ts < '%s 00:00:00'", quote(nextDate(cfg.DateTo))), - "protocol = 'JT808'", + fmt.Sprintf("protocol = '%s'", quote(string(protocol))), "vin IS NOT NULL", "vin <> ''", "longitude BETWEEN -180 AND 180", "latitude BETWEEN -90 AND 90", "NOT (longitude = 0 AND latitude = 0)", } + if protocol != envelope.ProtocolJT808 { + where = append(where, "speed_kmh > 0") + } sqlText := fmt.Sprintf(`SELECT vin, ts, longitude, latitude FROM %s.vehicle_locations WHERE %s @@ -566,7 +584,7 @@ ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) current := stats.GPSMileagePoint{ VIN: vin, StatDate: date, - Protocol: envelope.ProtocolJT808, + Protocol: protocol, EventTime: eventTime, Longitude: longitude, Latitude: latitude, @@ -612,18 +630,18 @@ ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) continue } vin, date := parts[0], parts[1] - sourceKey := string(envelope.ProtocolJT808) + ":" + vin + "@GPS_COORDINATE" - key := vin + "|" + date + "|" + string(envelope.ProtocolJT808) + "|" + sourceKey + sourceKey := string(protocol) + ":" + vin + "@GPS_COORDINATE" + key := vin + "|" + date + "|" + string(protocol) + "|" + sourceKey aggregates[key] = &metricAgg{ VIN: vin, Date: date, - Protocol: envelope.ProtocolJT808, + Protocol: protocol, FirstKM: 0, LatestKM: state.distanceKM, Count: state.pointCount, SourceKey: sourceKey, SourceEndpoint: "gps-coordinate", - PlatformName: "JT808 GPS轨迹估算", + PlatformName: backfillGPSPlatformName(protocol), SourceKind: "UNKNOWN", FirstEventTime: state.firstEventTime, LatestEventTime: state.latestEventTime, @@ -633,23 +651,44 @@ ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) added++ } if len(states) > 0 { - slog.Info("JT808 GPS-coordinate fallbacks loaded", "states", len(states), "added", added) + slog.Info("GPS-coordinate fallbacks loaded", "protocol", protocol, "states", len(states), "added", added) } return added, nil } -func existingJT808OdometerTargets(ctx context.Context, db *sql.DB, cfg config) (map[string]struct{}, error) { +func supportsBackfillGPSFallback(protocol envelope.Protocol) bool { + switch protocol { + case envelope.ProtocolGB32960, envelope.ProtocolJT808, envelope.ProtocolYutongMQTT: + return true + default: + return false + } +} + +func backfillGPSPlatformName(protocol envelope.Protocol) string { + switch protocol { + case envelope.ProtocolGB32960: + return "GB32960 GPS轨迹估算" + case envelope.ProtocolYutongMQTT: + return "宇通 GPS轨迹估算" + default: + return "JT808 GPS轨迹估算" + } +} + +func existingPositiveOdometerTargets(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol) (map[string]struct{}, error) { result := map[string]struct{}{} if db == nil { return result, nil } rows, err := db.QueryContext(ctx, `SELECT DISTINCT vin, DATE_FORMAT(stat_date, '%Y-%m-%d') FROM vehicle_daily_mileage_source -WHERE protocol = 'JT808' +WHERE protocol = ? AND stat_date >= ? AND stat_date <= ? AND quality_status = ? AND COALESCE(quality_reason, '') <> ? - AND latest_total_mileage_km IS NOT NULL`, cfg.DateFrom, cfg.DateTo, stats.QualityOK, stats.QualityReasonGPSCoordinate) + AND daily_mileage_km > 0 + AND latest_total_mileage_km IS NOT NULL`, string(protocol), cfg.DateFrom, cfg.DateTo, stats.QualityOK, stats.QualityReasonGPSCoordinate) if err != nil { return nil, err } @@ -1184,7 +1223,7 @@ func loadConfig() (config, error) { Debug: envBool("BACKFILL_DEBUG", false), EventTimeFullScan: envBool("BACKFILL_EVENT_TIME_FULL_SCAN", false), StationaryCarry: envBool("BACKFILL_STATIONARY_CARRY_FORWARD", false), - JT808GPSFallback: envBool("BACKFILL_JT808_GPS_FALLBACK", true), + GPSFallback: envBool("BACKFILL_GPS_FALLBACK", envBool("BACKFILL_JT808_GPS_FALLBACK", true)), ProgressEvery: int64(envInt("BACKFILL_PROGRESS_EVERY", 100000)), BaselineLookback: envInt("BACKFILL_BASELINE_LOOKBACK_DAYS", 7), Location: loc, diff --git a/go/vehicle-gateway/cmd/stats-backfill/main_test.go b/go/vehicle-gateway/cmd/stats-backfill/main_test.go index e9709585..12cd84c4 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main_test.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main_test.go @@ -990,7 +990,7 @@ func TestResolveBackfillDateRangeDefaultsToToday(t *testing.T) { } } -func TestAddJT808GPSCoordinateFallbackAggregatesOnlyMissingOdometerVehicles(t *testing.T) { +func TestAddGPSCoordinateFallbackAggregatesOnlyMissingPositiveOdometerVehicles(t *testing.T) { mysqlDB, mysqlMock, err := sqlmock.New() if err != nil { t.Fatalf("mysql sqlmock.New() error = %v", err) @@ -1003,7 +1003,7 @@ func TestAddJT808GPSCoordinateFallbackAggregatesOnlyMissingOdometerVehicles(t *t defer tdDB.Close() mysqlMock.ExpectQuery("SELECT DISTINCT vin, DATE_FORMAT"). - WithArgs("2026-07-19", "2026-07-19", stats.QualityOK, stats.QualityReasonGPSCoordinate). + WithArgs("JT808", "2026-07-19", "2026-07-19", stats.QualityOK, stats.QualityReasonGPSCoordinate). WillReturnRows(sqlmock.NewRows([]string{"vin", "stat_date"}). AddRow("VIN-WITH-ODOMETER", "2026-07-19")) loc := time.FixedZone("Asia/Shanghai", 8*3600) @@ -1016,16 +1016,16 @@ func TestAddJT808GPSCoordinateFallbackAggregatesOnlyMissingOdometerVehicles(t *t AddRow("VIN-WITH-ODOMETER", start.Add(5*time.Minute), 121.4837, 31.2304)) aggregates := map[string]*metricAgg{} - added, err := addJT808GPSCoordinateFallbackAggregates(context.Background(), mysqlDB, tdDB, config{ + added, err := addGPSCoordinateFallbackAggregates(context.Background(), mysqlDB, tdDB, config{ DateFrom: "2026-07-19", DateTo: "2026-07-19", Protocols: []envelope.Protocol{envelope.ProtocolJT808}, - JT808GPSFallback: true, + GPSFallback: true, Location: loc, TDengineDatabase: "vehicle_ts", }, aggregates) if err != nil { - t.Fatalf("addJT808GPSCoordinateFallbackAggregates() error = %v", err) + t.Fatalf("addGPSCoordinateFallbackAggregates() error = %v", err) } if added != 1 || len(aggregates) != 1 { t.Fatalf("added=%d aggregates=%d", added, len(aggregates)) @@ -1046,6 +1046,58 @@ func TestAddJT808GPSCoordinateFallbackAggregatesOnlyMissingOdometerVehicles(t *t } } +func TestAddGPSCoordinateFallbackAggregatesUsesMovingYutongPoints(t *testing.T) { + mysqlDB, mysqlMock, err := sqlmock.New() + if err != nil { + t.Fatalf("mysql sqlmock.New() error = %v", err) + } + defer mysqlDB.Close() + tdDB, tdMock, err := sqlmock.New() + if err != nil { + t.Fatalf("tdengine sqlmock.New() error = %v", err) + } + defer tdDB.Close() + + mysqlMock.ExpectQuery("SELECT DISTINCT vin, DATE_FORMAT"). + WithArgs("YUTONG_MQTT", "2026-07-19", "2026-07-19", stats.QualityOK, stats.QualityReasonGPSCoordinate). + WillReturnRows(sqlmock.NewRows([]string{"vin", "stat_date"})) + loc := time.FixedZone("Asia/Shanghai", 8*3600) + start := time.Date(2026, 7, 19, 8, 0, 0, 0, loc) + tdMock.ExpectQuery("protocol = 'YUTONG_MQTT'.*speed_kmh > 0"). + WillReturnRows(sqlmock.NewRows([]string{"vin", "ts", "longitude", "latitude"}). + AddRow("VIN-YUTONG-GPS", start, 121.4737, 31.2304). + AddRow("VIN-YUTONG-GPS", start.Add(time.Minute), 121.4837, 31.2304)) + + aggregates := map[string]*metricAgg{} + added, err := addGPSCoordinateFallbackAggregates(context.Background(), mysqlDB, tdDB, config{ + DateFrom: "2026-07-19", + DateTo: "2026-07-19", + Protocols: []envelope.Protocol{envelope.ProtocolYutongMQTT}, + GPSFallback: true, + Location: loc, + TDengineDatabase: "vehicle_ts", + }, aggregates) + if err != nil { + t.Fatalf("addGPSCoordinateFallbackAggregates() error = %v", err) + } + if added != 1 || len(aggregates) != 1 { + t.Fatalf("added=%d aggregates=%d", added, len(aggregates)) + } + for _, aggregate := range aggregates { + if aggregate.Protocol != envelope.ProtocolYutongMQTT || + aggregate.PlatformName != "宇通 GPS轨迹估算" || + aggregate.QualityReason != stats.QualityReasonGPSCoordinate { + t.Fatalf("aggregate = %+v", aggregate) + } + } + if err := mysqlMock.ExpectationsWereMet(); err != nil { + t.Fatalf("mysql expectations: %v", err) + } + if err := tdMock.ExpectationsWereMet(); err != nil { + t.Fatalf("tdengine expectations: %v", err) + } +} + func TestResolveBackfillDateRangeUsesRelativeWindow(t *testing.T) { t.Setenv("BACKFILL_DATE_FROM", "") t.Setenv("BACKFILL_DATE_TO", "") diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index 1a237238..fa9152ee 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -16,12 +16,13 @@ import ( ) const ( - maxNegativeMileageJitterKM = 1.0 - defaultCacheRetention = 72 * time.Hour - defaultCacheCleanupInterval = 10 * time.Minute - defaultBaselineMissTTL = time.Minute - defaultBaselineHitTTL = 5 * time.Minute - defaultMaxCacheEntries = 1000000 + maxNegativeMileageJitterKM = 1.0 + defaultCacheRetention = 72 * time.Hour + defaultCacheCleanupInterval = 10 * time.Minute + defaultBaselineMissTTL = time.Minute + defaultBaselineHitTTL = 5 * time.Minute + defaultMaxCacheEntries = 1000000 + defaultGPSAccumulationInterval = 10 * time.Second ) type Execer interface { @@ -29,28 +30,30 @@ type Execer interface { } type Writer struct { - exec Execer - query Queryer - loc *time.Location - sourceTouchInterval time.Duration - projectionInterval time.Duration - cacheRetention time.Duration - cacheCleanupInterval time.Duration - baselineMissTTL time.Duration - baselineHitTTL time.Duration - maxCacheEntries int - lastCacheCleanup time.Time - lastCacheCleanupStats cacheCleanupStats - cacheEvictions cacheCleanupStats - mu sync.Mutex - lastTotalMileage map[string]float64 - lastSourceSeen map[string]time.Time - lastProjection map[string]projectionCacheEntry - pendingProjection map[string]pendingProjectionEntry - baselineCache map[string]sourceBaselineCacheEntry - mileageKeysByPrefix map[string]map[string]struct{} - projectionKeysByPrefix map[string]map[string]struct{} - baselineKeysByPrefix map[string]map[string]struct{} + exec Execer + query Queryer + loc *time.Location + sourceTouchInterval time.Duration + projectionInterval time.Duration + cacheRetention time.Duration + cacheCleanupInterval time.Duration + baselineMissTTL time.Duration + baselineHitTTL time.Duration + maxCacheEntries int + lastCacheCleanup time.Time + lastCacheCleanupStats cacheCleanupStats + cacheEvictions cacheCleanupStats + mu sync.Mutex + lastTotalMileage map[string]float64 + lastSourceSeen map[string]time.Time + lastProjection map[string]projectionCacheEntry + pendingProjection map[string]pendingProjectionEntry + baselineCache map[string]sourceBaselineCacheEntry + mileageKeysByPrefix map[string]map[string]struct{} + projectionKeysByPrefix map[string]map[string]struct{} + baselineKeysByPrefix map[string]map[string]struct{} + lastGPSAccumulation map[string]time.Time + gpsAccumulationInterval time.Duration } type MetricSample struct { @@ -109,16 +112,19 @@ type CacheStats struct { LastProjectionEntries int PendingProjectionEntries int BaselineEntries int + GPSAccumulationEntries int MaxEntries int LastCleanupAt time.Time LastCleanupTotalMileage int LastCleanupSourceSeen int LastCleanupProjection int LastCleanupBaseline int + LastCleanupGPS int TotalMileageEvictions int SourceSeenEvictions int ProjectionEvictions int BaselineEvictions int + GPSEvictions int } type cacheCleanupStats struct { @@ -126,6 +132,7 @@ type cacheCleanupStats struct { sourceSeen int projection int baseline int + gps int } func NewWriter(exec Execer, loc *time.Location) *Writer { @@ -136,23 +143,25 @@ func NewWriter(exec Execer, loc *time.Location) *Writer { loc = time.FixedZone("Asia/Shanghai", 8*3600) } writer := &Writer{ - exec: exec, - loc: loc, - sourceTouchInterval: time.Minute, - projectionInterval: 15 * time.Second, - cacheRetention: defaultCacheRetention, - cacheCleanupInterval: defaultCacheCleanupInterval, - baselineMissTTL: defaultBaselineMissTTL, - baselineHitTTL: defaultBaselineHitTTL, - maxCacheEntries: defaultMaxCacheEntries, - lastTotalMileage: map[string]float64{}, - lastSourceSeen: map[string]time.Time{}, - lastProjection: map[string]projectionCacheEntry{}, - pendingProjection: map[string]pendingProjectionEntry{}, - baselineCache: map[string]sourceBaselineCacheEntry{}, - mileageKeysByPrefix: map[string]map[string]struct{}{}, - projectionKeysByPrefix: map[string]map[string]struct{}{}, - baselineKeysByPrefix: map[string]map[string]struct{}{}, + exec: exec, + loc: loc, + sourceTouchInterval: time.Minute, + projectionInterval: 15 * time.Second, + cacheRetention: defaultCacheRetention, + cacheCleanupInterval: defaultCacheCleanupInterval, + baselineMissTTL: defaultBaselineMissTTL, + baselineHitTTL: defaultBaselineHitTTL, + maxCacheEntries: defaultMaxCacheEntries, + lastTotalMileage: map[string]float64{}, + lastSourceSeen: map[string]time.Time{}, + lastProjection: map[string]projectionCacheEntry{}, + pendingProjection: map[string]pendingProjectionEntry{}, + baselineCache: map[string]sourceBaselineCacheEntry{}, + mileageKeysByPrefix: map[string]map[string]struct{}{}, + projectionKeysByPrefix: map[string]map[string]struct{}{}, + baselineKeysByPrefix: map[string]map[string]struct{}{}, + lastGPSAccumulation: map[string]time.Time{}, + gpsAccumulationInterval: defaultGPSAccumulationInterval, } if query, ok := exec.(Queryer); ok { writer.query = query @@ -224,6 +233,15 @@ func (w *Writer) SetMaxCacheEntries(maxEntries int) { w.enforceCacheLimitsLocked() } +func (w *Writer) SetGPSAccumulationInterval(interval time.Duration) { + w.mu.Lock() + defer w.mu.Unlock() + if interval < 0 { + interval = 0 + } + w.gpsAccumulationInterval = interval +} + func (w *Writer) CacheStats() CacheStats { w.mu.Lock() defer w.mu.Unlock() @@ -233,16 +251,19 @@ func (w *Writer) CacheStats() CacheStats { LastProjectionEntries: len(w.lastProjection), PendingProjectionEntries: len(w.pendingProjection), BaselineEntries: len(w.baselineCache), + GPSAccumulationEntries: len(w.lastGPSAccumulation), MaxEntries: w.maxCacheEntries, LastCleanupAt: w.lastCacheCleanup, LastCleanupTotalMileage: w.lastCacheCleanupStats.totalMileage, LastCleanupSourceSeen: w.lastCacheCleanupStats.sourceSeen, LastCleanupProjection: w.lastCacheCleanupStats.projection, LastCleanupBaseline: w.lastCacheCleanupStats.baseline, + LastCleanupGPS: w.lastCacheCleanupStats.gps, TotalMileageEvictions: w.cacheEvictions.totalMileage, SourceSeenEvictions: w.cacheEvictions.sourceSeen, ProjectionEvictions: w.cacheEvictions.projection, BaselineEvictions: w.cacheEvictions.baseline, + GPSEvictions: w.cacheEvictions.gps, } } @@ -306,11 +327,12 @@ func (w *Writer) AppendWithResult(ctx context.Context, env envelope.FrameEnvelop } var stationaryCandidate *SourceMileageSample if len(samples) == 0 && hasSource && result.SamplesSkippedMissingMileage > 0 { - if point, ok := GPSMileagePointFromEnvelope(env, identity, w.loc); ok { - _, recovered, recoverErr := AccumulateJT808GPSMileage(ctx, w.exec, point) + if point, ok := GPSMileagePointFromEnvelope(env, identity, w.loc); ok && w.shouldAccumulateGPS(point) { + _, recovered, recoverErr := AccumulateGPSMileage(ctx, w.exec, point) if recoverErr != nil { return result, recoverErr } + w.markGPSAccumulated(point) if recovered { result.SamplesFound++ result.SamplesWritten++ @@ -371,6 +393,31 @@ func (w *Writer) AppendWithResult(ctx context.Context, env envelope.FrameEnvelop return result, nil } +func (w *Writer) shouldAccumulateGPS(point GPSMileagePoint) bool { + key := gpsMileageStateCacheKey(point) + w.mu.Lock() + defer w.mu.Unlock() + last, ok := w.lastGPSAccumulation[key] + if !ok || w.gpsAccumulationInterval == 0 { + return true + } + return point.EventTime.After(last.Add(w.gpsAccumulationInterval)) +} + +func (w *Writer) markGPSAccumulated(point GPSMileagePoint) { + key := gpsMileageStateCacheKey(point) + w.mu.Lock() + if previous, ok := w.lastGPSAccumulation[key]; !ok || point.EventTime.After(previous) { + w.lastGPSAccumulation[key] = point.EventTime + } + w.enforceCacheLimitsLocked() + w.mu.Unlock() +} + +func gpsMileageStateCacheKey(point GPSMileagePoint) string { + return point.VIN + "|" + point.StatDate + "|" + string(point.Protocol) + "|" + point.SourceKey +} + func (w *Writer) stationaryCarryForward(ctx context.Context, env envelope.FrameEnvelope, identity SourceIdentity) (MetricSample, SourceMileageSample, bool, error) { if env.Protocol != envelope.ProtocolYutongMQTT || strings.TrimSpace(env.VIN) == "" { return MetricSample{}, SourceMileageSample{}, false, nil @@ -739,6 +786,12 @@ func (w *Writer) maybeCleanupCaches(now time.Time) { cleanupStats.baseline++ } } + for key, eventTime := range w.lastGPSAccumulation { + if eventTime.IsZero() || eventTime.Before(cutoff) { + delete(w.lastGPSAccumulation, key) + cleanupStats.gps++ + } + } w.lastCacheCleanupStats = cleanupStats w.enforceCacheLimitsLocked() } @@ -966,6 +1019,14 @@ func (w *Writer) enforceCacheLimitsLocked() { delete(w.lastSourceSeen, victim) w.cacheEvictions.sourceSeen++ } + for len(w.lastGPSAccumulation) > w.maxCacheEntries { + victim := oldestSourceSeenKey(w.lastGPSAccumulation) + if victim == "" { + break + } + delete(w.lastGPSAccumulation, victim) + w.cacheEvictions.gps++ + } } func mileageCachePrefix(sample MetricSample) string { diff --git a/go/vehicle-gateway/internal/stats/gps_mileage.go b/go/vehicle-gateway/internal/stats/gps_mileage.go index 1abec90f..bbf7ba0a 100644 --- a/go/vehicle-gateway/internal/stats/gps_mileage.go +++ b/go/vehicle-gateway/internal/stats/gps_mileage.go @@ -99,13 +99,20 @@ func CalculateGPSMileageSegment(previous, current GPSMileagePoint) GPSMileageSeg } func GPSMileagePointFromEnvelope(env envelope.FrameEnvelope, identity SourceIdentity, loc *time.Location) (GPSMileagePoint, bool) { - if env.Protocol != envelope.ProtocolJT808 || strings.TrimSpace(env.VIN) == "" { + if !supportsGPSMileageFallback(env.Protocol) || strings.TrimSpace(env.VIN) == "" { return GPSMileagePoint{}, false } location, ok := telemetry.LocationProjectionForProtocol(env.Protocol, env.Fields) if !ok || !validGPSCoordinate(location.Longitude, location.Latitude) { return GPSMileagePoint{}, false } + // JT808 terminals often omit speed together with odometer data, so keep the + // established coordinate-only fallback. GB32960 and Yutong report much more + // frequently; require positive device speed to prevent stationary GPS jitter + // from becoming mileage and to keep the MySQL state path inexpensive. + if env.Protocol != envelope.ProtocolJT808 && (location.SpeedKMH == nil || *location.SpeedKMH <= 0) { + return GPSMileagePoint{}, false + } eventMS, _, ok := envelope.NormalizedEventTimeMSWithReason(env) if !ok { return GPSMileagePoint{}, false @@ -131,7 +138,16 @@ func GPSMileagePointFromEnvelope(env envelope.FrameEnvelope, identity SourceIden }, true } -func AccumulateJT808GPSMileage(ctx context.Context, exec Execer, point GPSMileagePoint) (GPSMileageState, bool, error) { +func supportsGPSMileageFallback(protocol envelope.Protocol) bool { + switch protocol { + case envelope.ProtocolGB32960, envelope.ProtocolJT808, envelope.ProtocolYutongMQTT: + return true + default: + return false + } +} + +func AccumulateGPSMileage(ctx context.Context, exec Execer, point GPSMileagePoint) (GPSMileageState, bool, error) { beginner, ok := exec.(txBeginner) if !ok || strings.TrimSpace(point.VIN) == "" || strings.TrimSpace(point.SourceKey) == "" || point.EventTime.IsZero() { return GPSMileageState{}, false, nil @@ -242,6 +258,12 @@ func AccumulateJT808GPSMileage(ctx context.Context, exec Execer, point GPSMileag return state, true, nil } +// AccumulateJT808GPSMileage remains as a compatibility wrapper for callers and +// operational tooling compiled against the original JT808-only fallback. +func AccumulateJT808GPSMileage(ctx context.Context, exec Execer, point GPSMileagePoint) (GPSMileageState, bool, error) { + return AccumulateGPSMileage(ctx, exec, point) +} + func selectExistingGPSMileageBaseline(ctx context.Context, tx *sql.Tx, point GPSMileagePoint) (float64, time.Time, int64, int64, error) { var mileage sql.NullFloat64 var firstEvent sql.NullTime @@ -293,7 +315,7 @@ func upsertGPSMileageCandidate(ctx context.Context, exec Execer, point GPSMileag point.SourceEndpoint, point.Phone, point.DeviceID, - "JT808 GPS轨迹估算", + gpsMileagePlatformName(point.Protocol), 0, state.DailyMileageKM, state.DailyMileageKM, @@ -306,6 +328,17 @@ func upsertGPSMileageCandidate(ctx context.Context, exec Execer, point GPSMileag return err } +func gpsMileagePlatformName(protocol envelope.Protocol) string { + switch protocol { + case envelope.ProtocolGB32960: + return "GB32960 GPS轨迹估算" + case envelope.ProtocolYutongMQTT: + return "宇通 GPS轨迹估算" + default: + return "JT808 GPS轨迹估算" + } +} + func gpsMileageCandidateSourceKey(point GPSMileagePoint) string { identity := strings.TrimSpace(point.Phone) if identity == "" { diff --git a/go/vehicle-gateway/internal/stats/gps_mileage_test.go b/go/vehicle-gateway/internal/stats/gps_mileage_test.go index f861ead0..f16af76d 100644 --- a/go/vehicle-gateway/internal/stats/gps_mileage_test.go +++ b/go/vehicle-gateway/internal/stats/gps_mileage_test.go @@ -87,6 +87,70 @@ func TestGPSMileagePointFromEnvelopeUsesNaturalDayAndStableSource(t *testing.T) } } +func TestGPSMileagePointFromYutongEnvelopeRequiresMovement(t *testing.T) { + loc := time.FixedZone("Asia/Shanghai", 8*3600) + eventTime := time.Date(2026, 7, 19, 8, 0, 0, 0, loc) + env := envelope.FrameEnvelope{ + EventID: "evt-yutong-gps", + Protocol: envelope.ProtocolYutongMQTT, + VIN: "LMRKH9AC9R1004099", + DeviceID: "LMRKH9AC9R1004099", + SourceEndpoint: "mqtt://yutong/ytforward/shln/2", + EventTimeMS: eventTime.UnixMilli(), + Fields: map[string]any{ + "yutong_mqtt.data.longitude": 121.4737, + "yutong_mqtt.data.latitude": 31.2304, + "yutong_mqtt.data.meter_speed": 18, + }, + } + identity := SourceIdentity{ + Protocol: envelope.ProtocolYutongMQTT, + SourceIP: "mqtt", + SourceCode: "yutong", + SourceKind: "PLATFORM", + PlatformName: "宇通", + } + point, ok := GPSMileagePointFromEnvelope(env, identity, loc) + if !ok { + t.Fatal("moving Yutong location must be eligible for GPS mileage fallback") + } + if point.Protocol != envelope.ProtocolYutongMQTT || point.SourceKey != "YUTONG_MQTT:LMRKH9AC9R1004099@PLATFORM:yutong" { + t.Fatalf("point = %+v", point) + } + env.Fields["yutong_mqtt.data.meter_speed"] = 0 + if _, ok := GPSMileagePointFromEnvelope(env, identity, loc); ok { + t.Fatal("stationary Yutong jitter must not enter GPS mileage fallback") + } +} + +func TestWriterThrottlesHighFrequencyGPSMileageStateWrites(t *testing.T) { + writer := NewWriter(&recordingExec{}, time.FixedZone("Asia/Shanghai", 8*3600)) + writer.SetGPSAccumulationInterval(10 * time.Second) + start := time.Date(2026, 7, 19, 8, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)) + point := GPSMileagePoint{ + VIN: "LMRKH9AC9R1004099", + StatDate: "2026-07-19", + Protocol: envelope.ProtocolYutongMQTT, + SourceKey: "YUTONG_MQTT:LMRKH9AC9R1004099@PLATFORM:yutong", + EventTime: start, + } + if !writer.shouldAccumulateGPS(point) { + t.Fatal("first point must be accepted") + } + writer.markGPSAccumulated(point) + point.EventTime = start.Add(5 * time.Second) + if writer.shouldAccumulateGPS(point) { + t.Fatal("high-frequency point inside the 10s interval must be throttled") + } + point.EventTime = start.Add(11 * time.Second) + if !writer.shouldAccumulateGPS(point) { + t.Fatal("point after the 10s interval must be accepted") + } + if got := writer.CacheStats().GPSAccumulationEntries; got != 1 { + t.Fatalf("GPS cache entries = %d, want 1", got) + } +} + func TestAccumulateJT808GPSMileagePersistsEstimateAndProjects(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { diff --git a/go/vehicle-gateway/internal/stats/source_mileage.go b/go/vehicle-gateway/internal/stats/source_mileage.go index e973f585..3e5f6725 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage.go +++ b/go/vehicle-gateway/internal/stats/source_mileage.go @@ -638,6 +638,28 @@ const projectSourceIdentitySampleCountSQL = `( = COALESCE(NULLIF(TRIM(s.phone), ''), NULLIF(TRIM(s.device_id), ''), s.source_key) )` +// A fresh cumulative odometer is authoritative. When an upstream odometer is +// stale at zero but device-speed-backed coordinates prove movement, the +// coordinate estimate must beat the synthetic zero-day carry-forward. A true +// positive odometer still always outranks the estimate. +const projectMileageEvidenceRankSQL = `CASE + WHEN COALESCE(s.daily_mileage_km, 0) > 0 + AND COALESCE(s.quality_reason, '') <> '` + QualityReasonGPSCoordinate + `' THEN 0 + WHEN COALESCE(s.daily_mileage_km, 0) > 0 + AND COALESCE(s.quality_reason, '') = '` + QualityReasonGPSCoordinate + `' THEN 1 + WHEN COALESCE(s.quality_reason, '') <> '` + QualityReasonGPSCoordinate + `' THEN 2 + ELSE 3 +END` + +const selectedMileageEvidenceRankSQL = `CASE + WHEN COALESCE(s2.daily_mileage_km, 0) > 0 + AND COALESCE(s2.quality_reason, '') <> '` + QualityReasonGPSCoordinate + `' THEN 0 + WHEN COALESCE(s2.daily_mileage_km, 0) > 0 + AND COALESCE(s2.quality_reason, '') = '` + QualityReasonGPSCoordinate + `' THEN 1 + WHEN COALESCE(s2.quality_reason, '') <> '` + QualityReasonGPSCoordinate + `' THEN 2 + ELSE 3 +END` + const selectedSourceIdentitySampleCountSQL = `( SELECT COALESCE(SUM(identity_source.sample_count), 0) FROM vehicle_daily_mileage_source identity_source @@ -672,7 +694,7 @@ WHERE s.vin = ? AND (ds.source_code IS NULL OR TRIM(ds.source_code) = '') AND (ds.platform_name IS NULL OR TRIM(ds.platform_name) = '') )) -ORDER BY CASE WHEN COALESCE(s.quality_reason, '') = '` + QualityReasonGPSCoordinate + `' THEN 1 ELSE 0 END, +ORDER BY ` + projectMileageEvidenceRankSQL + `, CASE COALESCE(NULLIF(NULLIF(TRIM(ds.source_kind), ''), 'UNKNOWN'), CASE WHEN s.source_key LIKE '%` + platformSourceKeyPrefix + `%' THEN 'PLATFORM' WHEN s.source_key LIKE '%` + directSourceKeySuffix + `' THEN 'DIRECT' ELSE 'UNKNOWN' END) WHEN 'PLATFORM' THEN 0 WHEN 'DIRECT' THEN 1 @@ -713,7 +735,7 @@ LEFT JOIN ( AND (ds.source_code IS NULL OR TRIM(ds.source_code) = '') AND (ds.platform_name IS NULL OR TRIM(ds.platform_name) = '') )) - ORDER BY CASE WHEN COALESCE(s2.quality_reason, '') = '` + QualityReasonGPSCoordinate + `' THEN 1 ELSE 0 END, + ORDER BY ` + selectedMileageEvidenceRankSQL + `, CASE COALESCE(NULLIF(NULLIF(TRIM(ds.source_kind), ''), 'UNKNOWN'), CASE WHEN s2.source_key LIKE '%` + platformSourceKeyPrefix + `%' THEN 'PLATFORM' WHEN s2.source_key LIKE '%` + directSourceKeySuffix + `' THEN 'DIRECT' ELSE 'UNKNOWN' END) WHEN 'PLATFORM' THEN 0 WHEN 'DIRECT' THEN 1 diff --git a/go/vehicle-gateway/internal/stats/source_mileage_test.go b/go/vehicle-gateway/internal/stats/source_mileage_test.go index bfc5c96b..d1d7ffc6 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage_test.go +++ b/go/vehicle-gateway/internal/stats/source_mileage_test.go @@ -505,7 +505,10 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) { "FROM vehicle_daily_mileage_source s", "LEFT JOIN vehicle_data_source ds", "ds.id", - "CASE WHEN COALESCE(s.quality_reason, '') = '" + QualityReasonGPSCoordinate + "' THEN 1 ELSE 0 END", + "WHEN COALESCE(s.daily_mileage_km, 0) > 0", + "AND COALESCE(s.quality_reason, '') <> '" + QualityReasonGPSCoordinate + "' THEN 0", + "AND COALESCE(s.quality_reason, '') = '" + QualityReasonGPSCoordinate + "' THEN 1", + "WHEN COALESCE(s.quality_reason, '') <> '" + QualityReasonGPSCoordinate + "' THEN 2", "CASE COALESCE(NULLIF(NULLIF(TRIM(ds.source_kind), ''), 'UNKNOWN'), CASE WHEN s.source_key LIKE '%@PLATFORM:%' THEN 'PLATFORM' WHEN s.source_key LIKE '%@DIRECT' THEN 'DIRECT' ELSE 'UNKNOWN' END)", "WHEN 'PLATFORM' THEN 0", "WHEN 'DIRECT' THEN 1", @@ -554,7 +557,10 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) { "COALESCE(NULLIF(TRIM(ds.source_kind), ''), 'UNKNOWN') = 'UNKNOWN'", "AND (ds.source_code IS NULL OR TRIM(ds.source_code) = '')", "AND (ds.platform_name IS NULL OR TRIM(ds.platform_name) = '')", - "CASE WHEN COALESCE(s2.quality_reason, '') = '" + QualityReasonGPSCoordinate + "' THEN 1 ELSE 0 END", + "WHEN COALESCE(s2.daily_mileage_km, 0) > 0", + "AND COALESCE(s2.quality_reason, '') <> '" + QualityReasonGPSCoordinate + "' THEN 0", + "AND COALESCE(s2.quality_reason, '') = '" + QualityReasonGPSCoordinate + "' THEN 1", + "WHEN COALESCE(s2.quality_reason, '') <> '" + QualityReasonGPSCoordinate + "' THEN 2", "CASE COALESCE(NULLIF(NULLIF(TRIM(ds.source_kind), ''), 'UNKNOWN'), CASE WHEN s2.source_key LIKE '%@PLATFORM:%' THEN 'PLATFORM' WHEN s2.source_key LIKE '%@DIRECT' THEN 'DIRECT' ELSE 'UNKNOWN' END)", "WHEN 'PLATFORM' THEN 0", "WHEN 'DIRECT' THEN 1", diff --git a/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go b/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go index a96d060f..f4cd978e 100644 --- a/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go +++ b/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go @@ -449,6 +449,7 @@ func buildVehicleRealtimeSQL(query url.Values) SQLQuery { `WHERE stat_date = CURDATE() GROUP BY vin, protocol` + `) m ON m.vin = v.vin AND m.protocol = l.protocol ` + groupSuffixSQL + todayMileageOrderExpr := `CASE WHEN COALESCE(m.daily_mileage_km, 0) > 0 THEN 0 ELSE 1 END, ` + orderExpr // Keep one authoritative coordinate source while all protocols are healthy. // Ordering only by updated_at makes the selected point flap whenever protocols // report at different cadences. When every source is stale, recency remains the @@ -476,8 +477,8 @@ func buildVehicleRealtimeSQL(query url.Values) SQLQuery { `COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(l.soc_percent AS CHAR) ORDER BY ` + orderExpr + `), ',', 1), '') AS soc_percent, ` + `COALESCE(CAST(SUBSTRING_INDEX(GROUP_CONCAT(CASE WHEN l.total_mileage_km IS NOT NULL THEN 1 ELSE 0 END ORDER BY ` + orderExpr + `), ',', 1) AS UNSIGNED), 0) AS mileage_available, ` + `COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(l.total_mileage_km AS CHAR) ORDER BY ` + orderExpr + `), ',', 1), '') AS total_mileage_km, ` + - `COALESCE(CAST(SUBSTRING_INDEX(GROUP_CONCAT(CASE WHEN m.daily_mileage_km IS NOT NULL THEN 1 ELSE 0 END ORDER BY ` + orderExpr + `), ',', 1) AS UNSIGNED), 0) AS today_mileage_available, ` + - `COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(m.daily_mileage_km AS CHAR) ORDER BY ` + orderExpr + `), ',', 1), '') AS today_mileage_km, ` + + `COALESCE(CAST(SUBSTRING_INDEX(GROUP_CONCAT(CASE WHEN m.daily_mileage_km IS NOT NULL THEN 1 ELSE 0 END ORDER BY ` + todayMileageOrderExpr + `), ',', 1) AS UNSIGNED), 0) AS today_mileage_available, ` + + `COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(m.daily_mileage_km AS CHAR) ORDER BY ` + todayMileageOrderExpr + `), ',', 1), '') AS today_mileage_km, ` + `COALESCE(DATE_FORMAT(MAX(l.updated_at), '%Y-%m-%d %H:%i:%s'), '') AS last_seen, ` + `CAST(SUBSTRING_INDEX(GROUP_CONCAT(COALESCE(s.access_report_interval_ms, -1) ORDER BY ` + orderExpr + `), ',', 1) AS SIGNED) AS report_interval_ms ` + groupSQL + diff --git a/vehicle-data-platform/apps/api/internal/platform/query_builders_test.go b/vehicle-data-platform/apps/api/internal/platform/query_builders_test.go index da3c226e..7d3e19d5 100644 --- a/vehicle-data-platform/apps/api/internal/platform/query_builders_test.go +++ b/vehicle-data-platform/apps/api/internal/platform/query_builders_test.go @@ -275,11 +275,16 @@ func TestBuildVehicleRealtimeSQL(t *testing.T) { "location_source", "location_conflict", "location_conflict_distance_m", + "GROUP_CONCAT(CAST(m.daily_mileage_km AS CHAR) ORDER BY CASE WHEN COALESCE(m.daily_mileage_km, 0) > 0 THEN 0 ELSE 1 END", } { if !strings.Contains(built.Text, want) { t.Fatalf("realtime source arbitration missing %q: %s", want, built.Text) } } + mileageOrder := strings.Index(built.Text, "CASE WHEN COALESCE(m.daily_mileage_km, 0) > 0 THEN 0 ELSE 1 END") + if mileageOrder < 0 || strings.Index(built.Text[mileageOrder:], "WHEN 'GB32960' THEN 10") < 0 { + t.Fatalf("today mileage must fall through a zero primary-location source before protocol location priority: %s", built.Text) + } if len(built.Args) != 8 || built.Args[0] != "JT808" || built.Args[1] != "%粤A%" || built.Args[6] != 10 || built.Args[7] != 20 { t.Fatalf("args = %#v", built.Args) }