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

feat(history): align fields quality and exports

Browse Source
This commit is contained in:
lingniu
2026-07-16 16:46:10 +08:00
parent 17c4591040
commit a2722e4afd
13 changed files with 273 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
vehicle-data-platform/apps/api/internal/platform/mock_store.go
View File

@@ -846,7 +846,8 @@ func (m *MockStore) mockHistoryExportRows(ctx context.Context, query HistoryExpo
for key, value := range item.ParsedFields {
values[key] = value
}
rows = append(rows, HistoryDataRow{ID: item.ID, VIN: item.VIN, Plate: item.Plate, Protocol: item.Protocol, DeviceTime: item.DeviceTime, ServerTime: item.ServerTime, Quality: "normal", EvidenceID: item.ID, Values: values})
quality, reason := historyRawQuality(item)
rows = append(rows, HistoryDataRow{ID: item.ID, VIN: item.VIN, Plate: item.Plate, Protocol: item.Protocol, DeviceTime: item.DeviceTime, ServerTime: item.ServerTime, Quality: quality, QualityReason: reason, EvidenceID: item.ID, Values: values})
}
return rows, nil
case "mileage":
@@ -856,7 +857,8 @@ func (m *MockStore) mockHistoryExportRows(ctx context.Context, query HistoryExpo
}
rows := make([]HistoryDataRow, 0, len(page.Items))
for index, item := range page.Items {
rows = append(rows, HistoryDataRow{ID: "mileage-" + item.VIN + "-" + strconv.Itoa(index), VIN: item.VIN, Plate: item.Plate, Protocol: item.Source, DeviceTime: item.Date, ServerTime: item.Date, Quality: "normal", Values: map[string]any{"startMileageKm": item.StartMileageKm, "endMileageKm": item.EndMileageKm, "dailyMileageKm": item.DailyMileageKm}})
quality, reason := historyMileageQuality(item)
rows = append(rows, HistoryDataRow{ID: "mileage-" + item.VIN + "-" + strconv.Itoa(index), VIN: item.VIN, Plate: item.Plate, Protocol: item.Source, DeviceTime: item.Date, ServerTime: item.Date, Quality: quality, QualityReason: reason, Values: map[string]any{"startMileageKm": item.StartMileageKm, "endMileageKm": item.EndMileageKm, "dailyMileageKm": item.DailyMileageKm}})
}
return rows, nil
default:
@@ -866,7 +868,8 @@ func (m *MockStore) mockHistoryExportRows(ctx context.Context, query HistoryExpo
}
rows := make([]HistoryDataRow, 0, len(page.Items))
for index, item := range page.Items {
rows = append(rows, HistoryDataRow{ID: "location-" + item.VIN + "-" + strconv.Itoa(index), VIN: item.VIN, Plate: item.Plate, Protocol: item.Protocol, DeviceTime: item.DeviceTime, ServerTime: item.ServerTime, Quality: "normal", Values: map[string]any{"speedKmh": item.SpeedKmh, "totalMileageKm": item.TotalMileageKm, "longitude": item.Longitude, "latitude": item.Latitude}})
quality, reason := historyLocationQuality(item)
rows = append(rows, HistoryDataRow{ID: "location-" + item.VIN + "-" + strconv.Itoa(index), VIN: item.VIN, Plate: item.Plate, Protocol: item.Protocol, DeviceTime: item.DeviceTime, ServerTime: item.ServerTime, Quality: quality, QualityReason: reason, Values: map[string]any{"speedKmh": item.SpeedKmh, "totalMileageKm": item.TotalMileageKm, "longitude": item.Longitude, "latitude": item.Latitude}})
}
return rows, nil
}

19
vehicle-data-platform/apps/api/internal/platform/model.go
View File

