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

refactor(go): map realtime fields to semantic names

Browse Source
This commit is contained in:
lingniu
2026-07-03 13:51:57 +08:00
parent c359930009
commit e1bf2d72e1
4 changed files with 306 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

238
go/vehicle-gateway/internal/realtime/kv.go
View File

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

36
go/vehicle-gateway/internal/realtime/kv_test.go
View File

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

73
go/vehicle-gateway/internal/realtime/repository.go
View File

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

10
go/vehicle-gateway/internal/realtime/repository_test.go
View File

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