From b948a46e649d5ba7011e568f1f201d627a7af712 Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 11:20:17 +0800 Subject: [PATCH] fix(go): normalize gb32960 realtime vendor fragments --- .../internal/realtime/repository.go | 56 +++++++++++ .../internal/realtime/repository_test.go | 71 ++++++++++++++ .../internal/realtime/snapshot_writer_test.go | 97 +++++++++++++++++++ 3 files changed, 224 insertions(+) diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index cc535194..fdf75db8 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -294,6 +294,9 @@ func mergeParsedForProtocol(protocol envelope.Protocol, existing map[string]any, for _, key := range replaceParsedKeys(protocol, incoming) { out[key] = cloneAny(incoming[key]) } + if protocol == envelope.ProtocolGB32960 { + normalizeGB32960RealtimeParsed(out) + } return out } @@ -404,6 +407,59 @@ func strconvAny(value any) string { } } +func normalizeGB32960RealtimeParsed(parsed map[string]any) { + units, ok := asAnySlice(parsed["data_units"]) + if !ok { + return + } + for _, unit := range units { + unitMap, ok := unit.(map[string]any) + if !ok || strings.TrimSpace(strconvAny(unitMap["name"])) != "gd_fc_stack" { + continue + } + value, ok := unitMap["value"].(map[string]any) + if !ok { + continue + } + collapseGDFCStackValue(value) + } +} + +func collapseGDFCStackValue(value map[string]any) { + summaries, ok := asAnySlice(value["summaries"]) + if !ok || len(summaries) == 0 { + return + } + collapsed := map[string]any{} + for _, item := range summaries { + summary, ok := item.(map[string]any) + if !ok { + continue + } + for key, fieldValue := range summary { + if isGDFCStackFragmentField(key) { + continue + } + collapsed[key] = cloneAny(fieldValue) + } + } + if len(collapsed) == 0 { + delete(value, "summaries") + return + } + value["stack_count"] = 1 + value["summaries"] = []any{collapsed} +} + +func isGDFCStackFragmentField(key string) bool { + switch key { + case "frame_cell_start", "frame_cell_count", "frame_max_cell_voltage_v", "frame_min_cell_voltage_v": + return true + default: + return false + } +} + func fieldTimes(fields map[string]any, eventMS int64) map[string]int64 { out := make(map[string]int64, len(fields)) for key := range fields { diff --git a/go/vehicle-gateway/internal/realtime/repository_test.go b/go/vehicle-gateway/internal/realtime/repository_test.go index 92f44beb..2e5ccf33 100644 --- a/go/vehicle-gateway/internal/realtime/repository_test.go +++ b/go/vehicle-gateway/internal/realtime/repository_test.go @@ -200,6 +200,77 @@ func TestRepositoryMergesNestedGB32960MotorSlicesBySerialNo(t *testing.T) { } } +func TestRepositoryCollapsesGB32960VendorStackFragmentsIntoSingleRealtimeSummary(t *testing.T) { + repo, closeFn := newTestRepository(t) + defer closeFn() + ctx := context.Background() + + first := map[string]any{ + "engine_work_state": 1, + "stack_water_outlet_temp_c": 65, + "hydrogen_inlet_pressure_kpa": 130, + "cell_count": 432, + "avg_cell_voltage_v": 0.78, + "frame_cell_start": 201, + "frame_cell_count": 200, + "frame_max_cell_voltage_v": 1.0, + "frame_min_cell_voltage_v": 0.78, + } + second := map[string]any{ + "engine_work_state": 1, + "stack_water_outlet_temp_c": 63, + "hydrogen_inlet_pressure_kpa": 130, + "cell_count": 432, + "avg_cell_voltage_v": 0.80, + "frame_cell_start": 401, + "frame_cell_count": 32, + "frame_max_cell_voltage_v": 1.0, + "frame_min_cell_voltage_v": 1.0, + } + for i, summary := range []map[string]any{first, second} { + if err := repo.Update(ctx, envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + VIN: "VIN001", + EventTimeMS: int64(1000 + i), + ReceivedAtMS: int64(1100 + i), + Parsed: map[string]any{ + "data_units": []any{ + map[string]any{ + "type": "0x30", + "name": "gd_fc_stack", + "value": map[string]any{ + "stack_count": 1, + "summaries": []any{summary}, + }, + }, + }, + }, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + } + + realtimeRaw, err := repo.GetRealtimeRaw(ctx, "VIN001", envelope.ProtocolGB32960) + if err != nil { + t.Fatalf("GetRealtimeRaw() error = %v", err) + } + unit := realtimeRaw["data_units"].([]any)[0].(map[string]any) + value := unit["value"].(map[string]any) + summaries := value["summaries"].([]any) + if len(summaries) != 1 { + t.Fatalf("stack summaries should be a current stack state, not appended fragments: %#v", summaries) + } + summary := summaries[0].(map[string]any) + for _, fragmentOnly := range []string{"frame_cell_start", "frame_cell_count", "frame_max_cell_voltage_v", "frame_min_cell_voltage_v"} { + if _, ok := summary[fragmentOnly]; ok { + t.Fatalf("stack summary should not expose fragment-only field %s: %#v", fragmentOnly, summary) + } + } + if summary["stack_water_outlet_temp_c"] != float64(63) { + t.Fatalf("summary should keep latest stack-level values: %#v", summary) + } +} + func TestRepositoryReplacesJT808LocationWhenUpdatingRealtimeRaw(t *testing.T) { repo, closeFn := newTestRepository(t) defer closeFn() diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 8f793ef1..86e0fcd5 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -247,6 +247,61 @@ func TestSnapshotWriterMergesGB32960ParsedJSONAcrossSplitRealtimeFrames(t *testi } } +func TestSnapshotWriterCollapsesGB32960StackFragmentsWhenMergingParsedJSON(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatal(err) + } + defer db.Close() + writer := NewSnapshotWriter(db) + + existing := `{"data_units":[{"type":"0x30","name":"gd_fc_stack","value":{"stack_count":1,"summaries":[{"cell_count":432,"stack_water_outlet_temp_c":65,"frame_cell_start":201,"frame_cell_count":200}]}}]}` + mock.ExpectQuery("SELECT parsed_json FROM vehicle_realtime_snapshot WHERE protocol = \\? AND vin = \\?"). + WithArgs("GB32960", "VIN001"). + WillReturnRows(sqlmock.NewRows([]string{"parsed_json"}).AddRow(existing)) + mock.ExpectExec("INSERT INTO vehicle_realtime_snapshot"). + WithArgs( + "GB32960", + "VIN001", + "", + "", + "", + jsonGDFCStackSummaryArg{}, + sqlmock.AnyArg(), + sqlmock.AnyArg(), + sqlmock.AnyArg(), + ). + WillReturnResult(sqlmock.NewResult(0, 1)) + + err = writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + MessageID: "0x02", + VIN: "VIN001", + EventTimeMS: 1782918600000, + ReceivedAtMS: 1782918601000, + Parsed: map[string]any{ + "data_units": []any{ + map[string]any{ + "type": "0x30", + "name": "gd_fc_stack", + "value": map[string]any{ + "stack_count": 1, + "summaries": []any{ + map[string]any{"cell_count": 432, "stack_water_outlet_temp_c": 63, "frame_cell_start": 401, "frame_cell_count": 32}, + }, + }, + }, + }, + }, + }) + if err != nil { + t.Fatalf("Update() error = %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + func TestRealtimeUpsertsDoNotOverwriteWithOlderEventTime(t *testing.T) { for _, tc := range []struct { name string @@ -641,6 +696,48 @@ type jsonDataUnitsArg struct { types []string } +type jsonGDFCStackSummaryArg struct{} + +func (jsonGDFCStackSummaryArg) Match(value driver.Value) bool { + text, ok := value.(string) + if !ok { + return false + } + var parsed map[string]any + if err := json.Unmarshal([]byte(text), &parsed); err != nil { + return false + } + units, ok := parsed["data_units"].([]any) + if !ok { + return false + } + for _, unit := range units { + unitMap, ok := unit.(map[string]any) + if !ok || unitMap["name"] != "gd_fc_stack" { + continue + } + valueMap, ok := unitMap["value"].(map[string]any) + if !ok { + return false + } + summaries, ok := valueMap["summaries"].([]any) + if !ok || len(summaries) != 1 { + return false + } + summary, ok := summaries[0].(map[string]any) + if !ok { + return false + } + for _, fragmentOnly := range []string{"frame_cell_start", "frame_cell_count", "frame_max_cell_voltage_v", "frame_min_cell_voltage_v"} { + if _, ok := summary[fragmentOnly]; ok { + return false + } + } + return summary["stack_water_outlet_temp_c"] == float64(63) + } + return false +} + func (a jsonDataUnitsArg) Match(value driver.Value) bool { text, ok := value.(string) if !ok {