fix: align jt808 parser with protocol variants

go/vehicle-gateway/internal/protocol/jt808/parser.go

@@ -10,8 +10,8 @@ import (
 	"time"
 	"unicode/utf8"
 
-	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
 	"golang.org/x/text/encoding/simplifiedchinese"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
 )
 
 var (
@@ -97,16 +97,19 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope
 		phoneEnd = 15
 		serialStart = 15
 	}
-	if props&0x2000 != 0 {
+	subpackage := props&0x2000 != 0
+	if subpackage {
 		headerLen += 4
+	}
+	if len(raw) < headerLen+int(bodySize)+1 {
+		return envelope.FrameEnvelope{}, fmt.Errorf("%w: bodySize=%d len=%d", ErrFrameTooShort, bodySize, len(raw))
+	}
+	if subpackage {
 		packageInfo = map[string]any{
 			"total": binary.BigEndian.Uint16(raw[serialStart+2 : serialStart+4]),
 			"index": binary.BigEndian.Uint16(raw[serialStart+4 : serialStart+6]),
 		}
 	}
-	if len(raw) < headerLen+int(bodySize)+1 {
-		return envelope.FrameEnvelope{}, fmt.Errorf("%w: bodySize=%d len=%d", ErrFrameTooShort, bodySize, len(raw))
-	}
 
 	phoneBCD := raw[phoneStart:phoneEnd]
 	phone := bcdString(phoneBCD)
@@ -126,7 +129,7 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope
 			"phone":            phone,
 			"phone_trim_zero":  phoneTrimZero,
 			"sequence":         binary.BigEndian.Uint16(raw[serialStart : serialStart+2]),
-			"subpackage":       props&0x2000 != 0,
+			"subpackage":       subpackage,
 			"package_info":     packageInfo,
 			"encryption_flags": (props >> 10) & 0x07,
 		},
