From e1bf2d72e165afac246979115066cd9673af4b07 Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 13:51:57 +0800 Subject: [PATCH] refactor(go): map realtime fields to semantic names --- go/vehicle-gateway/internal/realtime/kv.go | 238 ++++++++++++++++-- .../internal/realtime/kv_test.go | 36 +-- .../internal/realtime/repository.go | 73 +++++- .../internal/realtime/repository_test.go | 10 +- 4 files changed, 306 insertions(+), 51 deletions(-) diff --git a/go/vehicle-gateway/internal/realtime/kv.go b/go/vehicle-gateway/internal/realtime/kv.go index 5bc12d72..e35a3af3 100644 --- a/go/vehicle-gateway/internal/realtime/kv.go +++ b/go/vehicle-gateway/internal/realtime/kv.go @@ -22,15 +22,64 @@ type RealtimeKVField struct { EventID string } +type protocolFieldMapping struct { + Prefix string + GBDataUnits bool + TopLevelName map[string]string + ArrayName map[string]string +} + +var realtimeFieldMappings = map[envelope.Protocol]protocolFieldMapping{ + envelope.ProtocolGB32960: { + Prefix: "gb32960", + GBDataUnits: true, + TopLevelName: map[string]string{ + "header": "header", + "platform_name": "platform", + "platform_login": "platform_login", + }, + ArrayName: map[string]string{ + "motors": "motor", + "probes": "probe", + "subsystems": "subsystem", + "summaries": "", + }, + }, + envelope.ProtocolJT808: { + Prefix: "jt808", + TopLevelName: map[string]string{ + "header": "header", + "location": "location", + "registration": "registration", + "authentication": "authentication", + }, + ArrayName: map[string]string{ + "additional": "additional", + }, + }, + envelope.ProtocolYutongMQTT: { + Prefix: "yutong_mqtt", + TopLevelName: map[string]string{ + "data": "data", + "root": "root", + "endpoint": "metadata", + "topic": "metadata", + }, + }, +} + +const realtimeFieldMappingVersion = "2026-07-03.v1" + func realtimeKVFields(env envelope.FrameEnvelope, parsed map[string]any) []RealtimeKVField { vin := strings.TrimSpace(env.VIN) if vin == "" { return nil } eventID := env.StableEventID() + mapping := realtimeMapping(env.Protocol) add := func(rows []RealtimeKVField, domain string, fields any) []RealtimeKVField { flat := map[string]any{} - flattenKV("", fields, flat) + flattenKV("", fields, flat, mapping) names := make([]string, 0, len(flat)) for name := range flat { names = append(names, name) @@ -57,36 +106,92 @@ func realtimeKVFields(env envelope.FrameEnvelope, parsed map[string]any) []Realt } var rows []RealtimeKVField + if mapping.GBDataUnits { + rows = append(rows, gb32960KVFields(env, parsed, mapping, eventID)...) + return rows + } + keys := make([]string, 0, len(parsed)) for key := range parsed { keys = append(keys, key) } sort.Strings(keys) - metadata := map[string]any{} for _, key := range keys { - domain := strings.TrimSpace(key) + domain := mappedTopLevelDomain(mapping, key) if domain == "" { continue } - value := parsed[key] - switch typed := value.(type) { - case map[string]any: - rows = add(rows, domain, typed) - case []any: - rows = add(rows, domain, typed) - case []map[string]any: - rows = add(rows, domain, typed) - default: - metadata[domain] = value - } - } - if len(metadata) > 0 { - rows = add(rows, "metadata", metadata) + rows = addMappedValue(rows, add, domain, key, parsed[key]) } return rows } -func flattenKV(prefix string, value any, out map[string]any) { +func gb32960KVFields(env envelope.FrameEnvelope, parsed map[string]any, mapping protocolFieldMapping, eventID string) []RealtimeKVField { + addRows := func(rows []RealtimeKVField, domain string, fields any) []RealtimeKVField { + flat := map[string]any{} + flattenKV("", fields, flat, mapping) + names := make([]string, 0, len(flat)) + for name := range flat { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + value, valueType, ok := stringifyKVValue(flat[name]) + if !ok { + continue + } + rows = append(rows, RealtimeKVField{ + Protocol: env.Protocol, + VIN: strings.TrimSpace(env.VIN), + Domain: domain, + Field: name, + Value: value, + ValueType: valueType, + EventTimeMS: env.EventTimeMS, + ReceivedAtMS: env.ReceivedAtMS, + EventID: eventID, + }) + } + return rows + } + + var rows []RealtimeKVField + for key, value := range parsed { + if key == "data_units" { + continue + } + if domain := mappedTopLevelDomain(mapping, key); domain != "" { + rows = addMappedValue(rows, addRows, domain, key, value) + } + } + units, ok := asAnySlice(parsed["data_units"]) + if !ok { + return rows + } + for _, unit := range units { + unitMap, ok := unit.(map[string]any) + if !ok { + continue + } + name := sanitizeFieldPart(strconvAny(unitMap["name"])) + if name == "" || name == "" { + name = strings.ToLower(strings.TrimPrefix(strings.TrimSpace(strconvAny(unitMap["type"])), "0x")) + } + if name == "" { + continue + } + domain := mapping.Prefix + "." + name + if value, ok := unitMap["value"]; ok { + rows = addRows(rows, domain, value) + } + if unitType := strings.TrimSpace(strconvAny(unitMap["type"])); unitType != "" && unitType != "" { + rows = addRows(rows, domain, map[string]any{"type": unitType}) + } + } + return rows +} + +func flattenKV(prefix string, value any, out map[string]any, mapping protocolFieldMapping) { switch typed := value.(type) { case map[string]any: keys := make([]string, 0, len(typed)) @@ -95,11 +200,21 @@ func flattenKV(prefix string, value any, out map[string]any) { } sort.Strings(keys) for _, key := range keys { - next := key - if prefix != "" { - next = prefix + "." + key + part := sanitizeFieldPart(key) + if part == "" { + continue } - flattenKV(next, typed[key], out) + if arrayName, ok := mapping.ArrayName[part]; ok && arrayName == "" { + if items, ok := asAnySlice(typed[key]); ok && len(items) == 1 { + flattenKV(prefix, items[0], out, mapping) + continue + } + } + next := part + if prefix != "" { + next = prefix + "." + part + } + flattenKV(next, typed[key], out, mapping) } case []any: for index, item := range typed { @@ -108,19 +223,29 @@ func flattenKV(prefix string, value any, out map[string]any) { out[prefix] = typed return } + arrayName, hasArrayName := mapping.ArrayName[lastPathPart(prefix)] itemKey := strconv.Itoa(index) + if hasArrayName { + if arrayName == "" && len(typed) == 1 { + flattenKV(prefix, itemMap, out, mapping) + continue + } + if arrayName != "" { + itemKey = arrayName + "_" + strconv.Itoa(index+1) + } + } next := itemKey if prefix != "" { next = prefix + "." + itemKey } - flattenKV(next, itemMap, out) + flattenKV(next, itemMap, out, mapping) } case []map[string]any: items := make([]any, len(typed)) for index, item := range typed { items[index] = item } - flattenKV(prefix, items, out) + flattenKV(prefix, items, out, mapping) default: if prefix != "" && value != nil { out[prefix] = value @@ -128,6 +253,71 @@ func flattenKV(prefix string, value any, out map[string]any) { } } +func addMappedValue(rows []RealtimeKVField, add func([]RealtimeKVField, string, any) []RealtimeKVField, domain string, key string, value any) []RealtimeKVField { + switch typed := value.(type) { + case map[string]any: + return add(rows, domain, typed) + case []any: + return add(rows, domain, typed) + case []map[string]any: + return add(rows, domain, typed) + default: + return add(rows, domain, map[string]any{key: value}) + } +} + +func realtimeMapping(protocol envelope.Protocol) protocolFieldMapping { + if mapping, ok := realtimeFieldMappings[protocol]; ok { + return mapping + } + return protocolFieldMapping{Prefix: strings.ToLower(string(protocol))} +} + +func mappedTopLevelDomain(mapping protocolFieldMapping, key string) string { + key = strings.TrimSpace(key) + if key == "" { + return "" + } + name := mapping.TopLevelName[key] + if name == "" { + name = sanitizeFieldPart(key) + } + if name == "" { + return "" + } + return mapping.Prefix + "." + name +} + +func sanitizeFieldPart(value string) string { + value = strings.TrimSpace(value) + if value == "" { + return "" + } + var b strings.Builder + for _, r := range value { + switch { + case r >= 'a' && r <= 'z': + b.WriteRune(r) + case r >= 'A' && r <= 'Z': + b.WriteRune(r + ('a' - 'A')) + case r >= '0' && r <= '9': + b.WriteRune(r) + case r == '_': + b.WriteRune(r) + default: + b.WriteByte('_') + } + } + return strings.Trim(b.String(), "_") +} + +func lastPathPart(path string) string { + if index := strings.LastIndex(path, "."); index >= 0 { + return path[index+1:] + } + return path +} + func stringifyKVValue(value any) (string, string, bool) { switch typed := value.(type) { case string: diff --git a/go/vehicle-gateway/internal/realtime/kv_test.go b/go/vehicle-gateway/internal/realtime/kv_test.go index fe6793d5..94cb6ef3 100644 --- a/go/vehicle-gateway/internal/realtime/kv_test.go +++ b/go/vehicle-gateway/internal/realtime/kv_test.go @@ -27,14 +27,14 @@ func TestRealtimeKVFieldsFromGB32960ParsedDomains(t *testing.T) { }) values := kvMap(rows) - if values["header/vin"] != "VIN001" || values["header/command"] != "0x02" { + if values["gb32960.header/vin"] != "VIN001" || values["gb32960.header/command"] != "0x02" { t.Fatalf("gb32960 header kv missing: %#v", values) } - if values["data_units/0.name"] != "vehicle" || values["data_units/0.type"] != "0x01" || values["data_units/0.value.soc_percent"] != "88" { + if values["gb32960.vehicle/type"] != "0x01" || values["gb32960.vehicle/soc_percent"] != "88" { t.Fatalf("vehicle soc kv missing: %#v", values) } - if values["data_units/1.value.summaries.0.stack_water_outlet_temp_c"] != "63" || - values["data_units/1.value.summaries.0.hydrogen_inlet_pressure_kpa"] != "130" { + if values["gb32960.gd_fc_stack/stack_water_outlet_temp_c"] != "63" || + values["gb32960.gd_fc_stack/hydrogen_inlet_pressure_kpa"] != "130" { t.Fatalf("stack pressure kv missing: %#v", values) } } @@ -58,15 +58,15 @@ func TestRealtimeKVFieldsFromJT808ParsedFields(t *testing.T) { }, }) jtValues := kvMap(jtRows) - if jtValues["header/message_id"] != "0x0200" || jtValues["header/phone"] != "13307795425" { + if jtValues["jt808.header/message_id"] != "0x0200" || jtValues["jt808.header/phone"] != "13307795425" { t.Fatalf("jt808 header kv missing: %#v", jtValues) } - if jtValues["location/total_mileage_km"] != "10241.2" || - jtValues["location/longitude"] != "121.1" || - jtValues["location/additional.0.id"] != "0x01" { + if jtValues["jt808.location/total_mileage_km"] != "10241.2" || + jtValues["jt808.location/longitude"] != "121.1" || + jtValues["jt808.location/additional.additional_1.id"] != "0x01" { t.Fatalf("jt808 location kv missing: %#v", jtValues) } - if jtValues["location/soc_percent"] != "" { + if jtValues["jt808.location/soc_percent"] != "" { t.Fatalf("jt808 kv should not include standardized env.Fields-only values: %#v", jtValues) } } @@ -96,19 +96,19 @@ func TestRealtimeKVFieldsFromYutongMQTTParsedFields(t *testing.T) { }, }) mqttValues := kvMap(mqttRows) - if mqttValues["data/ACC_PEDAL_APT"] != "14" || - mqttValues["data/CURRENT_OF_FC"] != "54.9" || - mqttValues["data/HYDROGEN_LOW_PRESSURE"] != "1.41" || - mqttValues["data/TOTAL_MILEAGE"] != "119925000" || - mqttValues["data/fuelCellCoolInTempt"] != "59" { + if mqttValues["yutong_mqtt.data/acc_pedal_apt"] != "14" || + mqttValues["yutong_mqtt.data/current_of_fc"] != "54.9" || + mqttValues["yutong_mqtt.data/hydrogen_low_pressure"] != "1.41" || + mqttValues["yutong_mqtt.data/total_mileage"] != "119925000" || + mqttValues["yutong_mqtt.data/fuelcellcoolintempt"] != "59" { t.Fatalf("mqtt raw data kv missing: %#v", mqttValues) } - if mqttValues["root/device"] != "LMRKH9AC2R1004087" || - mqttValues["metadata/endpoint"] != "yutong" || - mqttValues["metadata/topic"] != "/ytforward/shln/3" { + if mqttValues["yutong_mqtt.root/device"] != "LMRKH9AC2R1004087" || + mqttValues["yutong_mqtt.metadata/endpoint"] != "yutong" || + mqttValues["yutong_mqtt.metadata/topic"] != "/ytforward/shln/3" { t.Fatalf("mqtt metadata kv missing: %#v", mqttValues) } - if mqttValues["data/soc_percent"] != "" || mqttValues["data/gear"] != "" { + if mqttValues["yutong_mqtt.data/soc_percent"] != "" || mqttValues["yutong_mqtt.data/gear"] != "" { t.Fatalf("mqtt kv should not include standardized env.Fields-only values: %#v", mqttValues) } } diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index 124d51ee..f15ba6cb 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -36,10 +36,7 @@ func (r *Repository) FastUpdate(ctx context.Context, env envelope.FrameEnvelope) if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") { return nil } - if err := r.setKV(ctx, vin, env, env.Parsed); err != nil { - return err - } - return r.setOnline(ctx, vehicleKey, vin, env.Protocol, env.ReceivedAtMS, env.SourceEndpoint) + return r.setFastProjection(ctx, vehicleKey, vin, env) } func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) error { @@ -234,6 +231,7 @@ func (r *Repository) setKV(ctx context.Context, vin string, env envelope.FrameEn "protocol": string(env.Protocol), "vin": vin, "source_endpoint": env.SourceEndpoint, + "field_mapping": realtimeFieldMappingVersion, } pipe := r.client.Pipeline() pipe.HSet(ctx, realtimeKVValuesKey(env.Protocol, vin), values) @@ -243,6 +241,73 @@ func (r *Repository) setKV(ctx context.Context, vin string, env envelope.FrameEn return err } +func (r *Repository) setFastProjection(ctx context.Context, vehicleKey string, vin string, env envelope.FrameEnvelope) error { + rows := realtimeKVFields(env, env.Parsed) + values, types := realtimeKVMaps(rows) + eventTimeMS := eventTimeOrReceivedMS(env) + offlineAfterMS := env.ReceivedAtMS + r.cfg.ttl().Milliseconds() + online := OnlineStatus{ + VehicleKey: vehicleKey, + VIN: vin, + Protocol: env.Protocol, + Online: true, + LastSeenMS: env.ReceivedAtMS, + OfflineAfterMS: offlineAfterMS, + SourceEndpoint: env.SourceEndpoint, + Protocols: []envelope.Protocol{env.Protocol}, + TTLSeconds: int64(r.cfg.ttl().Seconds()), + } + payload, err := json.Marshal(online) + if err != nil { + return err + } + meta := map[string]any{ + "event_time_ms": strconv.FormatInt(eventTimeMS, 10), + "received_at_ms": strconv.FormatInt(env.ReceivedAtMS, 10), + "event_id": env.StableEventID(), + "protocol": string(env.Protocol), + "vin": vin, + "source_endpoint": env.SourceEndpoint, + "field_mapping": realtimeFieldMappingVersion, + } + state := map[string]any{ + "vehicle_key": vehicleKey, + "vin": vin, + "protocol": string(env.Protocol), + "online": strconv.FormatBool(true), + "last_seen_ms": strconv.FormatInt(env.ReceivedAtMS, 10), + "offline_after_ms": strconv.FormatInt(offlineAfterMS, 10), + "ttl_seconds": strconv.FormatInt(int64(r.cfg.ttl().Seconds()), 10), + "source_endpoint": env.SourceEndpoint, + } + pipe := r.client.Pipeline() + if len(values) > 0 { + pipe.HSet(ctx, realtimeKVValuesKey(env.Protocol, vin), values) + pipe.HSet(ctx, realtimeKVTypesKey(env.Protocol, vin), types) + pipe.HSet(ctx, realtimeKVMetaKey(env.Protocol, vin), meta) + } + pipe.Set(ctx, onlineKey(env.Protocol, vin), payload, r.cfg.ttl()) + pipe.HSet(ctx, onlineStateKey(env.Protocol, vin), state) + pipe.ZAdd(ctx, "vehicle:last_seen", redis.Z{Score: float64(env.ReceivedAtMS), Member: onlineMember(env.Protocol, vin)}) + pipe.SAdd(ctx, realtimeIndexKey(env.Protocol), vin) + _, err = pipe.Exec(ctx) + return err +} + +func realtimeKVMaps(rows []RealtimeKVField) (map[string]any, map[string]any) { + values := make(map[string]any, len(rows)) + types := make(map[string]any, len(rows)) + for _, row := range rows { + field := realtimeKVFieldPath(row.Domain, row.Field) + if field == "" { + continue + } + values[field] = row.Value + types[field] = row.ValueType + } + return values, types +} + func (r *Repository) setOnline(ctx context.Context, vehicleKey string, vin string, protocol envelope.Protocol, lastSeenMS int64, sourceEndpoint string) error { if protocol == "" { return nil diff --git a/go/vehicle-gateway/internal/realtime/repository_test.go b/go/vehicle-gateway/internal/realtime/repository_test.go index fdee435c..793585e7 100644 --- a/go/vehicle-gateway/internal/realtime/repository_test.go +++ b/go/vehicle-gateway/internal/realtime/repository_test.go @@ -345,17 +345,17 @@ func TestRepositoryWritesRealtimeKVHashes(t *testing.T) { if err != nil { t.Fatalf("values kv HGetAll error = %v", err) } - if dataUnitsKV["data_units.0.value.soc_percent"] != "88" { + if dataUnitsKV["gb32960.vehicle.soc_percent"] != "88" { t.Fatalf("data_units kv = %#v", dataUnitsKV) } - if dataUnitsKV["data_units.1.value.summaries.0.stack_water_outlet_temp_c"] != "63" { + if dataUnitsKV["gb32960.gd_fc_stack.stack_water_outlet_temp_c"] != "63" { t.Fatalf("data_units stack kv = %#v", dataUnitsKV) } typeKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:types").Result() if err != nil { t.Fatalf("types kv HGetAll error = %v", err) } - if typeKV["data_units.0.value.soc_percent"] != "number" || typeKV["data_units.1.name"] != "string" { + if typeKV["gb32960.vehicle.soc_percent"] != "number" || typeKV["gb32960.gd_fc_stack.type"] != "string" { t.Fatalf("types kv = %#v", typeKV) } metaKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:meta").Result() @@ -396,14 +396,14 @@ func TestRepositoryFastUpdateOnlyWritesPermanentKVAndMinuteOnline(t *testing.T) if err != nil { t.Fatalf("values kv HGetAll error = %v", err) } - if dataUnitsKV["data_units.0.value.soc_percent"] != "88" { + if dataUnitsKV["gb32960.vehicle.soc_percent"] != "88" { t.Fatalf("data_units kv = %#v", dataUnitsKV) } typeKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:types").Result() if err != nil { t.Fatalf("types kv HGetAll error = %v", err) } - if typeKV["data_units.0.value.soc_percent"] != "number" { + if typeKV["gb32960.vehicle.soc_percent"] != "number" { t.Fatalf("types kv = %#v", typeKV) } if ttl := repo.client.TTL(ctx, "vehicle:rt-kv:GB32960:VIN001:values").Val(); ttl != -1 {