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

feat: 按35MPa满充模型提供剩余氢量估算与百分比

Browse Source
This commit is contained in:
lingniu
2026-09-08 23:12:57 +08:00
parent a6e31effc2
commit 4ab654e1c1
9 changed files with 497 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vehicle-data-platform/apps/api/internal/openplatform/realtime_live.go
+35 -18
View File
@@ -15,21 +15,31 @@ import (
)
// GB32960 live freshness is an API policy, not an assertion about the device's
// negotiated reporting interval. No supported protocol currently has a verified
// full-vehicle percentage/capacity contract.
// negotiated reporting interval.
const realtimeHydrogenStaleSeconds int64 = 300
type RealtimeHydrogenData struct {
RemainingHydrogenKg *float64 `json:"remainingHydrogenKg"`
RemainingHydrogenPercent *float64 `json:"remainingHydrogenPercent"`
HydrogenRecordTime *string `json:"hydrogenRecordTime"`
HydrogenDataStatus string `json:"hydrogenDataStatus"`
RemainingHydrogenKgStatus string `json:"remainingHydrogenKgStatus"`
RemainingHydrogenPercentStatus string `json:"remainingHydrogenPercentStatus"`
HydrogenValueSource *string `json:"hydrogenValueSource"`
HydrogenSourceProtocol *string `json:"hydrogenSourceProtocol"`
HydrogenStaleAfterSeconds *int64 `json:"hydrogenStaleAfterSeconds"`
HydrogenExpectedIntervalSeconds *int64 `json:"hydrogenExpectedIntervalSeconds"`
HydrogenPressureTemperatureSource *string `json:"hydrogenPressureTemperatureSource"`
RemainingHydrogenPercentSource *string `json:"remainingHydrogenPercentSource"`
HydrogenFullCapacityKg *float64 `json:"hydrogenFullCapacityKg"`
HydrogenTankCapacityL *float64 `json:"hydrogenTankCapacityL"`
HydrogenFullPressureMPa *float64 `json:"hydrogenFullPressureMPa"`
HydrogenReferenceTemperatureC *float64 `json:"hydrogenReferenceTemperatureC"`
HydrogenEstimatePressureMPa *float64 `json:"hydrogenEstimatePressureMPa"`
HydrogenEstimateTemperatureC *float64 `json:"hydrogenEstimateTemperatureC"`
HydrogenCalculationVersion *string `json:"hydrogenCalculationVersion"`
HydrogenCapacitySource *string `json:"hydrogenCapacitySource"`
HydrogenPercentReason *string `json:"hydrogenPercentReason"`
RemainingHydrogenKg *float64 `json:"remainingHydrogenKg"`
RemainingHydrogenPercent *float64 `json:"remainingHydrogenPercent"`
HydrogenRecordTime *string `json:"hydrogenRecordTime"`
HydrogenDataStatus string `json:"hydrogenDataStatus"`
RemainingHydrogenKgStatus string `json:"remainingHydrogenKgStatus"`
RemainingHydrogenPercentStatus string `json:"remainingHydrogenPercentStatus"`
HydrogenValueSource *string `json:"hydrogenValueSource"`
HydrogenSourceProtocol *string `json:"hydrogenSourceProtocol"`
HydrogenStaleAfterSeconds *int64 `json:"hydrogenStaleAfterSeconds"`
HydrogenExpectedIntervalSeconds *int64 `json:"hydrogenExpectedIntervalSeconds"`
}
func missingRealtimeHydrogen(protocol string) RealtimeHydrogenData {
@@ -37,7 +47,7 @@ func missingRealtimeHydrogen(protocol string) RealtimeHydrogenData {
if protocol == "" || protocol == "GB32960" {
status = "MISSING"
}
result := RealtimeHydrogenData{HydrogenDataStatus: status, RemainingHydrogenKgStatus: status, RemainingHydrogenPercentStatus: "UNSUPPORTED"}
result := RealtimeHydrogenData{HydrogenDataStatus: status, RemainingHydrogenKgStatus: status, RemainingHydrogenPercentStatus: status}
if protocol == "GB32960" {
threshold := realtimeHydrogenStaleSeconds
result.HydrogenStaleAfterSeconds = &threshold
@@ -231,6 +241,10 @@ func (r *MySQLRepository) loadRealtimeLiveData(ctx context.Context, vins []strin
// At most two references per requested VIN. Batches bound SQL size; primary
// timestamp and VIN tag filters prevent unbounded history scans.
frames, loadErr := r.loadRealtimeRawFrameBatches(ctx, ordered)
capacities, capacityErr := r.loadRealtimeHydrogenCapacities(ctx, vins)
if capacityErr != nil {
log.Printf("openplatform realtime hydrogen capacity unavailable: %s", realtimeEnrichmentErrorReason(capacityErr))
}
for _, frame := range frames {
ref, exists := refs[realtimeReferenceKey(frame.VIN, frame.Protocol, frame.EventID)]
if !exists {
@@ -245,7 +259,7 @@ func (r *MySQLRepository) loadRealtimeLiveData(ctx context.Context, vins []strin
if frame.EventMS.Valid && frame.EventMS.Int64 > 0 {
at = time.UnixMilli(frame.EventMS.Int64)
}
point.LiveHydrogen = realtimeHydrogenFromFrame(frame.Parsed.String, at, now)
point.LiveHydrogen = realtimeHydrogenWithCapacity(frame.Parsed.String, at, now, capacities[frame.VIN])
}
if ref.Location {
point.GPSFixStatus, point.CoordinateSystem = realtimeGPSFromFrame(frame.Protocol, frame.Parsed.String)
@@ -260,7 +274,7 @@ func (r *MySQLRepository) loadRealtimeLiveData(ctx context.Context, vins []strin
}
}
fallback, err := r.loadRealtimeRawFrameBatchesWithQuery(ctx, missing, realtimeHydrogenFallbackQuery, 1)
applyRealtimeHydrogenFallback(points, missing, fallback, now)
applyRealtimeHydrogenFallback(points, missing, fallback, now, capacities)
if err != nil {
return err
}
@@ -270,7 +284,7 @@ func (r *MySQLRepository) loadRealtimeLiveData(ctx context.Context, vins []strin
// Only explicitly missing hydrogen can be filled. In particular, the newest
// invalid measurement must never be hidden by an older valid reading.
func applyRealtimeHydrogenFallback(points map[string]RealtimeVehiclePoint, refs []realtimeFrameReference, frames []realtimeRawFrame, now time.Time) {
func applyRealtimeHydrogenFallback(points map[string]RealtimeVehiclePoint, refs []realtimeFrameReference, frames []realtimeRawFrame, now time.Time, capacityMaps ...map[string]float64) {
allowed := make(map[string]bool, len(refs))
for _, ref := range refs {
allowed[ref.VIN] = true
@@ -291,6 +305,9 @@ func applyRealtimeHydrogenFallback(points map[string]RealtimeVehiclePoint, refs
at = time.UnixMilli(frame.EventMS.Int64)
}
data := realtimeHydrogenFromFrame(frame.Parsed.String, at, now)
if len(capacityMaps) > 0 {
data = realtimeHydrogenWithCapacity(frame.Parsed.String, at, now, capacityMaps[0][frame.VIN])
}
if data.HydrogenDataStatus == "MISSING" {
continue
} // LIKE is a prefilter, never a JSON parser.
@@ -451,10 +468,10 @@ func realtimeHydrogenFallbackQuery(database string, refs []realtimeFrameReferenc
vins = append(vins, quote(ref.VIN))
clauses = append(clauses, "(vin="+quote(ref.VIN)+" AND ts>="+strconv.FormatInt(start, 10)+" AND ts<="+strconv.FormatInt(end, 10)+")")
}
for _, key := range hydrogenMassFields {
for _, key := range append(append([]string{}, hydrogenMassFields...), realtimeHydrogenPressureField, realtimeHydrogenTemperatureField) {
fields = append(fields, "parsed_json LIKE "+quote("%\""+key+"\":%"))
}
return `SELECT vin,protocol,event_id,CAST(event_time AS BIGINT),parsed_json FROM ` + database + `.raw_frames WHERE protocol='GB32960' AND ts>=` + strconv.FormatInt(min, 10) + ` AND ts<=` + strconv.FormatInt(max, 10) + ` AND vin IN (` + strings.Join(vins, ",") + `) AND parse_status='OK' AND (` + strings.Join(clauses, " OR ") + `) AND (` + strings.Join(fields, " OR ") + `) ORDER BY event_time DESC,ts DESC LIMIT 1`, nil
return `SELECT vin,protocol,event_id,CAST(event_time AS BIGINT),parsed_json FROM ` + database + `.raw_frames WHERE protocol='GB32960' AND ts>=` + strconv.FormatInt(min, 10) + ` AND ts<=` + strconv.FormatInt(max, 10) + ` AND vin IN (` + strings.Join(vins, ",") + `) AND parse_status='OK' AND (` + strings.Join(clauses, " OR ") + `) AND (` + strings.Join(fields, " OR ") + `) ORDER BY event_time DESC,ts DESC LIMIT 5`, nil
}
var realtimeErrorQuotedText = regexp.MustCompile(`'[^']*'|"[^"]*"`)