diff --git a/docs/architecture/storage-minimal-contract.md b/docs/architecture/storage-minimal-contract.md index fdb1939f..4e5042e1 100644 --- a/docs/architecture/storage-minimal-contract.md +++ b/docs/architecture/storage-minimal-contract.md @@ -24,6 +24,7 @@ | MySQL | `vehicle_realtime_snapshot` | 每个协议+VIN 的最新事件态、合并扁平字段，以及独立的首次/前次/最新接收时间、连续间隔、样本数和证据来源 | phone、device、原始嵌套解析树、raw 报文、无限接收历史 | | MySQL | `vehicle_realtime_location` | 每个协议+VIN 的最新位置核心字段 | raw、parsed JSON、消息头内部字段 | | MySQL | `vehicle_daily_mileage` | 日期、VIN、协议、日里程、首末总里程、样本数 | 临时 vehicle key、泛化 metric key/value、自增 id、created_at、每帧细节、位置点列表 | +| MySQL | `vehicle_daily_gps_mileage_state` | 仅 JT808 缺累计里程时的单车/自然日/物理来源滚动坐标、累计距离和质量计数 | 轨迹点列表、RAW 报文、跨日状态、GB32960/宇通状态 | | MySQL | `vehicle_identity_binding` | 人工维护或导入的 VIN、车牌、phone、oem 映射 | 注册历史、协议状态、device_id、自增 id、created_at | | MySQL | `jt808_registration` | JT808 phone 主键下的注册、鉴权、VIN 匹配状态、来源端点 | GB32960/MQTT 注册信息、位置历史、created_at | | Redis | `vehicle:realtime-raw:{protocol}:{vin}` | 每协议每 VIN 最新完整 parsed 状态 | 无 VIN 临时身份、历史数据、统计结果 | diff --git a/docs/ops/vehicle-ingest-runbook.md b/docs/ops/vehicle-ingest-runbook.md index affc491c..5e5e064e 100644 --- a/docs/ops/vehicle-ingest-runbook.md +++ b/docs/ops/vehicle-ingest-runbook.md @@ -511,6 +511,8 @@ 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` 禁用该降级路径。 + 生产定时补算使用 systemd timer，默认每天 01:30 补算昨天往前 3 天，覆盖上游晚到或断传后恢复的总里程字段： ```bash diff --git a/go/vehicle-gateway/cmd/stat-writer/main.go b/go/vehicle-gateway/cmd/stat-writer/main.go index a8775d73..20cbc450 100644 --- a/go/vehicle-gateway/cmd/stat-writer/main.go +++ b/go/vehicle-gateway/cmd/stat-writer/main.go @@ -783,6 +783,7 @@ func recordStatSampleMetrics(registry *metrics.Registry, message kafka.Message, addStatSampleMetric(registry, message, protocol, "skipped_missing_vin", result.SamplesSkippedMissingVIN) addStatSampleMetric(registry, message, protocol, "skipped_missing_mileage", result.SamplesSkippedMissingMileage) addStatSampleMetric(registry, message, protocol, "recovered_stationary", result.SamplesRecoveredStationary) + addStatSampleMetric(registry, message, protocol, "recovered_gps_coordinate", result.SamplesRecoveredGPSCoordinate) addStatSampleMetric(registry, message, protocol, "skipped_non_mileage_frame", result.SamplesSkippedNonMileageFrame) addStatSampleMetric(registry, message, protocol, "skipped_non_positive_mileage", result.SamplesSkippedNonPositiveMileage) addStatSampleMetric(registry, message, protocol, "skipped_missing_time", result.SamplesSkippedMissingTime) diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go index a394cf27..899d9f36 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main.go @@ -36,6 +36,7 @@ type config struct { Debug bool EventTimeFullScan bool StationaryCarry bool + JT808GPSFallback bool ProgressEvery int64 BaselineLookback int Location *time.Location @@ -150,6 +151,10 @@ func main() { if err != nil { fail("build realtime-location fallback aggregates", err) } + gpsFallbacks, err := addJT808GPSCoordinateFallbackAggregates(ctx, mysqlDB, td, cfg, aggregates) + if err != nil { + fail("build JT808 GPS-coordinate fallback aggregates", err) + } var written int64 var normalized int if !cfg.DryRun { @@ -162,7 +167,7 @@ func main() { fail("normalize historical platform sources", err) } } - slog.Info("stats backfill complete", "method", cfg.Method, "dry_run", cfg.DryRun, "aggregates", len(aggregates), "realtimeLocationFallbacks", fallbacks, "written", written, "platformSourcesNormalized", normalized) + slog.Info("stats backfill complete", "method", cfg.Method, "dry_run", cfg.DryRun, "aggregates", len(aggregates), "realtimeLocationFallbacks", fallbacks, "jt808GPSFallbacks", gpsFallbacks, "written", written, "platformSourcesNormalized", normalized) return } @@ -322,8 +327,10 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met } clearedTargets[target] = struct{}{} } - if err := stats.UpsertDataSource(ctx, db, identity, agg.LatestEventTime); err != nil { - return written, err + if agg.QualityReason != stats.QualityReasonGPSCoordinate { + if err := stats.UpsertDataSource(ctx, db, identity, agg.LatestEventTime); err != nil { + return written, err + } } dailyKM := stats.DailyMileageFromDayBoundary(agg.FirstKM, agg.LatestKM) candidate := stats.SourceMileageSample{ @@ -498,6 +505,174 @@ func addRealtimeLocationFallbackAggregates(ctx context.Context, mysqlDB *sql.DB, return added, nil } +type gpsCoordinateBackfillState struct { + previous stats.GPSMileagePoint + firstEventTime time.Time + latestEventTime time.Time + distanceKM float64 + pointCount int64 + usableSegments int64 + badJumps int64 + 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) { + return 0, nil + } + existing, err := existingJT808OdometerTargets(ctx, mysqlDB, cfg) + if err != nil { + return 0, err + } + for _, agg := range aggregates { + if agg != nil && agg.Protocol == envelope.ProtocolJT808 && agg.QualityReason != stats.QualityReasonGPSCoordinate { + 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'", + "vin IS NOT NULL", + "vin <> ''", + "longitude BETWEEN -180 AND 180", + "latitude BETWEEN -90 AND 90", + "NOT (longitude = 0 AND latitude = 0)", + } + sqlText := fmt.Sprintf(`SELECT vin, ts, longitude, latitude +FROM %s.vehicle_locations +WHERE %s +ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) + rows, err := tdDB.QueryContext(ctx, sqlText) + if err != nil { + return 0, err + } + defer rows.Close() + states := map[string]*gpsCoordinateBackfillState{} + for rows.Next() { + var vin string + var eventTime time.Time + var longitude, latitude float64 + if err := rows.Scan(&vin, &eventTime, &longitude, &latitude); err != nil { + return 0, err + } + vin = strings.TrimSpace(vin) + eventTime = eventTime.In(cfg.Location) + date := eventTime.Format("2006-01-02") + target := vin + "|" + date + if _, ok := existing[target]; ok { + continue + } + current := stats.GPSMileagePoint{ + VIN: vin, + StatDate: date, + Protocol: envelope.ProtocolJT808, + EventTime: eventTime, + Longitude: longitude, + Latitude: latitude, + } + state := states[target] + if state == nil { + states[target] = &gpsCoordinateBackfillState{ + previous: current, + firstEventTime: eventTime, + latestEventTime: eventTime, + pointCount: 1, + } + continue + } + segment := stats.CalculateGPSMileageSegment(state.previous, current) + if segment.OutOfOrder { + continue + } + state.previous = current + state.latestEventTime = eventTime + state.pointCount++ + if segment.Usable { + state.distanceKM += segment.DistanceKM + state.usableSegments++ + } + if segment.BadJump { + state.badJumps++ + } + if segment.LongGap { + state.longGaps++ + } + } + if err := rows.Err(); err != nil { + return 0, err + } + added := 0 + for target, state := range states { + if state.usableSegments == 0 { + continue + } + parts := strings.SplitN(target, "|", 2) + if len(parts) != 2 { + continue + } + vin, date := parts[0], parts[1] + sourceKey := string(envelope.ProtocolJT808) + ":" + vin + "@GPS_COORDINATE" + key := vin + "|" + date + "|" + string(envelope.ProtocolJT808) + "|" + sourceKey + aggregates[key] = &metricAgg{ + VIN: vin, + Date: date, + Protocol: envelope.ProtocolJT808, + FirstKM: 0, + LatestKM: state.distanceKM, + Count: state.pointCount, + SourceKey: sourceKey, + SourceEndpoint: "gps-coordinate", + PlatformName: "JT808 GPS轨迹估算", + SourceKind: "UNKNOWN", + FirstEventTime: state.firstEventTime, + LatestEventTime: state.latestEventTime, + QualityStatus: stats.QualityOK, + QualityReason: stats.QualityReasonGPSCoordinate, + } + added++ + } + if len(states) > 0 { + slog.Info("JT808 GPS-coordinate fallbacks loaded", "states", len(states), "added", added) + } + return added, nil +} + +func existingJT808OdometerTargets(ctx context.Context, db *sql.DB, cfg config) (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' + 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) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var vin, date string + if err := rows.Scan(&vin, &date); err != nil { + return nil, err + } + result[strings.TrimSpace(vin)+"|"+strings.TrimSpace(date)] = struct{}{} + } + return result, rows.Err() +} + +func containsProtocol(protocols []envelope.Protocol, target envelope.Protocol) bool { + for _, protocol := range protocols { + if protocol == target { + return true + } + } + return false +} + func indexAggregateSourceRowsByDate(aggregates map[string]*metricAgg, protocol envelope.Protocol) map[string]map[string][]dailySourceLast { indexed := map[string]map[string][]dailySourceLast{} for _, agg := range aggregates { @@ -1009,6 +1184,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), 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 41ab785c..e9709585 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main_test.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main_test.go @@ -990,6 +990,62 @@ func TestResolveBackfillDateRangeDefaultsToToday(t *testing.T) { } } +func TestAddJT808GPSCoordinateFallbackAggregatesOnlyMissingOdometerVehicles(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("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) + start := time.Date(2026, 7, 19, 8, 0, 0, 0, loc) + tdMock.ExpectQuery("SELECT vin, ts, longitude, latitude"). + WillReturnRows(sqlmock.NewRows([]string{"vin", "ts", "longitude", "latitude"}). + AddRow("VIN-GPS-FALLBACK", start, 121.4737, 31.2304). + AddRow("VIN-WITH-ODOMETER", start, 121.4737, 31.2304). + AddRow("VIN-GPS-FALLBACK", start.Add(5*time.Minute), 121.4837, 31.2304). + 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{ + DateFrom: "2026-07-19", + DateTo: "2026-07-19", + Protocols: []envelope.Protocol{envelope.ProtocolJT808}, + JT808GPSFallback: true, + Location: loc, + TDengineDatabase: "vehicle_ts", + }, aggregates) + if err != nil { + t.Fatalf("addJT808GPSCoordinateFallbackAggregates() error = %v", err) + } + if added != 1 || len(aggregates) != 1 { + t.Fatalf("added=%d aggregates=%d", added, len(aggregates)) + } + for _, aggregate := range aggregates { + if aggregate.VIN != "VIN-GPS-FALLBACK" || aggregate.QualityReason != stats.QualityReasonGPSCoordinate { + t.Fatalf("aggregate = %+v", aggregate) + } + if aggregate.LatestKM < 0.9 || aggregate.LatestKM > 1.1 || aggregate.Count != 2 { + t.Fatalf("GPS aggregate distance/count = %.3f/%d", aggregate.LatestKM, aggregate.Count) + } + } + 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 6f7f6b5e..1a237238 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -73,6 +73,7 @@ type AppendResult struct { SamplesSkippedMissingVIN int SamplesSkippedMissingMileage int SamplesRecoveredStationary int + SamplesRecoveredGPSCoordinate int SamplesSkippedNonMileageFrame int SamplesSkippedNonPositiveMileage int SamplesSkippedMissingTime int @@ -248,6 +249,7 @@ func (w *Writer) CacheStats() CacheStats { func (w *Writer) EnsureSchema(ctx context.Context) error { for _, statement := range []string{ DataSourceTableSQL, + GPSMileageStateTableSQL, DailyMileageSourceTableSQL, DailyMileageTableSQL, } { @@ -304,6 +306,20 @@ 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 recoverErr != nil { + return result, recoverErr + } + if recovered { + result.SamplesFound++ + result.SamplesWritten++ + result.SamplesRecoveredGPSCoordinate++ + result.ProjectionsAttempted++ + result.ProjectionsWritten++ + return result, nil + } + } sample, candidate, recovered, recoverErr := w.stationaryCarryForward(ctx, env, identity) if recoverErr != nil { return result, recoverErr diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 29d27d6e..efff4092 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -1836,12 +1836,13 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { t.Fatalf("Append() error = %v", err) } - schemaCalls := 3 + len(DailyMileageAlterSQL) + schemaCalls := 4 + len(DailyMileageAlterSQL) if len(exec.calls) != schemaCalls+5 { t.Fatalf("exec calls = %d", len(exec.calls)) } for i, want := range []string{ "CREATE TABLE IF NOT EXISTS vehicle_data_source", + "CREATE TABLE IF NOT EXISTS vehicle_daily_gps_mileage_state", "CREATE TABLE IF NOT EXISTS vehicle_daily_mileage_source", "CREATE TABLE IF NOT EXISTS vehicle_daily_mileage", } { @@ -1850,20 +1851,20 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { } } for _, column := range []string{"vehicle_key", "AUTO_INCREMENT", "created_at", "first_total_mileage_km", "trusted_source_key", "trusted_phone", "trusted_source_endpoint", "sample_count"} { - if strings.Contains(exec.calls[2].query, column) { - t.Fatalf("schema should not include %s: %s", column, exec.calls[2].query) + if strings.Contains(exec.calls[3].query, column) { + t.Fatalf("schema should not include %s: %s", column, exec.calls[3].query) } } for _, column := range []string{"source_id BIGINT NULL", "daily_mileage_km", "latest_total_mileage_km", "KEY idx_source_id (source_id)"} { - if !strings.Contains(exec.calls[2].query, column) { - t.Fatalf("schema should include %s: %s", column, exec.calls[2].query) + if !strings.Contains(exec.calls[3].query, column) { + t.Fatalf("schema should include %s: %s", column, exec.calls[3].query) } } - if !strings.Contains(exec.calls[2].query, "PRIMARY KEY (vin, stat_date, protocol)") { - t.Fatalf("daily mileage table should key by vin/stat_date/protocol: %s", exec.calls[2].query) + if !strings.Contains(exec.calls[3].query, "PRIMARY KEY (vin, stat_date, protocol)") { + t.Fatalf("daily mileage table should key by vin/stat_date/protocol: %s", exec.calls[3].query) } - if strings.Contains(exec.calls[2].query, "KEY idx_vin (vin)") { - t.Fatalf("daily mileage table should not keep redundant vin index covered by the primary key: %s", exec.calls[2].query) + if strings.Contains(exec.calls[3].query, "KEY idx_vin (vin)") { + t.Fatalf("daily mileage table should not keep redundant vin index covered by the primary key: %s", exec.calls[3].query) } projectCall := exec.calls[schemaCalls+2] if !strings.Contains(projectCall.query, "INSERT INTO vehicle_daily_mileage") { diff --git a/go/vehicle-gateway/internal/stats/gps_mileage.go b/go/vehicle-gateway/internal/stats/gps_mileage.go new file mode 100644 index 00000000..1abec90f --- /dev/null +++ b/go/vehicle-gateway/internal/stats/gps_mileage.go @@ -0,0 +1,398 @@ +package stats + +import ( + "context" + "database/sql" + "errors" + "math" + "strings" + "time" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/telemetry" +) + +const ( + QualityReasonGPSCoordinate = "gps_coordinate_accumulation" + gpsCoordinateSourceSuffix = "#GPS_COORDINATE" + gpsMaxSegmentGap = 10 * time.Minute + gpsMaxImpliedSpeedKMH = 220.0 +) + +const GPSMileageStateTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_gps_mileage_state ( + vin VARCHAR(32) NOT NULL, + stat_date DATE NOT NULL, + protocol VARCHAR(32) NOT NULL, + source_key VARCHAR(256) NOT NULL, + source_ip VARCHAR(64) NOT NULL, + source_endpoint VARCHAR(128) NULL, + phone VARCHAR(32) NULL, + device_id VARCHAR(64) NULL, + first_event_time DATETIME(3) NOT NULL, + latest_event_time DATETIME(3) NOT NULL, + latest_event_id VARCHAR(128) NULL, + latest_longitude DOUBLE NOT NULL, + latest_latitude DOUBLE NOT NULL, + daily_mileage_km DECIMAL(18,6) NOT NULL DEFAULT 0, + point_count BIGINT NOT NULL DEFAULT 1, + usable_segment_count BIGINT NOT NULL DEFAULT 0, + bad_jump_count BIGINT NOT NULL DEFAULT 0, + long_gap_count BIGINT NOT NULL DEFAULT 0, + out_of_order_count BIGINT NOT NULL DEFAULT 0, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (vin, stat_date, protocol, source_key), + KEY idx_gps_mileage_date (stat_date, protocol, vin) +)` + +type GPSMileagePoint struct { + VIN string + StatDate string + Protocol envelope.Protocol + SourceKey string + SourceIP string + SourceEndpoint string + Phone string + DeviceID string + EventID string + EventTime time.Time + Longitude float64 + Latitude float64 +} + +type GPSMileageState struct { + FirstEventTime time.Time + LatestEventTime time.Time + LatestEventID string + LatestLongitude float64 + LatestLatitude float64 + DailyMileageKM float64 + PointCount int64 + UsableSegmentCount int64 + BadJumpCount int64 + LongGapCount int64 + OutOfOrderCount int64 +} + +type GPSMileageSegment struct { + DistanceKM float64 + Usable bool + LongGap bool + BadJump bool + OutOfOrder bool +} + +func CalculateGPSMileageSegment(previous, current GPSMileagePoint) GPSMileageSegment { + if current.EventTime.IsZero() || previous.EventTime.IsZero() || !current.EventTime.After(previous.EventTime) { + return GPSMileageSegment{OutOfOrder: true} + } + elapsed := current.EventTime.Sub(previous.EventTime) + if elapsed > gpsMaxSegmentGap { + return GPSMileageSegment{LongGap: true} + } + distanceKM := haversineDistanceKM(previous.Latitude, previous.Longitude, current.Latitude, current.Longitude) + impliedSpeed := distanceKM / elapsed.Hours() + if !isFiniteNonNegative(distanceKM) || !isFiniteNonNegative(impliedSpeed) || impliedSpeed > gpsMaxImpliedSpeedKMH { + return GPSMileageSegment{BadJump: true} + } + return GPSMileageSegment{DistanceKM: distanceKM, Usable: true} +} + +func GPSMileagePointFromEnvelope(env envelope.FrameEnvelope, identity SourceIdentity, loc *time.Location) (GPSMileagePoint, bool) { + if env.Protocol != envelope.ProtocolJT808 || 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 + } + eventMS, _, ok := envelope.NormalizedEventTimeMSWithReason(env) + if !ok { + return GPSMileagePoint{}, false + } + if loc == nil { + loc = time.FixedZone("Asia/Shanghai", 8*3600) + } + eventTime := time.UnixMilli(eventMS).In(loc) + sourceKey := SourceKeyForSource(env.Protocol, env.Phone, env.DeviceID, identity.SourceIP, identity.SourceKind, identity.SourceCode) + return GPSMileagePoint{ + VIN: strings.TrimSpace(env.VIN), + StatDate: eventTime.Format("2006-01-02"), + Protocol: env.Protocol, + SourceKey: sourceKey, + SourceIP: strings.TrimSpace(identity.SourceIP), + SourceEndpoint: strings.TrimSpace(env.SourceEndpoint), + Phone: strings.TrimSpace(env.Phone), + DeviceID: strings.TrimSpace(env.DeviceID), + EventID: strings.TrimSpace(env.EventID), + EventTime: eventTime, + Longitude: location.Longitude, + Latitude: location.Latitude, + }, true +} + +func AccumulateJT808GPSMileage(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 + } + tx, err := beginner.BeginTx(ctx, nil) + if err != nil { + return GPSMileageState{}, false, err + } + state, found, err := selectGPSMileageState(ctx, tx, point) + if err != nil { + _ = tx.Rollback() + return GPSMileageState{}, false, err + } + if !found { + baselineMileage, baselineFirstEvent, baselinePointCount, baselineSegmentCount, err := selectExistingGPSMileageBaseline(ctx, tx, point) + if err != nil { + _ = tx.Rollback() + return GPSMileageState{}, false, err + } + state = GPSMileageState{ + FirstEventTime: point.EventTime, + LatestEventTime: point.EventTime, + LatestEventID: point.EventID, + LatestLongitude: point.Longitude, + LatestLatitude: point.Latitude, + DailyMileageKM: baselineMileage, + PointCount: baselinePointCount + 1, + UsableSegmentCount: baselineSegmentCount, + } + if !baselineFirstEvent.IsZero() && baselineFirstEvent.Before(point.EventTime) { + state.FirstEventTime = baselineFirstEvent + } + if _, err := tx.ExecContext(ctx, insertGPSMileageStateSQL, + point.VIN, point.StatDate, string(point.Protocol), point.SourceKey, + point.SourceIP, point.SourceEndpoint, point.Phone, point.DeviceID, + state.FirstEventTime, point.EventTime, point.EventID, point.Longitude, point.Latitude, + state.DailyMileageKM, state.PointCount, state.UsableSegmentCount, + ); err != nil { + _ = tx.Rollback() + return GPSMileageState{}, false, err + } + if err := tx.Commit(); err != nil { + return GPSMileageState{}, false, err + } + return state, false, nil + } + + segment := CalculateGPSMileageSegment(GPSMileagePoint{ + EventTime: state.LatestEventTime, + Longitude: state.LatestLongitude, + Latitude: state.LatestLatitude, + }, point) + if segment.OutOfOrder { + state.OutOfOrderCount++ + if _, err := tx.ExecContext(ctx, incrementGPSMileageOutOfOrderSQL, + point.VIN, point.StatDate, string(point.Protocol), point.SourceKey, + ); err != nil { + _ = tx.Rollback() + return GPSMileageState{}, false, err + } + if err := tx.Commit(); err != nil { + return GPSMileageState{}, false, err + } + return state, false, nil + } + state.LatestEventTime = point.EventTime + state.LatestEventID = point.EventID + state.LatestLongitude = point.Longitude + state.LatestLatitude = point.Latitude + state.PointCount++ + if segment.Usable { + state.DailyMileageKM += segment.DistanceKM + state.UsableSegmentCount++ + } + if segment.BadJump { + state.BadJumpCount++ + } + if segment.LongGap { + state.LongGapCount++ + } + if _, err := tx.ExecContext(ctx, updateGPSMileageStateSQL, + point.SourceIP, point.SourceEndpoint, point.Phone, point.DeviceID, + state.LatestEventTime, state.LatestEventID, state.LatestLongitude, state.LatestLatitude, + state.DailyMileageKM, state.PointCount, state.UsableSegmentCount, + state.BadJumpCount, state.LongGapCount, state.OutOfOrderCount, + point.VIN, point.StatDate, string(point.Protocol), point.SourceKey, + ); err != nil { + _ = tx.Rollback() + return GPSMileageState{}, false, err + } + if state.UsableSegmentCount == 0 { + if err := tx.Commit(); err != nil { + return GPSMileageState{}, false, err + } + return state, false, nil + } + if err := upsertGPSMileageCandidate(ctx, tx, point, state); err != nil { + _ = tx.Rollback() + return GPSMileageState{}, false, err + } + if err := projectDailyMileageWithExec(ctx, tx, point.VIN, point.StatDate, point.Protocol); err != nil { + _ = tx.Rollback() + return GPSMileageState{}, false, err + } + if err := tx.Commit(); err != nil { + return GPSMileageState{}, false, err + } + return state, true, nil +} + +func selectExistingGPSMileageBaseline(ctx context.Context, tx *sql.Tx, point GPSMileagePoint) (float64, time.Time, int64, int64, error) { + var mileage sql.NullFloat64 + var firstEvent sql.NullTime + var pointCount sql.NullInt64 + err := tx.QueryRowContext(ctx, selectExistingGPSMileageBaselineSQL, + point.VIN, point.StatDate, string(point.Protocol), QualityReasonGPSCoordinate, + ).Scan(&mileage, &firstEvent, &pointCount) + if errors.Is(err, sql.ErrNoRows) { + return 0, time.Time{}, 0, 0, nil + } + if err != nil { + return 0, time.Time{}, 0, 0, err + } + segments := pointCount.Int64 - 1 + if segments < 0 { + segments = 0 + } + return mileage.Float64, firstEvent.Time, pointCount.Int64, segments, nil +} + +func selectGPSMileageState(ctx context.Context, tx *sql.Tx, point GPSMileagePoint) (GPSMileageState, bool, error) { + var state GPSMileageState + err := tx.QueryRowContext(ctx, selectGPSMileageStateSQL, + point.VIN, point.StatDate, string(point.Protocol), point.SourceKey, + ).Scan( + &state.FirstEventTime, &state.LatestEventTime, &state.LatestEventID, + &state.LatestLongitude, &state.LatestLatitude, &state.DailyMileageKM, + &state.PointCount, &state.UsableSegmentCount, &state.BadJumpCount, + &state.LongGapCount, &state.OutOfOrderCount, + ) + if errors.Is(err, sql.ErrNoRows) { + return GPSMileageState{}, false, nil + } + return state, err == nil, err +} + +func upsertGPSMileageCandidate(ctx context.Context, exec Execer, point GPSMileagePoint, state GPSMileageState) error { + sourceIP := strings.TrimSpace(point.SourceIP) + if len(sourceIP) > 56 { + sourceIP = sourceIP[:56] + } + sourceIP += "#gps" + _, err := exec.ExecContext(ctx, upsertGPSMileageCandidateSQL, + point.VIN, + point.StatDate, + string(point.Protocol), + gpsMileageCandidateSourceKey(point), + sourceIP, + point.SourceEndpoint, + point.Phone, + point.DeviceID, + "JT808 GPS轨迹估算", + 0, + state.DailyMileageKM, + state.DailyMileageKM, + state.PointCount, + state.FirstEventTime, + state.LatestEventTime, + QualityOK, + QualityReasonGPSCoordinate, + ) + return err +} + +func gpsMileageCandidateSourceKey(point GPSMileagePoint) string { + identity := strings.TrimSpace(point.Phone) + if identity == "" { + identity = strings.TrimSpace(point.DeviceID) + } + if identity == "" { + identity = strings.TrimSpace(point.VIN) + } + sourceIP := strings.TrimSpace(point.SourceIP) + if sourceIP != "" { + return string(point.Protocol) + ":" + identity + gpsCoordinateSourceSuffix + ":" + sourceIP + } + return string(point.Protocol) + ":" + identity + gpsCoordinateSourceSuffix +} + +func validGPSCoordinate(longitude, latitude float64) bool { + return longitude >= -180 && longitude <= 180 && + latitude >= -90 && latitude <= 90 && + (longitude != 0 || latitude != 0) && + !math.IsNaN(longitude) && !math.IsNaN(latitude) && + !math.IsInf(longitude, 0) && !math.IsInf(latitude, 0) +} + +func isFiniteNonNegative(value float64) bool { + return value >= 0 && !math.IsNaN(value) && !math.IsInf(value, 0) +} + +func haversineDistanceKM(lat1, lon1, lat2, lon2 float64) float64 { + const earthRadiusKM = 6371.0088 + toRadians := math.Pi / 180 + lat1Rad := lat1 * toRadians + lat2Rad := lat2 * toRadians + dLat := (lat2 - lat1) * toRadians + dLon := (lon2 - lon1) * toRadians + a := math.Sin(dLat/2)*math.Sin(dLat/2) + + math.Cos(lat1Rad)*math.Cos(lat2Rad)*math.Sin(dLon/2)*math.Sin(dLon/2) + return earthRadiusKM * 2 * math.Atan2(math.Sqrt(a), math.Sqrt(1-a)) +} + +const selectGPSMileageStateSQL = `SELECT first_event_time, latest_event_time, COALESCE(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 +FROM vehicle_daily_gps_mileage_state +WHERE vin = ? AND stat_date = ? AND protocol = ? AND source_key = ? +FOR UPDATE` + +const insertGPSMileageStateSQL = `INSERT INTO vehicle_daily_gps_mileage_state + (vin, stat_date, protocol, source_key, source_ip, source_endpoint, phone, device_id, + first_event_time, latest_event_time, latest_event_id, latest_longitude, latest_latitude, + daily_mileage_km, point_count, usable_segment_count) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + +const selectExistingGPSMileageBaselineSQL = `SELECT daily_mileage_km, first_event_time, sample_count +FROM vehicle_daily_mileage_source +WHERE vin = ? AND stat_date = ? AND protocol = ? AND quality_reason = ? +ORDER BY latest_event_time DESC, sample_count DESC +LIMIT 1` + +const incrementGPSMileageOutOfOrderSQL = `UPDATE vehicle_daily_gps_mileage_state +SET out_of_order_count = out_of_order_count + 1, updated_at = CURRENT_TIMESTAMP +WHERE vin = ? AND stat_date = ? AND protocol = ? AND source_key = ?` + +const updateGPSMileageStateSQL = `UPDATE vehicle_daily_gps_mileage_state +SET source_ip = ?, source_endpoint = ?, phone = ?, device_id = ?, + 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 = ?, + updated_at = CURRENT_TIMESTAMP +WHERE vin = ? AND stat_date = ? AND protocol = ? AND source_key = ?` + +const upsertGPSMileageCandidateSQL = `INSERT INTO vehicle_daily_mileage_source + (vin, stat_date, protocol, source_key, source_ip, source_endpoint, phone, device_id, platform_name, + first_total_mileage_km, latest_total_mileage_km, daily_mileage_km, sample_count, + first_event_time, latest_event_time, quality_status, quality_reason) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON DUPLICATE KEY UPDATE + source_ip = VALUES(source_ip), + source_endpoint = VALUES(source_endpoint), + phone = VALUES(phone), + device_id = VALUES(device_id), + platform_name = VALUES(platform_name), + first_total_mileage_km = VALUES(first_total_mileage_km), + latest_total_mileage_km = VALUES(latest_total_mileage_km), + daily_mileage_km = VALUES(daily_mileage_km), + sample_count = VALUES(sample_count), + first_event_time = VALUES(first_event_time), + latest_event_time = VALUES(latest_event_time), + quality_status = VALUES(quality_status), + quality_reason = VALUES(quality_reason), + updated_at = CURRENT_TIMESTAMP` diff --git a/go/vehicle-gateway/internal/stats/gps_mileage_test.go b/go/vehicle-gateway/internal/stats/gps_mileage_test.go new file mode 100644 index 00000000..f861ead0 --- /dev/null +++ b/go/vehicle-gateway/internal/stats/gps_mileage_test.go @@ -0,0 +1,188 @@ +package stats + +import ( + "context" + "math" + "strings" + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +func TestCalculateGPSMileageSegmentMatchesTrackSafetyWindow(t *testing.T) { + start := time.Date(2026, 7, 19, 8, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)) + previous := GPSMileagePoint{EventTime: start, Longitude: 121.4737, Latitude: 31.2304} + + valid := CalculateGPSMileageSegment(previous, GPSMileagePoint{ + EventTime: start.Add(5 * time.Minute), + Longitude: 121.4837, + Latitude: 31.2304, + }) + if !valid.Usable || valid.DistanceKM <= 0.9 || valid.DistanceKM >= 1.1 { + t.Fatalf("valid segment = %+v, want about 0.95 km", valid) + } + + longGap := CalculateGPSMileageSegment(previous, GPSMileagePoint{ + EventTime: start.Add(11 * time.Minute), + Longitude: 121.4837, + Latitude: 31.2304, + }) + if !longGap.LongGap || longGap.Usable { + t.Fatalf("long gap segment = %+v", longGap) + } + + jump := CalculateGPSMileageSegment(previous, GPSMileagePoint{ + EventTime: start.Add(time.Minute), + Longitude: 121.5737, + Latitude: 31.2304, + }) + if !jump.BadJump || jump.Usable { + t.Fatalf("jump segment = %+v", jump) + } + + outOfOrder := CalculateGPSMileageSegment(previous, GPSMileagePoint{ + EventTime: start, + Longitude: 121.4747, + Latitude: 31.2304, + }) + if !outOfOrder.OutOfOrder || outOfOrder.Usable { + t.Fatalf("out-of-order segment = %+v", outOfOrder) + } +} + +func TestGPSMileagePointFromEnvelopeUsesNaturalDayAndStableSource(t *testing.T) { + loc := time.FixedZone("Asia/Shanghai", 8*3600) + eventTime := time.Date(2026, 7, 19, 23, 59, 58, 0, loc) + env := envelope.FrameEnvelope{ + EventID: "evt-gps-1", + Protocol: envelope.ProtocolJT808, + VIN: "LMRKH9AC7R1004120", + Phone: "64341233712", + SourceEndpoint: "122.152.221.156:33805", + EventTimeMS: eventTime.UnixMilli(), + Fields: map[string]any{ + "jt808.location.longitude": 121.4737, + "jt808.location.latitude": 31.2304, + }, + } + identity := SourceIdentity{ + Protocol: envelope.ProtocolJT808, + SourceIP: "122.152.221.156", + SourceCode: "dongfang_beidou", + SourceKind: "PLATFORM", + PlatformName: "东方北斗", + } + point, ok := GPSMileagePointFromEnvelope(env, identity, loc) + if !ok { + t.Fatal("expected valid GPS point") + } + if point.StatDate != "2026-07-19" || point.SourceKey != "JT808:64341233712@PLATFORM:dongfang_beidou" { + t.Fatalf("point = %+v", point) + } + candidateKey := gpsMileageCandidateSourceKey(point) + if candidateKey != "JT808:64341233712#GPS_COORDINATE:122.152.221.156" || strings.Contains(candidateKey, platformSourceKeyPrefix) { + t.Fatalf("coordinate candidate key must remain lower-trust than an odometer platform source: %q", candidateKey) + } +} + +func TestAccumulateJT808GPSMileagePersistsEstimateAndProjects(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + + loc := time.FixedZone("Asia/Shanghai", 8*3600) + firstTime := time.Date(2026, 7, 19, 8, 0, 0, 0, loc) + point := GPSMileagePoint{ + VIN: "LMRKH9AC7R1004120", + StatDate: "2026-07-19", + Protocol: envelope.ProtocolJT808, + SourceKey: "JT808:64341233712@PLATFORM:dongfang_beidou", + SourceIP: "122.152.221.156", + SourceEndpoint: "122.152.221.156:33805", + Phone: "64341233712", + EventID: "evt-2", + EventTime: firstTime.Add(5 * time.Minute), + Longitude: 121.4837, + Latitude: 31.2304, + } + + mock.ExpectBegin() + mock.ExpectQuery("SELECT first_event_time, latest_event_time"). + WithArgs(point.VIN, point.StatDate, string(point.Protocol), point.SourceKey). + 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(firstTime, firstTime, "evt-1", 121.4737, 31.2304, 0, 1, 0, 0, 0, 0)) + mock.ExpectExec("UPDATE vehicle_daily_gps_mileage_state"). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("INSERT INTO vehicle_daily_mileage_source"). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("INSERT INTO vehicle_daily_mileage"). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("UPDATE vehicle_daily_mileage_source"). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("DELETE FROM vehicle_daily_mileage"). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectCommit() + + state, recovered, err := AccumulateJT808GPSMileage(context.Background(), db, point) + if err != nil { + t.Fatalf("AccumulateJT808GPSMileage() error = %v", err) + } + if !recovered || state.UsableSegmentCount != 1 || math.Abs(state.DailyMileageKM-0.9529) > 0.02 { + t.Fatalf("state = %+v recovered=%v", state, recovered) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet SQL expectations: %v", err) + } +} + +func TestAccumulateJT808GPSMileageNeedsUsableSegmentBeforeProjection(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + + point := GPSMileagePoint{ + VIN: "LMRKH9AC7R1004120", + StatDate: "2026-07-19", + Protocol: envelope.ProtocolJT808, + SourceKey: "JT808:64341233712@PLATFORM:dongfang_beidou", + SourceIP: "122.152.221.156", + EventID: "evt-1", + EventTime: time.Date(2026, 7, 19, 8, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)), + Longitude: 121.4737, + Latitude: 31.2304, + } + 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", + })) + mock.ExpectQuery("SELECT daily_mileage_km, first_event_time, sample_count"). + WillReturnRows(sqlmock.NewRows([]string{"daily_mileage_km", "first_event_time", "sample_count"})) + mock.ExpectExec("INSERT INTO vehicle_daily_gps_mileage_state"). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectCommit() + + _, recovered, err := AccumulateJT808GPSMileage(context.Background(), db, point) + if err != nil { + t.Fatalf("AccumulateJT808GPSMileage() error = %v", err) + } + if recovered { + t.Fatal("first point must not project an estimated mileage") + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("unmet SQL expectations: %v", err) + } +} diff --git a/go/vehicle-gateway/internal/stats/source_mileage.go b/go/vehicle-gateway/internal/stats/source_mileage.go index 75f149b9..e973f585 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage.go +++ b/go/vehicle-gateway/internal/stats/source_mileage.go @@ -511,6 +511,7 @@ WHERE s.vin = ? AND s.stat_date = ? AND s.protocol = ? AND s.quality_status IN ('` + QualityOK + `', '` + QualityNoPreviousBaseline + `') + AND COALESCE(s.quality_reason, '') <> '` + QualityReasonGPSCoordinate + `' AND s.first_total_mileage_km IS NOT NULL AND s.latest_total_mileage_km IS NOT NULL AND COALESCE(NULLIF(TRIM(ds.source_code), ''), vi.source_code) IS NOT NULL @@ -581,6 +582,7 @@ WHERE s.vin = ? AND s.stat_date = ? AND s.protocol = ? AND s.quality_status IN ('` + QualityOK + `', '` + QualityNoPreviousBaseline + `') + AND COALESCE(s.quality_reason, '') <> '` + QualityReasonGPSCoordinate + `' AND s.first_total_mileage_km IS NOT NULL AND s.latest_total_mileage_km IS NOT NULL AND COALESCE(NULLIF(TRIM(ds.source_code), ''), vi.source_code) IS NOT NULL @@ -614,6 +616,7 @@ LEFT JOIN ( WHERE s.stat_date = ? AND s.protocol = ? AND s.quality_status IN ('` + QualityOK + `', '` + QualityNoPreviousBaseline + `') + AND COALESCE(s.quality_reason, '') <> '` + QualityReasonGPSCoordinate + `' AND COALESCE(NULLIF(TRIM(ds.source_code), ''), vi.source_code) IS NOT NULL AND COALESCE(NULLIF(TRIM(s.phone), ''), NULLIF(TRIM(s.device_id), '')) IS NOT NULL AND s.source_key <> CONCAT(s.protocol, ':', COALESCE(NULLIF(TRIM(s.phone), ''), NULLIF(TRIM(s.device_id), '')), '` + platformSourceKeyPrefix + `', COALESCE(NULLIF(TRIM(ds.source_code), ''), vi.source_code)) @@ -669,7 +672,8 @@ 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 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) +ORDER BY CASE WHEN COALESCE(s.quality_reason, '') = '` + QualityReasonGPSCoordinate + `' THEN 1 ELSE 0 END, + 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 WHEN 'UNKNOWN' THEN 2 @@ -709,7 +713,8 @@ 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 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) + ORDER BY CASE WHEN COALESCE(s2.quality_reason, '') = '` + QualityReasonGPSCoordinate + `' THEN 1 ELSE 0 END, + 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 WHEN 'UNKNOWN' THEN 2 diff --git a/go/vehicle-gateway/internal/stats/source_mileage_test.go b/go/vehicle-gateway/internal/stats/source_mileage_test.go index f1c1cd5f..bfc5c96b 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage_test.go +++ b/go/vehicle-gateway/internal/stats/source_mileage_test.go @@ -505,6 +505,7 @@ 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", "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", @@ -553,6 +554,7 @@ 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", "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",