@@ -241,15 +241,16 @@ type HistoryDataSummary struct {
}
type HistoryDataRow struct {
ID string `json:"id"`
VIN string `json:"vin"`
Plate string `json:"plate"`
Protocol string `json:"protocol"`
DeviceTime string `json:"deviceTime"`
ServerTime string `json:"serverTime"`
Quality string `json:"quality"`
EvidenceID string `json:"evidenceId,omitempty"`
Values map[string]any `json:"values"`
ID string `json:"id"`
VIN string `json:"vin"`
Plate string `json:"plate"`
Protocol string `json:"protocol"`
DeviceTime string `json:"deviceTime"`
ServerTime string `json:"serverTime"`
Quality string `json:"quality"`
QualityReason string `json:"qualityReason"`
EvidenceID string `json:"evidenceId,omitempty"`
Values map[string]any `json:"values"`
}
type HistorySeriesResponse struct {

20
vehicle-data-platform/apps/api/internal/platform/production_store.go
View File

@@ -757,7 +757,8 @@ func (s *ProductionStore) HistoryExportBatch(ctx context.Context, query HistoryE
}
result := make([]HistoryDataRow, 0, len(page.Items))
for index, item := range page.Items {
result = append(result, HistoryDataRow{ID: "mileage-" + item.VIN + "-" + strconv.Itoa(cursor.Offset+index), VIN: item.VIN, Plate: item.Plate, Protocol: item.Source, DeviceTime: item.Date, ServerTime: item.Date, Quality: "normal", Values: map[string]any{"startMileageKm": item.StartMileageKm, "endMileageKm": item.EndMileageKm, "dailyMileageKm": item.DailyMileageKm}})
quality, reason := historyMileageQuality(item)
result = append(result, HistoryDataRow{ID: "mileage-" + item.VIN + "-" + strconv.Itoa(cursor.Offset+index), VIN: item.VIN, Plate: item.Plate, Protocol: item.Source, DeviceTime: item.Date, ServerTime: item.Date, Quality: quality, QualityReason: reason, Values: map[string]any{"startMileageKm": item.StartMileageKm, "endMileageKm": item.EndMileageKm, "dailyMileageKm": item.DailyMileageKm}})
}
cursor.Offset += len(result)
return result, cursor, nil
@@ -792,12 +793,9 @@ func scanLocationHistoryExportRows(rows *sql.Rows, cursor HistoryExportCursor, p
if err := rows.Scan(&ts, &vin, &protocol, &longitude, &latitude, &speed, &mileage, &receivedAt); err != nil {
return nil, cursor, err
}
quality := "normal"
location := HistoryLocationRow{Longitude: nullFloat64(longitude), Latitude: nullFloat64(latitude)}
if !validTrackCoordinate(location) {
quality = "warning"
}
items = append(items, HistoryDataRow{ID: "location-" + vin + "-" + protocol + "-" + ts, VIN: vin, Plate: plate, Protocol: protocol, DeviceTime: ts, ServerTime: firstNonEmpty(receivedAt, ts), Quality: quality, Values: map[string]any{"speedKmh": nullFloat64(speed), "totalMileageKm": nullFloat64(mileage), "longitude": nullFloat64(longitude), "latitude": nullFloat64(latitude)}})
location := HistoryLocationRow{Longitude: nullFloat64(longitude), Latitude: nullFloat64(latitude), SpeedKmh: nullFloat64(speed), TotalMileageKm: nullFloat64(mileage), DeviceTime: ts, ServerTime: firstNonEmpty(receivedAt, ts)}
quality, reason := historyLocationQuality(location)
items = append(items, HistoryDataRow{ID: "location-" + vin + "-" + protocol + "-" + ts, VIN: vin, Plate: plate, Protocol: protocol, DeviceTime: ts, ServerTime: firstNonEmpty(receivedAt, ts), Quality: quality, QualityReason: reason, Values: map[string]any{"speedKmh": nullFloat64(speed), "totalMileageKm": nullFloat64(mileage), "longitude": nullFloat64(longitude), "latitude": nullFloat64(latitude)}})
cursor = HistoryExportCursor{Time: ts, Protocol: protocol}
}
return items, cursor, rows.Err()
@@ -819,11 +817,9 @@ func scanRawHistoryExportRows(rows *sql.Rows, query HistoryExportStoreQuery, cur
for key, value := range fields {
values[key] = value
}
quality := "normal"
if parseStatus != "" && !strings.EqualFold(parseStatus, "ok") {
quality = "warning"
}
items = append(items, HistoryDataRow{ID: frameID, VIN: vin, Plate: plate, Protocol: protocol, DeviceTime: firstNonEmpty(eventTime, ts), ServerTime: firstNonEmpty(receivedAt, ts), Quality: quality, EvidenceID: frameID, Values: values})
frame := RawFrameRow{ParseStatus: parseStatus, ParseError: parseError, DeviceTime: firstNonEmpty(eventTime, ts), ServerTime: firstNonEmpty(receivedAt, ts)}
quality, reason := historyRawQuality(frame)
items = append(items, HistoryDataRow{ID: frameID, VIN: vin, Plate: plate, Protocol: protocol, DeviceTime: frame.DeviceTime, ServerTime: frame.ServerTime, Quality: quality, QualityReason: reason, EvidenceID: frameID, Values: values})
cursor = HistoryExportCursor{Time: ts, Protocol: protocol, ID: frameID}
}
return items, cursor, rows.Err()

92
vehicle-data-platform/apps/api/internal/platform/service.go
View File

@@ -2127,7 +2127,7 @@ func (s *Service) HistoryData(ctx context.Context, query url.Values) (HistoryDat
pageRows := append([]HistoryDataRow(nil), rows[offset:end]...)
columns := historyMetricsForCategory(category)
if category == "raw" {
columns = mergeDiscoveredRawMetrics(columns, pageRows)
columns = mergeDiscoveredRawMetrics(columns, rows)
}
vehicles := map[string]struct{}{}
sources := map[string]struct{}{}
@@ -2396,11 +2396,8 @@ func (s *Service) historyDataForVehicle(ctx context.Context, category, keyword s
}
rows := make([]HistoryDataRow, 0, len(page.Items))
for index, item := range page.Items {
quality := "normal"
if !validTrackCoordinate(item) {
quality = "warning"
}
rows = append(rows, HistoryDataRow{ID: "location-" + item.VIN + "-" + strconv.Itoa(index) + "-" + item.DeviceTime, VIN: item.VIN, Plate: item.Plate, Protocol: item.Protocol, DeviceTime: item.DeviceTime, ServerTime: item.ServerTime, Quality: quality, Values: map[string]any{"speedKmh": item.SpeedKmh, "totalMileageKm": item.TotalMileageKm, "longitude": item.Longitude, "latitude": item.Latitude}})
quality, reason := historyLocationQuality(item)
rows = append(rows, HistoryDataRow{ID: "location-" + item.VIN + "-" + strconv.Itoa(index) + "-" + item.DeviceTime, VIN: item.VIN, Plate: item.Plate, Protocol: item.Protocol, DeviceTime: item.DeviceTime, ServerTime: item.ServerTime, Quality: quality, QualityReason: reason, Values: map[string]any{"speedKmh": item.SpeedKmh, "totalMileageKm": item.TotalMileageKm, "longitude": item.Longitude, "latitude": item.Latitude}})
}
return rows, page.Total, nil
case "raw":
@@ -2418,7 +2415,8 @@ func (s *Service) historyDataForVehicle(ctx context.Context, category, keyword s
for key, value := range item.ParsedFields {
values[key] = value
}
rows = append(rows, HistoryDataRow{ID: item.ID, VIN: item.VIN, Plate: item.Plate, Protocol: item.Protocol, DeviceTime: item.DeviceTime, ServerTime: item.ServerTime, Quality: "normal", EvidenceID: item.ID, Values: values})
quality, reason := historyRawQuality(item)
rows = append(rows, HistoryDataRow{ID: item.ID, VIN: item.VIN, Plate: item.Plate, Protocol: item.Protocol, DeviceTime: item.DeviceTime, ServerTime: item.ServerTime, Quality: quality, QualityReason: reason, EvidenceID: item.ID, Values: values})
}
return rows, page.Total, nil
case "mileage":
@@ -2437,17 +2435,83 @@ func (s *Service) historyDataForVehicle(ctx context.Context, category, keyword s
}
rows := make([]HistoryDataRow, 0, len(page.Items))
for index, item := range page.Items {
quality := "normal"
if item.AnomalySeverity != "" {
quality = item.AnomalySeverity
}
rows = append(rows, HistoryDataRow{ID: "mileage-" + item.VIN + "-" + strconv.Itoa(index) + "-" + item.Date, VIN: item.VIN, Plate: item.Plate, Protocol: item.Source, DeviceTime: item.Date, ServerTime: item.Date, Quality: quality, Values: map[string]any{"startMileageKm": item.StartMileageKm, "endMileageKm": item.EndMileageKm, "dailyMileageKm": item.DailyMileageKm}})
quality, reason := historyMileageQuality(item)
rows = append(rows, HistoryDataRow{ID: "mileage-" + item.VIN + "-" + strconv.Itoa(index) + "-" + item.Date, VIN: item.VIN, Plate: item.Plate, Protocol: item.Source, DeviceTime: item.Date, ServerTime: item.Date, Quality: quality, QualityReason: reason, Values: map[string]any{"startMileageKm": item.StartMileageKm, "endMileageKm": item.EndMileageKm, "dailyMileageKm": item.DailyMileageKm}})
}
return rows, page.Total, nil
}
return nil, 0, nil
}
func historyLocationQuality(item HistoryLocationRow) (string, string) {
if !validTrackCoordinate(item) {
return "warning", "坐标超出有效范围或为无效零点"
}
if item.SpeedKmh < 0 || item.SpeedKmh > 200 {
return "warning", fmt.Sprintf("速度 %.2f km/h 超出 0–200 km/h 基础校验范围", item.SpeedKmh)
}
if item.TotalMileageKm < 0 {
return "warning", "总里程为负值"
}
deviceTime, deviceOK := parseVehicleServiceTime(item.DeviceTime)
serverTime, serverOK := parseVehicleServiceTime(item.ServerTime)
if !deviceOK || !serverOK {
return "warning", "设备时间或服务接收时间无法解析"
}
delaySeconds := int64(math.Abs(serverTime.Sub(deviceTime).Seconds()))
if delaySeconds > int64(latestTelemetryStaleAfter/time.Second) {
return "warning", fmt.Sprintf("设备时间与服务接收时间相差 %d 秒，超过 5 分钟", delaySeconds)
}
return "normal", "坐标、速度、里程及设备/服务时间通过基础校验"
}
func historyRawQuality(item RawFrameRow) (string, string) {
status := strings.TrimSpace(item.ParseStatus)
if status != "" && !strings.EqualFold(status, "ok") && !strings.EqualFold(status, "success") && !strings.EqualFold(status, "parsed") {
reason := "原始报文解析状态为 " + status
if detail := strings.TrimSpace(item.ParseError); detail != "" {
reason += "：" + detail
}
return "warning", reason
}
deviceTime, deviceOK := parseVehicleServiceTime(item.DeviceTime)
serverTime, serverOK := parseVehicleServiceTime(item.ServerTime)
if !deviceOK || !serverOK {
return "warning", "设备时间或服务接收时间无法解析"
}
delaySeconds := int64(math.Abs(serverTime.Sub(deviceTime).Seconds()))
if delaySeconds > int64(latestTelemetryStaleAfter/time.Second) {
return "warning", fmt.Sprintf("设备时间与服务接收时间相差 %d 秒，超过 5 分钟", delaySeconds)
}
return "normal", "报文解析成功，设备时间与服务接收时间关系正常"
}
func historyMileageQuality(item DailyMileageRow) (string, string) {
severity := strings.ToLower(strings.TrimSpace(item.AnomalySeverity))
if severity != "" && severity != "normal" && severity != "good" && severity != "ok" {
quality := "warning"
if strings.Contains(severity, "error") || strings.Contains(severity, "critical") || strings.Contains(severity, "严重") {
quality = "error"
}
return quality, "里程聚合记录标记异常：" + item.AnomalySeverity
}
if item.StartMileageKm < 0 || item.EndMileageKm < 0 || item.DailyMileageKm < 0 {
return "warning", "起始、结束或日里程存在负值"
}
if item.EndMileageKm < item.StartMileageKm {
return "warning", "结束里程小于起始里程"
}
calculated := item.EndMileageKm - item.StartMileageKm
tolerance := math.Max(1, calculated*0.05)
if math.Abs(calculated-item.DailyMileageKm) > tolerance {
return "warning", fmt.Sprintf("日里程 %.2f km 与起止里程差 %.2f km 不一致", item.DailyMileageKm, calculated)
}
if item.DailyMileageKm > 1500 {
return "warning", fmt.Sprintf("单日里程 %.2f km 超过 1500 km 基础校验范围", item.DailyMileageKm)
}
return "normal", "起止里程与日里程关系通过基础校验"
}
func copyHistoryFilters(target, source url.Values, protocol string) {
if protocol != "" {
target.Set("protocol", protocol)
@@ -2931,7 +2995,7 @@ func writeHistoryExportMetadata(writer *csv.Writer, request HistoryExportRequest
rows := [][]string{
{"导出元数据", "查询开始", request.DateFrom, "查询结束", request.DateTo, "车辆", strings.Join(request.Keywords, "、"), "数据类型", request.Category, "协议", firstNonEmpty(request.Protocol, "全部")},
{"指标与单位", strings.Join(exportMetricLabels(columns), "；")},
append([]string{"设备时间", "服务时间", "车牌", "VIN", "协议", "数据质量", "证据ID"}, exportMetricLabels(columns)...),
append([]string{"设备时间", "服务时间", "车牌", "VIN", "协议", "数据质量", "质量原因", "证据ID"}, exportMetricLabels(columns)...),
}
for _, row := range rows {
if err := writer.Write(row); err != nil {
@@ -2942,7 +3006,7 @@ func writeHistoryExportMetadata(writer *csv.Writer, request HistoryExportRequest
}
func historyExportRecord(row HistoryDataRow, columns []HistoryMetricDefinition) []string {
record := []string{row.DeviceTime, row.ServerTime, row.Plate, row.VIN, row.Protocol, row.Quality, row.EvidenceID}
record := []string{row.DeviceTime, row.ServerTime, row.Plate, row.VIN, row.Protocol, row.Quality, row.QualityReason, row.EvidenceID}
for _, column := range columns {
value, ok := row.Values[column.Key]
if !ok || value == nil {

36
vehicle-data-platform/apps/api/internal/platform/service_test.go
View File

@@ -580,6 +580,9 @@ func TestHistoryExportRunsAsControlledAsyncJob(t *testing.T) {
if len(body) < 3 || string(body[:3]) != "\xEF\xBB\xBF" {
t.Fatalf("CSV should include UTF-8 BOM")
}
if !strings.Contains(string(body), "数据质量,质量原因,证据ID") {
t.Fatalf("CSV should explain platform quality: %q", body[:min(len(body), 512)])
}
return
}
time.Sleep(10 * time.Millisecond)
@@ -671,6 +674,39 @@ func TestRawMetricMetadataKeepsProtocolAndUnit(t *testing.T) {
}
}
func TestHistoryQualityReasonsExplainLocationRawAndMileageWarnings(t *testing.T) {
quality, reason := historyLocationQuality(HistoryLocationRow{Longitude: 0, Latitude: 0, SpeedKmh: 30, DeviceTime: "2026-07-14 09:29:58", ServerTime: "2026-07-14 09:29:59"})
if quality != "warning" || !strings.Contains(reason, "坐标") {
t.Fatalf("invalid coordinate should be explained: %s %s", quality, reason)
}
quality, reason = historyRawQuality(RawFrameRow{DeviceTime: "2026-07-14 09:29:58", ServerTime: "2026-07-14 09:29:59", ParseStatus: "failed", ParseError: "checksum"})
if quality != "warning" || !strings.Contains(reason, "checksum") {
t.Fatalf("parse failure should retain detail: %s %s", quality, reason)
}
quality, reason = historyMileageQuality(DailyMileageRow{StartMileageKm: 100, EndMileageKm: 130, DailyMileageKm: 12})
if quality != "warning" || !strings.Contains(reason, "不一致") {
t.Fatalf("mileage mismatch should be explained: %s %s", quality, reason)
}
}
func TestMergeDiscoveredRawMetricsScansFetchedScope(t *testing.T) {
rows := []HistoryDataRow{
{Values: map[string]any{"frameType": "realtime"}},
{Values: map[string]any{"gb32960.vehicle.soc_percent": 88}},
}
columns := mergeDiscoveredRawMetrics(historyRawMetrics(), rows)
found := false
for _, column := range columns {
if column.Key == "gb32960.vehicle.soc_percent" {
found = true
break
}
}
if !found {
t.Fatalf("discovered field outside the visible page should remain selectable: %+v", columns)
}
}
func TestBuildLatestTelemetryResponseUsesCatalogAndNewestSourceEvidence(t *testing.T) {
now := time.Date(2026, 7, 14, 9, 30, 0, 0, time.FixedZone("Asia/Shanghai", 8*60*60))
definitions := []MetricDefinition{