shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(stats): calculate mileage without previous-day baseline

Browse Source
This commit is contained in:
lingniu
2026-07-08 17:26:34 +08:00
parent 536b0bfee6
commit 469e9c75f6
5 changed files with 315 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

182
go/vehicle-gateway/cmd/stats-backfill/main.go
View File

@@ -74,8 +74,11 @@ type dailySourceLast struct {
Phone string
DeviceID string
SourceEndpoint string
FirstTS time.Time
TS time.Time
FirstTotalKM float64
TotalKM float64
RawSampleCount int64
}
func main() {
@@ -233,21 +236,17 @@ func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample)
SourceEndpoint: sample.SourceEndpoint,
FirstEventTime: sample.EventTime,
LatestEventTime: sample.EventTime,
QualityStatus: stats.QualityNoPreviousBaseline,
QualityReason: "scan_backfill_missing_previous_source",
QualityStatus: stats.QualityOK,
QualityReason: "current_day_first_sample",
}
continue
}
if sample.TotalMileageKM < agg.FirstKM {
agg.FirstKM = sample.TotalMileageKM
}
if sample.TotalMileageKM > agg.LatestKM {
agg.LatestKM = sample.TotalMileageKM
}
if agg.FirstEventTime.IsZero() || sample.EventTime.Before(agg.FirstEventTime) {
agg.FirstKM = sample.TotalMileageKM
agg.FirstEventTime = sample.EventTime
}
if sample.EventTime.After(agg.LatestEventTime) {
agg.LatestKM = sample.TotalMileageKM
agg.LatestEventTime = sample.EventTime
agg.SourceEndpoint = sample.SourceEndpoint
if strings.TrimSpace(sample.Phone) != "" {
@@ -339,22 +338,6 @@ func clearBackfillTargetMileage(ctx context.Context, db *sql.DB, vin string, sta
}
func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[string]*metricAgg, error) {
dates, err := dateRangeWithPrevious(cfg.DateFrom, cfg.DateTo)
if err != nil {
return nil, err
}
lastByProtocolDate := map[envelope.Protocol]map[string]map[string][]dailySourceLast{}
for _, protocol := range cfg.Protocols {
lastByProtocolDate[protocol] = map[string]map[string][]dailySourceLast{}
for _, date := range dates {
rows, err := queryDailyLastSourceRows(ctx, db, cfg, protocol, date)
if err != nil {
return nil, err
}
lastByProtocolDate[protocol][date] = rows
slog.Info("daily last loaded", "protocol", protocol, "date", date, "vehicles", len(rows))
}
}
targetDates, err := dateRange(cfg.DateFrom, cfg.DateTo)
if err != nil {
return nil, err
@@ -362,9 +345,15 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s
aggregates := map[string]*metricAgg{}
for _, protocol := range cfg.Protocols {
for _, date := range targetDates {
prevDate := previousDate(date)
current := lastByProtocolDate[protocol][date]
previous := lastByProtocolDate[protocol][prevDate]
current, err := queryDailyLastSourceRows(ctx, db, cfg, protocol, date)
if err != nil {
return nil, err
}
previous, err := queryPreviousLastSourceRows(ctx, db, cfg, protocol, date)
if err != nil {
return nil, err
}
slog.Info("daily last loaded", "protocol", protocol, "date", date, "vehicles", len(current), "previousVehicles", len(previous))
for vin, currentSources := range current {
previousBySource := map[string]dailySourceLast{}
for _, previousRow := range previous[vin] {
@@ -372,29 +361,8 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s
}
for _, currentRow := range currentSources {
key := vin + "|" + date + "|" + string(protocol) + "|" + currentRow.SourceKey
agg := &metricAgg{
VIN: vin,
Date: date,
Protocol: protocol,
FirstKM: currentRow.TotalKM,
LatestKM: currentRow.TotalKM,
Count: 1,
SourceKey: currentRow.SourceKey,
Phone: currentRow.Phone,
DeviceID: currentRow.DeviceID,
SourceEndpoint: currentRow.SourceEndpoint,
FirstEventTime: currentRow.TS,
LatestEventTime: currentRow.TS,
QualityStatus: stats.QualityNoPreviousBaseline,
QualityReason: "missing_previous_source",
}
if previousRow, ok := previousBySource[currentRow.SourceKey]; ok {
agg.FirstKM = previousRow.TotalKM
agg.FirstEventTime = previousRow.TS
agg.QualityStatus = stats.QualityOK
agg.QualityReason = "same_source_previous_day"
}
aggregates[key] = agg
previousRow, hasPrevious := previousBySource[currentRow.SourceKey]
aggregates[key] = aggregateFromDailySource(date, protocol, currentRow, previousRow, hasPrevious)
}
}
}
@@ -402,6 +370,40 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s
return aggregates, nil
}
func aggregateFromDailySource(date string, protocol envelope.Protocol, current dailySourceLast, previous dailySourceLast, hasPrevious bool) *metricAgg {
firstKM := current.FirstTotalKM
firstEventTime := current.FirstTS
qualityReason := "current_day_first_sample"
if hasPrevious {
firstKM = previous.TotalKM
firstEventTime = previous.TS
qualityReason = "historical_source_baseline"
}
if firstEventTime.IsZero() {
firstEventTime = current.TS
}
count := current.RawSampleCount
if count <= 0 {
count = 1
}
return &metricAgg{
VIN: current.VIN,
Date: date,
Protocol: protocol,
FirstKM: firstKM,
LatestKM: current.TotalKM,
Count: count,
SourceKey: current.SourceKey,
Phone: current.Phone,
DeviceID: current.DeviceID,
SourceEndpoint: current.SourceEndpoint,
FirstEventTime: firstEventTime,
LatestEventTime: current.TS,
QualityStatus: stats.QualityOK,
QualityReason: qualityReason,
}
}
type trustedChoice struct {
current dailySourceLast
previous dailySourceLast
@@ -443,10 +445,30 @@ func queryDailyLastSourceRows(ctx context.Context, db *sql.DB, cfg config, proto
fmt.Sprintf("protocol = '%s'", quote(string(protocol))),
realtimeMileageFramePredicate(),
}
sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint, LAST(ts), LAST(parsed_json)
sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint, FIRST(ts), FIRST(parsed_json), LAST(ts), LAST(parsed_json), COUNT(*)
FROM %s.raw_frames
WHERE %s
GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
return querySourceRows(ctx, db, cfg, protocol, sqlText, true)
}
func queryPreviousLastSourceRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, date string) (map[string][]dailySourceLast, error) {
where := []string{
fmt.Sprintf("ts < '%s 00:00:00'", quote(date)),
"parse_status = 'OK'",
"vin IS NOT NULL",
"vin <> ''",
fmt.Sprintf("protocol = '%s'", quote(string(protocol))),
realtimeMileageFramePredicate(),
}
sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint, LAST(ts), LAST(parsed_json), COUNT(*)
FROM %s.raw_frames
WHERE %s
GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
return querySourceRows(ctx, db, cfg, protocol, sqlText, false)
}
func querySourceRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, sqlText string, includeFirst bool) (map[string][]dailySourceLast, error) {
rows, err := db.QueryContext(ctx, sqlText)
if err != nil {
return nil, err
@@ -458,25 +480,32 @@ GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), s
var phone string
var deviceID string
var sourceEndpoint string
var firstTS time.Time
var firstParsedJSON string
var ts time.Time
var parsedJSON string
if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &ts, &parsedJSON); err != nil {
return nil, err
var rawSampleCount int64
if includeFirst {
if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &firstTS, &firstParsedJSON, &ts, &parsedJSON, &rawSampleCount); err != nil {
return nil, err
}
} else {
if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &ts, &parsedJSON, &rawSampleCount); err != nil {
return nil, err
}
firstTS = ts
firstParsedJSON = parsedJSON
}
fields := fieldsForStats(protocol, vin, parsedJSON)
env := envelope.FrameEnvelope{
Protocol: protocol,
VIN: strings.TrimSpace(vin),
EventTimeMS: time.Now().UnixMilli(),
ReceivedAtMS: time.Now().UnixMilli(),
Fields: fields,
ParsedFields: fields,
ParseStatus: envelope.ParseOK,
}
samples, err := stats.SamplesFromEnvelope(env, cfg.Location)
if err != nil || len(samples) == 0 {
latestTotalKM, ok := mileageFromParsed(protocol, vin, parsedJSON, ts, cfg.Location)
if !ok {
continue
}
firstTotalKM := latestTotalKM
if includeFirst {
if parsedFirst, ok := mileageFromParsed(protocol, vin, firstParsedJSON, firstTS, cfg.Location); ok {
firstTotalKM = parsedFirst
}
}
sourceKey := normalizedSourceKey(string(protocol), phone, deviceID, sourceEndpoint)
if sourceKey == "" {
continue
@@ -487,8 +516,11 @@ GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), s
Phone: strings.TrimSpace(phone),
DeviceID: strings.TrimSpace(deviceID),
SourceEndpoint: strings.TrimSpace(sourceEndpoint),
FirstTS: firstTS.In(cfg.Location),
TS: ts,
TotalKM: samples[0].TotalMileageKM,
FirstTotalKM: firstTotalKM,
TotalKM: latestTotalKM,
RawSampleCount: rawSampleCount,
}
key := row.VIN + "|" + row.SourceKey
if existing, ok := latestBySource[key]; !ok || row.TS.After(existing.TS) {
@@ -505,6 +537,24 @@ GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), s
return out, nil
}
func mileageFromParsed(protocol envelope.Protocol, vin string, parsedJSON string, eventTime time.Time, loc *time.Location) (float64, bool) {
fields := fieldsForStats(protocol, vin, parsedJSON)
env := envelope.FrameEnvelope{
Protocol: protocol,
VIN: strings.TrimSpace(vin),
EventTimeMS: eventTime.UnixMilli(),
ReceivedAtMS: eventTime.UnixMilli(),
Fields: fields,
ParsedFields: fields,
ParseStatus: envelope.ParseOK,
}
samples, err := stats.SamplesFromEnvelope(env, loc)
if err != nil || len(samples) == 0 {
return 0, false
}
return samples[0].TotalMileageKM, true
}
func normalizedSourceKey(protocol string, phone string, deviceID string, endpoint string) string {
sourceIP := stats.NormalizeSourceIP(endpoint)
return stats.SourceKey(envelope.Protocol(protocol), phone, deviceID, sourceIP)

95
go/vehicle-gateway/cmd/stats-backfill/main_test.go
View File

@@ -67,6 +67,69 @@ func TestDailySourceLastBuildsCandidateKeysBySourceIP(t *testing.T) {
}
}
func TestAggregateFromDailySourceUsesOlderHistoricalBaseline(t *testing.T) {
current := dailySourceLast{
VIN: "LMRKH9AC2R1004087",
SourceKey: normalizedSourceKey("YUTONG_MQTT", "", "LMRKH9AC2R1004087", "mqtt://yutong/ytforward/shln/3"),
DeviceID: "LMRKH9AC2R1004087",
SourceEndpoint: "mqtt://yutong/ytforward/shln/3",
FirstTS: time.Date(2026, 7, 8, 8, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
TS: time.Date(2026, 7, 8, 17, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
FirstTotalKM: 120778,
TotalKM: 120788,
RawSampleCount: 15,
}
previous := dailySourceLast{
VIN: "LMRKH9AC2R1004087",
SourceKey: current.SourceKey,
DeviceID: "LMRKH9AC2R1004087",
SourceEndpoint: "mqtt://yutong/ytforward/shln/3",
TS: time.Date(2026, 7, 4, 23, 58, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
TotalKM: 120672,
}
agg := aggregateFromDailySource("2026-07-08", envelope.ProtocolYutongMQTT, current, previous, true)
if agg.FirstKM != 120672 || agg.LatestKM != 120788 {
t.Fatalf("km range = %v -> %v", agg.FirstKM, agg.LatestKM)
}
if agg.FirstEventTime != previous.TS || agg.LatestEventTime != current.TS {
t.Fatalf("event range = %v -> %v", agg.FirstEventTime, agg.LatestEventTime)
}
if agg.QualityStatus != stats.QualityOK || agg.QualityReason != "historical_source_baseline" {
t.Fatalf("quality = %s/%s", agg.QualityStatus, agg.QualityReason)
}
if agg.Count != 15 {
t.Fatalf("sample count = %d, want 15", agg.Count)
}
}
func TestAggregateFromDailySourceUsesCurrentFirstSampleWithoutHistory(t *testing.T) {
current := dailySourceLast{
VIN: "LMRKH9AC2R1004087",
SourceKey: normalizedSourceKey("YUTONG_MQTT", "", "LMRKH9AC2R1004087", "mqtt://yutong/ytforward/shln/3"),
DeviceID: "LMRKH9AC2R1004087",
SourceEndpoint: "mqtt://yutong/ytforward/shln/3",
FirstTS: time.Date(2026, 7, 8, 8, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
TS: time.Date(2026, 7, 8, 17, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
FirstTotalKM: 120778,
TotalKM: 120788,
RawSampleCount: 15,
}
agg := aggregateFromDailySource("2026-07-08", envelope.ProtocolYutongMQTT, current, dailySourceLast{}, false)
if agg.FirstKM != 120778 || agg.LatestKM != 120788 {
t.Fatalf("km range = %v -> %v", agg.FirstKM, agg.LatestKM)
}
if agg.FirstEventTime != current.FirstTS || agg.LatestEventTime != current.TS {
t.Fatalf("event range = %v -> %v", agg.FirstEventTime, agg.LatestEventTime)
}
if agg.QualityStatus != stats.QualityOK || agg.QualityReason != "current_day_first_sample" {
t.Fatalf("quality = %s/%s", agg.QualityStatus, agg.QualityReason)
}
}
func TestClearBackfillTargetMileageClearsExactKey(t *testing.T) {
db, mock, err := sqlmock.New()
if err != nil {
@@ -219,9 +282,9 @@ func TestWriteAggregatesSkipsBlankSourceBeforeClearingTarget(t *testing.T) {
}
}
func TestAddSamplesKeepsScanCandidatesNoPreviousBaseline(t *testing.T) {
func TestAddSamplesUsesCurrentDayFirstSampleBaseline(t *testing.T) {
loc := time.FixedZone("Asia/Shanghai", 8*3600)
samples, err := stats.SamplesFromEnvelope(envelope.FrameEnvelope{
firstSamples, err := stats.SamplesFromEnvelope(envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
VIN: "LA9GG64L7PBAF4001",
Phone: "13307765812",
@@ -234,18 +297,38 @@ func TestAddSamplesKeepsScanCandidatesNoPreviousBaseline(t *testing.T) {
if err != nil {
t.Fatalf("SamplesFromEnvelope() error = %v", err)
}
secondSamples, err := stats.SamplesFromEnvelope(envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
VIN: "LA9GG64L7PBAF4001",
Phone: "13307765812",
SourceEndpoint: "115.231.168.135:20215",
EventTimeMS: time.Date(2026, 7, 8, 13, 0, 0, 0, loc).UnixMilli(),
Fields: map[string]any{
"jt808.location.total_mileage_km": 4129.9,
},
}, loc)
if err != nil {
t.Fatalf("SamplesFromEnvelope() error = %v", err)
}
aggregates := map[string]*metricAgg{}
addSamples(aggregates, samples)
addSamples(aggregates, firstSamples)
addSamples(aggregates, secondSamples)
if len(aggregates) != 1 {
t.Fatalf("aggregate count = %d, want 1", len(aggregates))
}
for _, agg := range aggregates {
if agg.QualityStatus != stats.QualityNoPreviousBaseline {
t.Fatalf("quality = %q, want %q", agg.QualityStatus, stats.QualityNoPreviousBaseline)
if agg.QualityStatus != stats.QualityOK {
t.Fatalf("quality = %q, want %q", agg.QualityStatus, stats.QualityOK)
}
if agg.QualityReason != "scan_backfill_missing_previous_source" {
if agg.QualityReason != "current_day_first_sample" {
t.Fatalf("quality reason = %q", agg.QualityReason)
}
if agg.FirstKM != 4123.9 || agg.LatestKM != 4129.9 {
t.Fatalf("km range = %v -> %v", agg.FirstKM, agg.LatestKM)
}
if agg.FirstEventTime != time.Date(2026, 7, 8, 12, 0, 0, 0, loc) {
t.Fatalf("first event time = %v", agg.FirstEventTime)
}
}
}