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

feat(platform): aggregate realtime by vehicle

Browse Source
This commit is contained in:
lingniu
2026-07-03 23:09:28 +08:00
parent 71b714bbf7
commit 37f9d7a307
12 changed files with 291 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
vehicle-data-platform/apps/api/internal/platform/handler.go
View File

@@ -29,6 +29,7 @@ func (h *Handler) routes() {
h.mux.HandleFunc("GET /api/vehicles", h.handleVehicles)
h.mux.HandleFunc("GET /api/vehicles/coverage", h.handleVehicleCoverage)
h.mux.HandleFunc("GET /api/vehicles/detail", h.handleVehicleDetail)
h.mux.HandleFunc("GET /api/realtime/vehicles", h.handleVehicleRealtime)
h.mux.HandleFunc("GET /api/realtime/locations", h.handleRealtimeLocations)
h.mux.HandleFunc("GET /api/history/locations", h.handleHistoryLocations)
h.mux.HandleFunc("GET /api/history/raw-frames", h.handleRawFramesGet)
@@ -63,6 +64,11 @@ func (h *Handler) handleVehicleDetail(w http.ResponseWriter, r *http.Request) {
h.write(w, r, data, err)
}
func (h *Handler) handleVehicleRealtime(w http.ResponseWriter, r *http.Request) {
data, err := h.service.VehicleRealtime(r.Context(), r.URL.Query())
h.write(w, r, data, err)
}
func (h *Handler) handleRealtimeLocations(w http.ResponseWriter, r *http.Request) {
data, err := h.service.RealtimeLocations(r.Context(), r.URL.Query())
h.write(w, r, data, err)

15
vehicle-data-platform/apps/api/internal/platform/handler_test.go
View File

@@ -89,6 +89,21 @@ func TestHandlerRealtimeLocations(t *testing.T) {
}
}
func TestHandlerVehicleRealtime(t *testing.T) {
handler := NewHandler(NewService(NewMockStore()))
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/realtime/vehicles?limit=10", nil)
handler.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
}
for _, want := range []string{"protocols", "sourceCount", "primaryProtocol", "LB9A32A24R0LS1426"} {
if !strings.Contains(rec.Body.String(), want) {
t.Fatalf("response missing %q: %s", want, rec.Body.String())
}
}
}
func TestHandlerHistoryMileageQualityOps(t *testing.T) {
cases := []struct {
path string

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

@@ -96,6 +96,82 @@ func (m *MockStore) VehicleCoverage(_ context.Context, query url.Values) (Page[V
return page(items, query), nil
}
func (m *MockStore) VehicleRealtime(_ context.Context, query url.Values) (Page[VehicleRealtimeRow], error) {
rows := m.locations
if vin := strings.TrimSpace(query.Get("vin")); vin != "" {
keyword := strings.ToLower(vin)
rows = keep(rows, func(row RealtimeLocationRow) bool {
return strings.Contains(strings.ToLower(row.VIN+row.Plate), keyword)
})
}
if protocol := strings.TrimSpace(query.Get("protocol")); protocol != "" {
rows = keep(rows, func(row RealtimeLocationRow) bool { return row.Protocol == protocol })
}
byVIN := map[string]*VehicleRealtimeRow{}
for _, location := range rows {
current := byVIN[location.VIN]
if current == nil {
vehicle := m.vehicleByVIN(location.VIN)
current = &VehicleRealtimeRow{
VIN: location.VIN,
Plate: firstNonEmpty(location.Plate, vehicle.Plate),
Phone: vehicle.Phone,
OEM: vehicle.OEM,
LastSeen: location.LastSeen,
}
byVIN[location.VIN] = current
}
if !containsString(current.Protocols, location.Protocol) {
current.Protocols = append(current.Protocols, location.Protocol)
current.SourceCount = len(current.Protocols)
}
if location.LastSeen >= current.LastSeen {
current.PrimaryProtocol = location.Protocol
current.Longitude = location.Longitude
current.Latitude = location.Latitude
current.SpeedKmh = location.SpeedKmh
current.SOCPercent = location.SOCPercent
current.TotalMileageKm = location.TotalMileageKm
current.LastSeen = location.LastSeen
}
if location.LastSeen >= "2026-07-03 20:11:00" {
current.Online = true
current.OnlineSourceCount++
}
}
items := make([]VehicleRealtimeRow, 0, len(byVIN))
for _, row := range byVIN {
sort.Strings(row.Protocols)
switch strings.TrimSpace(query.Get("online")) {
case "online":
if !row.Online {
continue
}
case "offline":
if row.Online {
continue
}
}
items = append(items, *row)
}
sort.Slice(items, func(i, j int) bool {
if items[i].LastSeen == items[j].LastSeen {
return items[i].VIN < items[j].VIN
}
return items[i].LastSeen > items[j].LastSeen
})
return page(items, query), nil
}
func (m *MockStore) vehicleByVIN(vin string) VehicleRow {
for _, vehicle := range m.vehicles {
if vehicle.VIN == vin {
return vehicle
}
}
return VehicleRow{}
}
func (m *MockStore) RealtimeLocations(_ context.Context, query url.Values) (Page[RealtimeLocationRow], error) {
items := m.locations
if vin := strings.TrimSpace(query.Get("vin")); vin != "" {

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

@@ -88,6 +88,24 @@ type RealtimeLocationRow struct {
LastSeen string `json:"lastSeen"`
}
type VehicleRealtimeRow struct {
VIN string `json:"vin"`
Plate string `json:"plate"`
Phone string `json:"phone"`
OEM string `json:"oem"`
Protocols []string `json:"protocols"`
SourceCount int `json:"sourceCount"`
OnlineSourceCount int `json:"onlineSourceCount"`
Online bool `json:"online"`
PrimaryProtocol string `json:"primaryProtocol"`
Longitude float64 `json:"longitude"`
Latitude float64 `json:"latitude"`
SpeedKmh float64 `json:"speedKmh"`
SOCPercent float64 `json:"socPercent"`
TotalMileageKm float64 `json:"totalMileageKm"`
LastSeen string `json:"lastSeen"`
}
type HistoryLocationRow struct {
VIN string `json:"vin"`
Plate string `json:"plate"`

56
vehicle-data-platform/apps/api/internal/platform/mysql_queries.go
View File

@@ -136,6 +136,62 @@ func buildRealtimeLocationSQL(query url.Values) SQLQuery {
}
}
func buildVehicleRealtimeSQL(query url.Values) SQLQuery {
limit := parsePositive(query.Get("limit"), 20)
offset := parsePositive(query.Get("offset"), 0)
args := []any{}
where := []string{"l.vin IS NOT NULL", "l.vin <> ''"}
having := []string{}
if protocol := strings.TrimSpace(query.Get("protocol")); protocol != "" {
where = append(where, "l.protocol = ?")
args = append(args, protocol)
}
if keyword := strings.TrimSpace(query.Get("vin")); keyword != "" {
where = append(where, "(l.vin LIKE ? OR l.plate LIKE ? OR b.plate LIKE ? OR b.phone LIKE ? OR b.oem LIKE ?)")
like := "%" + keyword + "%"
args = append(args, like, like, like, like, like)
}
switch strings.TrimSpace(query.Get("online")) {
case "online":
having = append(having, "COUNT(DISTINCT CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN l.protocol END) > 0")
case "offline":
having = append(having, "COUNT(DISTINCT CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN l.protocol END) = 0")
}
countArgs := append([]any(nil), args...)
args = append(args, limit, offset)
havingSQL := ""
if len(having) > 0 {
havingSQL = ` HAVING ` + strings.Join(having, " AND ") + ` `
}
groupSQL := `FROM vehicle_realtime_location l ` +
`LEFT JOIN vehicle_identity_binding b ON b.vin = l.vin ` +
`WHERE ` + strings.Join(where, " AND ") + ` ` +
`GROUP BY l.vin, b.plate, b.phone, b.oem ` +
havingSQL
orderExpr := `l.updated_at DESC, l.protocol ASC`
return SQLQuery{
Text: `SELECT l.vin, ` +
`COALESCE(NULLIF(MAX(NULLIF(l.plate, '')), ''), b.plate, '') AS plate, ` +
`COALESCE(b.phone, '') AS phone, COALESCE(b.oem, '') AS oem, ` +
`COALESCE(GROUP_CONCAT(DISTINCT l.protocol ORDER BY l.protocol SEPARATOR ','), '') AS protocols, ` +
`COUNT(DISTINCT l.protocol) AS source_count, ` +
`COUNT(DISTINCT CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN l.protocol END) AS online_source_count, ` +
`CASE WHEN COUNT(DISTINCT CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN l.protocol END) > 0 THEN 1 ELSE 0 END AS online, ` +
`COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(l.protocol ORDER BY ` + orderExpr + `), ',', 1), '') AS primary_protocol, ` +
`COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(l.longitude AS CHAR) ORDER BY ` + orderExpr + `), ',', 1), '') AS longitude, ` +
`COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(l.latitude AS CHAR) ORDER BY ` + orderExpr + `), ',', 1), '') AS latitude, ` +
`COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(l.speed_kmh AS CHAR) ORDER BY ` + orderExpr + `), ',', 1), '') AS speed_kmh, ` +
`COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(l.soc_percent AS CHAR) ORDER BY ` + orderExpr + `), ',', 1), '') AS soc_percent, ` +
`COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(l.total_mileage_km AS CHAR) ORDER BY ` + orderExpr + `), ',', 1), '') AS total_mileage_km, ` +
`COALESCE(DATE_FORMAT(MAX(l.updated_at), '%Y-%m-%d %H:%i:%s'), '') AS last_seen ` +
groupSQL +
`ORDER BY MAX(l.updated_at) DESC, l.vin ASC LIMIT ? OFFSET ?`,
Args: args,
CountText: `SELECT COUNT(*) FROM (SELECT l.vin ` + groupSQL + `) vehicle_realtime_count`,
CountArgs: countArgs,
}
}
func buildDailyMileageSQL(query url.Values) SQLQuery {
limit := parsePositive(query.Get("limit"), 20)
offset := parsePositive(query.Get("offset"), 0)

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

@@ -149,6 +149,42 @@ func (s *ProductionStore) VehicleCoverage(ctx context.Context, query url.Values)
return Page[VehicleCoverageRow]{Items: items, Total: total, Limit: limit, Offset: offset}, nil
}
func (s *ProductionStore) VehicleRealtime(ctx context.Context, query url.Values) (Page[VehicleRealtimeRow], error) {
built := buildVehicleRealtimeSQL(query)
total := 0
if err := s.db.QueryRowContext(ctx, built.CountText, built.CountArgs...).Scan(&total); err != nil {
return Page[VehicleRealtimeRow]{}, err
}
rows, err := s.db.QueryContext(ctx, built.Text, built.Args...)
if err != nil {
return Page[VehicleRealtimeRow]{}, err
}
defer rows.Close()
items := make([]VehicleRealtimeRow, 0)
for rows.Next() {
var row VehicleRealtimeRow
var protocols string
var online int
var longitude, latitude, speed, soc, mileage string
if err := rows.Scan(&row.VIN, &row.Plate, &row.Phone, &row.OEM, &protocols, &row.SourceCount, &row.OnlineSourceCount, &online, &row.PrimaryProtocol, &longitude, &latitude, &speed, &soc, &mileage, &row.LastSeen); err != nil {
return Page[VehicleRealtimeRow]{}, err
}
row.Protocols = splitCSV(protocols)
row.Online = online == 1
row.Longitude = parseFloatString(longitude)
row.Latitude = parseFloatString(latitude)
row.SpeedKmh = parseFloatString(speed)
row.SOCPercent = parseFloatString(soc)
row.TotalMileageKm = parseFloatString(mileage)
items = append(items, row)
}
if err := rows.Err(); err != nil {
return Page[VehicleRealtimeRow]{}, err
}
limit, offset := buildLimitOffset(query)
return Page[VehicleRealtimeRow]{Items: items, Total: total, Limit: limit, Offset: offset}, nil
}
func (s *ProductionStore) RealtimeLocations(ctx context.Context, query url.Values) (Page[RealtimeLocationRow], error) {
built := buildRealtimeLocationSQL(query)
total := 0
@@ -175,6 +211,14 @@ func (s *ProductionStore) RealtimeLocations(ctx context.Context, query url.Value
return Page[RealtimeLocationRow]{Items: items, Total: total, Limit: limit, Offset: offset}, nil
}
func parseFloatString(value string) float64 {
parsed, err := strconv.ParseFloat(strings.TrimSpace(value), 64)
if err != nil {
return 0
}
return parsed
}
func (s *ProductionStore) HistoryLocations(ctx context.Context, query url.Values) (Page[HistoryLocationRow], error) {
realtime, err := s.RealtimeLocations(ctx, query)
if err != nil {

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

@@ -67,6 +67,25 @@ func TestBuildRealtimeLocationSQL(t *testing.T) {
}
}
func TestBuildVehicleRealtimeSQL(t *testing.T) {
query := url.Values{"vin": {"粤A"}, "protocol": {"JT808"}, "online": {"online"}, "limit": {"10"}, "offset": {"20"}}
built := buildVehicleRealtimeSQL(query)
for _, want := range []string{"vehicle_realtime_location", "vehicle_identity_binding", "GROUP BY l.vin", "GROUP_CONCAT(DISTINCT l.protocol", "online_source_count", "vehicle_realtime_count"} {
if !strings.Contains(built.Text+built.CountText, want) {
t.Fatalf("SQL missing %q: %s / %s", want, built.Text, built.CountText)
}
}
if !strings.Contains(built.Text, "ORDER BY MAX(l.updated_at) DESC, l.vin ASC") {
t.Fatalf("SQL should keep vehicle-level stable pagination order: %s", built.Text)
}
if len(built.Args) != 8 || built.Args[0] != "JT808" || built.Args[1] != "%粤A%" || built.Args[6] != 10 || built.Args[7] != 20 {
t.Fatalf("args = %#v", built.Args)
}
if len(built.CountArgs) != 6 || built.CountArgs[0] != "JT808" || built.CountArgs[1] != "%粤A%" {
t.Fatalf("count args = %#v", built.CountArgs)
}
}
func TestBuildDailyMileageSQL(t *testing.T) {
query := url.Values{"vin": {"VIN001"}, "protocol": {"JT808"}, "dateFrom": {"2026-07-01"}, "dateTo": {"2026-07-03"}}
built := buildDailyMileageSQL(query)

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

@@ -11,6 +11,7 @@ type Store interface {
DashboardSummary(context.Context) (DashboardSummary, error)
Vehicles(context.Context, url.Values) (Page[VehicleRow], error)
VehicleCoverage(context.Context, url.Values) (Page[VehicleCoverageRow], error)
VehicleRealtime(context.Context, url.Values) (Page[VehicleRealtimeRow], error)
RealtimeLocations(context.Context, url.Values) (Page[RealtimeLocationRow], error)
HistoryLocations(context.Context, url.Values) (Page[HistoryLocationRow], error)
HistoryLocationsFromTDengine(context.Context, url.Values) (Page[HistoryLocationRow], error)
@@ -51,6 +52,10 @@ func (s *Service) VehicleCoverage(ctx context.Context, query url.Values) (Page[V
return s.store.VehicleCoverage(ctx, query)
}
func (s *Service) VehicleRealtime(ctx context.Context, query url.Values) (Page[VehicleRealtimeRow], error) {
return s.store.VehicleRealtime(ctx, query)
}
func (s *Service) VehicleDetail(ctx context.Context, vin string, protocol string) (VehicleDetail, error) {
keyword := strings.TrimSpace(vin)
protocol = strings.TrimSpace(protocol)