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

fix(mileage): recover sparse source daily distance

Browse Source
This commit is contained in:
lingniu
2026-07-19 22:57:31 +08:00
parent e45daff528
commit 4c64c4c738
11 changed files with 364 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
go/vehicle-gateway/internal/stats/daily_metric.go
View File

@@ -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 {

39
go/vehicle-gateway/internal/stats/gps_mileage.go
View File

@@ -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 == "" {

64
go/vehicle-gateway/internal/stats/gps_mileage_test.go
View File

@@ -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 {

26
go/vehicle-gateway/internal/stats/source_mileage.go
View File

@@ -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

10
go/vehicle-gateway/internal/stats/source_mileage_test.go
View File

@@ -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",