@@ -155,7 +158,7 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope
 			fields[envelope.FieldTotalMileageKM] = *location.TotalMileageKM
 		}
 	} else if messageID == 0x0100 {
-		registration, err := parseRegistration(body)
+		registration, err := parseRegistration(body, versioned)
 		if err != nil {
 			return envelope.FrameEnvelope{}, err
 		}
@@ -167,7 +170,7 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope
 			fields["plate"] = plate
 		}
 	} else if messageID == 0x0102 {
-		authentication := parseAuthentication(body)
+		authentication := parseAuthentication(body, versioned)
 		parsed["authentication"] = authentication
 		if token, ok := authentication["token"].(string); ok {
 			fields["auth_token"] = token
@@ -195,37 +198,107 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope
 	return env, nil
 }
 
-func parseRegistration(body []byte) (map[string]any, error) {
+func parseRegistration(body []byte, versioned bool) (map[string]any, error) {
-	if len(body) < 37 {
+	schemaVersion := "2011_2013"
+	manufacturerSize := 5
+	deviceTypeSize := 20
+	deviceIDSize := 7
+	if versioned {
+		schemaVersion = "2019"
+		manufacturerSize = 11
+		deviceTypeSize = 30
+		deviceIDSize = 30
+	}
+	minLen := 2 + 2 + manufacturerSize + deviceTypeSize + deviceIDSize + 1
+	if len(body) < minLen {
 		return nil, fmt.Errorf("%w: registration body len=%d", ErrFrameTooShort, len(body))
 	}
+	manufacturerStart := 4
+	deviceTypeStart := manufacturerStart + manufacturerSize
+	deviceIDStart := deviceTypeStart + deviceTypeSize
+	plateColorStart := deviceIDStart + deviceIDSize
 	registration := map[string]any{
-		"province":     binary.BigEndian.Uint16(body[0:2]),
+		"schema_version": schemaVersion,
-		"city":         binary.BigEndian.Uint16(body[2:4]),
+		"province":       binary.BigEndian.Uint16(body[0:2]),
-		"manufacturer": fixedText(body[4:9]),
+		"city":           binary.BigEndian.Uint16(body[2:4]),
-		"device_type":  fixedText(body[9:29]),
+		"manufacturer":   fixedText(body[manufacturerStart:deviceTypeStart]),
-		"device_id":    fixedText(body[29:36]),
+		"device_type":    fixedText(body[deviceTypeStart:deviceIDStart]),
-		"plate_color":  body[36],
+		"device_id":      fixedText(body[deviceIDStart:plateColorStart]),
-		"plate":        freeText(body[37:]),
+		"plate_color":    body[plateColorStart],
+		"plate":          freeText(body[plateColorStart+1:]),
 	}
 	return registration, nil
 }
 
-func parseAuthentication(body []byte) map[string]any {
+func parseAuthentication(body []byte, versioned bool) map[string]any {
-	token := freeText(body)
+	if versioned {
-	out := map[string]any{"token": token}
+		if out, ok := parseVersionedAuthentication(body); ok {
-	return out
+			return out
+		}
+		if out, ok := parseDelimitedVersionedAuthentication(body); ok {
+			return out
+		}
+	}
+	return map[string]any{"token": freeText(body)}
+}
+
+func parseVersionedAuthentication(body []byte) (map[string]any, bool) {
+	if len(body) < 17 {
+		return nil, false
+	}
+	tokenLen := int(body[0])
+	imeiStart := 1 + tokenLen
+	softwareLenStart := imeiStart + 15
+	if tokenLen <= 0 || len(body) < softwareLenStart+1 {
+		return nil, false
+	}
+	softwareLen := int(body[softwareLenStart])
+	softwareStart := softwareLenStart + 1
+	if len(body) < softwareStart+softwareLen {
+		return nil, false
+	}
+	return map[string]any{
+		"schema_version":   "2019",
+		"token":            freeText(body[1:imeiStart]),
+		"imei":             fixedText(body[imeiStart:softwareLenStart]),
+		"software_version": freeText(body[softwareStart : softwareStart+softwareLen]),
+	}, true
+}
+
+func parseDelimitedVersionedAuthentication(body []byte) (map[string]any, bool) {
+	parts := bytes.Split(body, []byte{0xff, 0xff})
+	if len(parts) < 2 {
+		return nil, false
+	}
+	out := map[string]any{
+		"schema_version": "2019_delimited",
+		"token":          freeText(parts[0]),
+		"imei":           freeText(parts[1]),
+	}
+	if len(parts) > 2 {
+		out["software_version"] = freeText(parts[2])
+	}
+	return out, true
 }
 
 func parseLocation(body []byte) (Location, error) {
 	if len(body) < 28 {
 		return Location{}, fmt.Errorf("%w: location body len=%d", ErrFrameTooShort, len(body))
 	}
+	status := binary.BigEndian.Uint32(body[4:8])
+	latitude := float64(binary.BigEndian.Uint32(body[8:12])) / 1_000_000
+	longitude := float64(binary.BigEndian.Uint32(body[12:16])) / 1_000_000
+	if status&(1<<2) != 0 {
+		latitude = -latitude
+	}
+	if status&(1<<3) != 0 {
+		longitude = -longitude
+	}
 	location := Location{
 		AlarmFlag:    binary.BigEndian.Uint32(body[0:4]),
-		StatusFlag:   binary.BigEndian.Uint32(body[4:8]),
+		StatusFlag:   status,
-		Latitude:     float64(binary.BigEndian.Uint32(body[8:12])) / 1_000_000,
+		Latitude:     latitude,
-		Longitude:    float64(binary.BigEndian.Uint32(body[12:16])) / 1_000_000,
+		Longitude:    longitude,
 		AltitudeM:    binary.BigEndian.Uint16(body[16:18]),
 		SpeedKMH:     float64(binary.BigEndian.Uint16(body[18:20])) / 10,
 		DirectionDeg: binary.BigEndian.Uint16(body[20:22]),
@@ -256,8 +329,8 @@ func parseAdditional(data []byte, location *Location) []AdditionalItem {
 			Length:   size,
 			ValueHex: strings.ToUpper(hex.EncodeToString(value)),
 		}
-		// JT/T 808-2011 table 27: id 0x01 is GPS total mileage, DWORD, 0.1 km.
+		switch {
-		if id == 0x01 && size == 4 {
+		case id == 0x01 && size == 4:
 			mileage := float64(binary.BigEndian.Uint32(value)) / 10
 			location.TotalMileageKM = &mileage
 			item.Parsed = map[string]any{
@@ -265,6 +338,57 @@ func parseAdditional(data []byte, location *Location) []AdditionalItem {
 				"value": mileage,
 				"unit":  "km",
 			}
+		case id == 0x02 && size == 2:
+			item.Parsed = map[string]any{
+				"name":  "fuel_l",
+				"value": float64(binary.BigEndian.Uint16(value)) / 10,
+				"unit":  "L",
+			}
+		case id == 0x03 && size == 2:
+			item.Parsed = map[string]any{
+				"name":  "recorder_speed_kmh",
+				"value": float64(binary.BigEndian.Uint16(value)) / 10,
+				"unit":  "km/h",
+			}
+		case id == 0x04 && size == 2:
+			item.Parsed = map[string]any{
+				"name":  "manual_alarm_event_id",
+				"value": binary.BigEndian.Uint16(value),
+			}
+		case id == 0x06 && size == 2:
+			item.Parsed = map[string]any{
+				"name":  "carriage_temperature_c",
+				"value": int16(binary.BigEndian.Uint16(value)),
+				"unit":  "C",
+			}
+		case id == 0x25 && size == 4:
+			item.Parsed = map[string]any{
+				"name":  "extended_vehicle_signal_status",
+				"value": binary.BigEndian.Uint32(value),
+			}
+		case id == 0x2a && size == 2:
+			item.Parsed = map[string]any{
+				"name":  "io_status",
+				"value": binary.BigEndian.Uint16(value),
+			}
+		case id == 0x2b && size == 4:
+			analog := binary.BigEndian.Uint32(value)
+			item.Parsed = map[string]any{
+				"name":  "analog",
+				"ad0":   analog & 0xffff,
+				"ad1":   analog >> 16,
+				"value": analog,
+			}
+		case id == 0x30 && size == 1:
+			item.Parsed = map[string]any{
+				"name":  "network_signal_strength",
+				"value": value[0],
+			}
+		case id == 0x31 && size == 1:
+			item.Parsed = map[string]any{
+				"name":  "gnss_satellite_count",
+				"value": value[0],
+			}
 		}
 		out = append(out, item)
 		data = data[size:]

go/vehicle-gateway/internal/protocol/jt808/parser_test.go

@@ -115,6 +115,36 @@ func TestParseFrameSupportsJT8082019VersionedHeader(t *testing.T) {
 	}
 }
 
+func TestParseFrameRejectsTruncatedSubpackageHeader(t *testing.T) {
+	payload := []byte{0x02, 0x00, 0x20, 0x00}
+	payload = append(payload, encodeBCD("013307795425", 6)...)
+	payload = binary.BigEndian.AppendUint16(payload, 1)
+	payload = append(payload, checksum(payload))
+
+	_, err := ParseFrame(payload, 1782914967713, "115.231.168.135:43625")
+	if err == nil {
+		t.Fatal("expected truncated subpackage header to fail")
+	}
+}
+
+func TestParseLocationAppliesSouthWestStatusBits(t *testing.T) {
+	body := make([]byte, 28)
+	binary.BigEndian.PutUint32(body[4:8], (1<<2)|(1<<3))
+	binary.BigEndian.PutUint32(body[8:12], 12345678)
+	binary.BigEndian.PutUint32(body[12:16], 98765432)
+
+	location, err := parseLocation(body)
+	if err != nil {
+		t.Fatalf("parseLocation() error = %v", err)
+	}
+	if location.Latitude != -12.345678 {
+		t.Fatalf("latitude = %v", location.Latitude)
+	}
+	if location.Longitude != -98.765432 {
+		t.Fatalf("longitude = %v", location.Longitude)
+	}
+}
+
 func TestParseFrameParsesRegistrationIdentity(t *testing.T) {
 	body := make([]byte, 0, 46)
 	body = binary.BigEndian.AppendUint16(body, 16)
@@ -153,6 +183,38 @@ func TestParseFrameParsesRegistrationIdentity(t *testing.T) {
 	}
 }
 
+func TestParseFrameParsesJT8082019RegistrationIdentity(t *testing.T) {
+	body := make([]byte, 0, 84)
+	body = binary.BigEndian.AppendUint16(body, 16)
+	body = binary.BigEndian.AppendUint16(body, 32)
+	body = appendFixedASCII(body, "YUTONG00001", 11)
+	body = appendFixedASCII(body, "ZK6105CHEVNPG4-2019", 30)
+	body = appendFixedASCII(body, "DEV201900000000000000000000001", 30)
+	body = append(body, 2)
+	body = append(body, []byte("TEST2019")...)
+
+	env, err := ParseFrame(buildVersionedFrame(0x0100, "14894135060", 11, body), 1782918600000, "115.231.168.135:43625")
+	if err != nil {
+		t.Fatalf("ParseFrame() error = %v", err)
+	}
+	if env.Phone != "14894135060" {
+		t.Fatalf("phone = %q", env.Phone)
+	}
+	if env.DeviceID != "DEV201900000000000000000000001" {
+		t.Fatalf("device id = %q", env.DeviceID)
+	}
+	if env.Plate != "TEST2019" {
+		t.Fatalf("plate = %q", env.Plate)
+	}
+	registration, ok := env.Parsed["registration"].(map[string]any)
+	if !ok {
+		t.Fatalf("registration missing: %#v", env.Parsed)
+	}
+	if registration["schema_version"] != "2019" || registration["manufacturer"] != "YUTONG00001" {
+		t.Fatalf("registration fields = %#v", registration)
+	}
+}
+
 func TestParseFrameDecodesGBKRegistrationPlate(t *testing.T) {
 	body := make([]byte, 0, 46)
 	body = binary.BigEndian.AppendUint16(body, 16)
@@ -172,6 +234,26 @@ func TestParseFrameDecodesGBKRegistrationPlate(t *testing.T) {
 	}
 }
 
+func TestParseFrameParsesJT8082019Authentication(t *testing.T) {
+	body := []byte{5}
+	body = append(body, []byte("TOKEN")...)
+	body = appendFixedASCII(body, "123456789012345", 15)
+	body = append(body, 4)
+	body = append(body, []byte("v1.2")...)
+
+	env, err := ParseFrame(buildVersionedFrame(0x0102, "14894135060", 12, body), 1782918600000, "115.231.168.135:43625")
+	if err != nil {
+		t.Fatalf("ParseFrame() error = %v", err)
+	}
+	auth, ok := env.Parsed["authentication"].(map[string]any)
+	if !ok {
+		t.Fatalf("authentication missing: %#v", env.Parsed)
+	}
+	if auth["token"] != "TOKEN" || auth["imei"] != "123456789012345" || auth["software_version"] != "v1.2" {
+		t.Fatalf("authentication = %#v", auth)
+	}
+}
+
 func TestParseFrameParsesAuthenticationToken(t *testing.T) {
 	env, err := ParseFrame(buildFrame(0x0102, "064646848757", 8, []byte("g7gps")), 1782918600000, "115.231.168.135:43625")
 	if err != nil {
@@ -217,6 +299,18 @@ func buildFrame(messageID uint16, phone string, sequence uint16, body []byte) []
 	return payload
 }
 
+func buildVersionedFrame(messageID uint16, phone string, sequence uint16, body []byte) []byte {
+	payload := make([]byte, 0, 17+len(body)+1)
+	payload = binary.BigEndian.AppendUint16(payload, messageID)
+	payload = binary.BigEndian.AppendUint16(payload, 0x4000|uint16(len(body)))
+	payload = append(payload, 1)
+	payload = append(payload, encodeBCD(phone, 10)...)
+	payload = binary.BigEndian.AppendUint16(payload, sequence)
+	payload = append(payload, body...)
+	payload = append(payload, checksum(payload))
+	return payload
+}
+
 func appendFixedASCII(out []byte, value string, size int) []byte {
 	start := len(out)
 	out = append(out, make([]byte, size)...)