From 192f1151769f30ac89c68f8bb3ccd8bdda73256c Mon Sep 17 00:00:00 2001 From: lingniu Date: Mon, 29 Jun 2026 13:59:06 +0800 Subject: [PATCH] feat: add telemetry field history api --- .../TelemetryFieldHistoryController.java | 189 ++++++++++++++++++ .../config/EventHistoryAutoConfiguration.java | 8 + .../TelemetryFieldHistoryControllerTest.java | 152 ++++++++++++++ .../EventHistoryAutoConfigurationTest.java | 9 + 4 files changed, 358 insertions(+) create mode 100644 modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryController.java create mode 100644 modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryControllerTest.java diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryController.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryController.java new file mode 100644 index 00000000..bcb23cbd --- /dev/null +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryController.java @@ -0,0 +1,189 @@ +package com.lingniu.ingest.eventhistory; + +import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader; +import com.lingniu.ingest.tdenginehistory.TdenginePage; +import com.lingniu.ingest.tdenginehistory.TdenginePageCursor; +import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder; +import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery; +import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.Parameter; +import io.swagger.v3.oas.annotations.tags.Tag; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.http.HttpStatus; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.server.ResponseStatusException; + +import java.io.IOException; +import java.time.Instant; +import java.util.List; +import java.util.Locale; + +@RestController +@ConditionalOnProperty(prefix = "lingniu.ingest.event-history", name = "enabled", havingValue = "true") +@ConditionalOnBean(TdengineHistoryReader.class) +@RequestMapping("/api/event-history/telemetry") +@Tag(name = "telemetry-field-history-controller", description = "遥测字段历史分页查询接口。") +public final class TelemetryFieldHistoryController { + + private final TdengineHistoryReader reader; + + public TelemetryFieldHistoryController(TdengineHistoryReader reader) { + if (reader == null) { + throw new IllegalArgumentException("reader must not be null"); + } + this.reader = reader; + } + + @GetMapping("/fields") + @Operation( + summary = "查询单车单字段历史值", + description = "按协议、车辆标识、字段 key 和时间范围查询 TDengine telemetry_fields。分页使用 cursorTs + cursorId，不使用 offset。") + public TelemetryFieldPageResponse fields( + @Parameter(description = "协议名，例如 GB32960 或 JT808。", required = true, example = "GB32960") + @RequestParam String protocol, + @Parameter(description = "字段 key，例如 FUEL_CELL_V2016.hydrogenComsuxxx。", required = true) + @RequestParam String fieldKey, + @Parameter(description = "内部车辆键。传入后优先使用。", example = "LNVFC000000000001") + @RequestParam(required = false) String vehicleKey, + @Parameter(description = "VIN。GB32960 常用该字段作为车辆键。", example = "LNVFC000000000001") + @RequestParam(required = false) String vin, + @Parameter(description = "JT808 终端手机号或终端标识；JT808 会自动映射为 jt808:。", example = "g7gps") + @RequestParam(required = false) String phone, + @Parameter(description = "开始时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:00:00") + @RequestParam String dateFrom, + @Parameter(description = "结束时间，支持日期或精确到秒的时间；无时区时按东八区解析。", example = "2026-06-29T13:10:00") + @RequestParam String dateTo, + @Parameter(description = "排序方向。", example = "DESC") + @RequestParam(defaultValue = "DESC") TdengineQueryOrder order, + @Parameter(description = "返回值数量上限，最大 1000。", example = "100") + @RequestParam(defaultValue = "100") int limit, + @Parameter(description = "上一页返回的 nextCursor.ts。", example = "2026-06-29T05:00:01Z") + @RequestParam(required = false) String cursorTs, + @Parameter(description = "上一页返回的 nextCursor.id。", example = "event-xxx#0") + @RequestParam(required = false) String cursorId) throws IOException { + String normalizedProtocol = require(protocol, "protocol is required for telemetry field query").toUpperCase(Locale.ROOT); + String normalizedFieldKey = require(fieldKey, "fieldKey is required for telemetry field query"); + QueryTimeRange range = QueryTimeRange.parse(dateFrom, dateTo); + TdenginePage page = reader.queryTelemetryFields(new TdengineTelemetryFieldQuery( + normalizedProtocol, + resolveVehicleKey(normalizedProtocol, vehicleKey, vin, phone), + normalizedFieldKey, + range.eventTimeFrom(), + range.eventTimeTo().plusMillis(1), + order, + limit, + cursor(cursorTs, cursorId))); + return TelemetryFieldPageResponse.from(page); + } + + private static String resolveVehicleKey(String protocol, String vehicleKey, String vin, String phone) { + String explicitVehicleKey = trimToNull(vehicleKey); + if (explicitVehicleKey != null) { + return explicitVehicleKey; + } + String vinValue = trimToNull(vin); + if (vinValue != null) { + return vinValue; + } + String phoneValue = trimToNull(phone); + if (phoneValue != null) { + if ("JT808".equals(protocol) && !phoneValue.startsWith("jt808:")) { + return "jt808:" + phoneValue; + } + return phoneValue; + } + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "vehicleKey, vin or phone is required"); + } + + private static String require(String value, String message) { + String trimmed = trimToNull(value); + if (trimmed == null) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, message); + } + return trimmed; + } + + private static String trimToNull(String value) { + if (value == null || value.isBlank()) { + return null; + } + return value.trim(); + } + + private static TdenginePageCursor cursor(String cursorTs, String cursorId) { + boolean hasTs = cursorTs != null && !cursorTs.isBlank(); + boolean hasId = cursorId != null && !cursorId.isBlank(); + if (!hasTs && !hasId) { + return null; + } + if (!hasTs || !hasId) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "cursorTs and cursorId must be provided together"); + } + return new TdenginePageCursor(Instant.parse(cursorTs.trim()), cursorId.trim()); + } + + public record TelemetryFieldPageResponse( + List items, + CursorResponse nextCursor) { + + private static TelemetryFieldPageResponse from(TdenginePage page) { + CursorResponse next = page.nextCursor() + .map(cursor -> new CursorResponse(cursor.ts().toString(), cursor.tieBreaker())) + .orElse(null); + return new TelemetryFieldPageResponse( + page.items().stream().map(TelemetryFieldResponse::from).toList(), + next); + } + } + + public record CursorResponse(String ts, String id) { + } + + public record TelemetryFieldResponse( + String eventTime, + String receivedAt, + String factId, + String frameId, + String fieldKey, + String valueType, + String valueText, + Double valueDouble, + Long valueLong, + String unit, + String quality, + String sourcePath, + String rawUri, + String metadataJson, + String protocol, + String vehicleKey, + String vin, + String phone) { + + private static TelemetryFieldResponse from(TdengineTelemetryFieldRow row) { + return new TelemetryFieldResponse( + row.ts().toString(), + row.receivedAt().toString(), + row.factId(), + row.frameId(), + row.fieldKey(), + row.valueType(), + row.valueText(), + row.valueDouble(), + row.valueLong(), + row.unit(), + row.quality(), + row.sourcePath(), + row.rawUri(), + row.metadataJson(), + row.protocol(), + row.vehicleKey(), + row.vin(), + row.phone()); + } + } +} diff --git a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java index 9936c81c..edcb8c12 100644 --- a/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java +++ b/modules/services/event-history-service/src/main/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfiguration.java @@ -8,6 +8,7 @@ import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor; import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService; import com.lingniu.ingest.eventhistory.Gb32960FrameController; import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController; +import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController; import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper; import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder; import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration; @@ -98,6 +99,13 @@ public class EventHistoryAutoConfiguration { return new Jt808LocationHistoryController(reader); } + @Bean + @ConditionalOnBean(TdengineHistoryReader.class) + @ConditionalOnMissingBean + public TelemetryFieldHistoryController telemetryFieldHistoryController(TdengineHistoryReader reader) { + return new TelemetryFieldHistoryController(reader); + } + static Path archiveRoot(String value) { if (value == null || value.isBlank()) { return Path.of(System.getProperty("java.io.tmpdir"), "lingniu-archive"); diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryControllerTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryControllerTest.java new file mode 100644 index 00000000..c820a2c5 --- /dev/null +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/TelemetryFieldHistoryControllerTest.java @@ -0,0 +1,152 @@ +package com.lingniu.ingest.eventhistory; + +import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader; +import com.lingniu.ingest.tdenginehistory.TdengineLocationQuery; +import com.lingniu.ingest.tdenginehistory.TdengineLocationRow; +import com.lingniu.ingest.tdenginehistory.TdenginePage; +import com.lingniu.ingest.tdenginehistory.TdenginePageCursor; +import com.lingniu.ingest.tdenginehistory.TdengineQueryOrder; +import com.lingniu.ingest.tdenginehistory.TdengineRawFrameQuery; +import com.lingniu.ingest.tdenginehistory.TdengineRawFrameRow; +import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldQuery; +import com.lingniu.ingest.tdenginehistory.TdengineTelemetryFieldRow; +import org.junit.jupiter.api.Test; +import org.springframework.http.HttpStatus; +import org.springframework.web.server.ResponseStatusException; + +import java.io.IOException; +import java.time.Instant; +import java.util.List; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class TelemetryFieldHistoryControllerTest { + + @Test + void queriesGb32960FieldByVinWithKeysetCursor() throws Exception { + CapturingReader reader = new CapturingReader(new TdenginePage<>( + List.of(row("field-1", "2026-06-29T05:00:01Z")), + Optional.of(new TdenginePageCursor(Instant.parse("2026-06-29T05:00:01Z"), "field-1")))); + TelemetryFieldHistoryController controller = new TelemetryFieldHistoryController(reader); + + TelemetryFieldHistoryController.TelemetryFieldPageResponse response = controller.fields( + "GB32960", + "FUEL_CELL_V2016.hydrogenComsuxxx", + null, + "LNVFC000000000001", + null, + "2026-06-29T13:00:00+08:00", + "2026-06-29T13:10:00+08:00", + TdengineQueryOrder.ASC, + 100, + "2026-06-29T05:00:00Z", + "field-0"); + + assertThat(reader.query.protocol()).isEqualTo("GB32960"); + assertThat(reader.query.vehicleKey()).isEqualTo("LNVFC000000000001"); + assertThat(reader.query.fieldKey()).isEqualTo("FUEL_CELL_V2016.hydrogenComsuxxx"); + assertThat(reader.query.order()).isEqualTo(TdengineQueryOrder.ASC); + assertThat(reader.query.limit()).isEqualTo(100); + assertThat(reader.query.cursor()).isEqualTo(new TdenginePageCursor( + Instant.parse("2026-06-29T05:00:00Z"), "field-0")); + assertThat(response.items()) + .extracting(TelemetryFieldHistoryController.TelemetryFieldResponse::factId) + .containsExactly("field-1"); + assertThat(response.items().getFirst().valueDouble()).isEqualTo(3.14); + assertThat(response.nextCursor()).isEqualTo(new TelemetryFieldHistoryController.CursorResponse( + "2026-06-29T05:00:01Z", "field-1")); + } + + @Test + void mapsJt808PhoneToVehicleKeyWhenVehicleKeyIsAbsent() throws Exception { + CapturingReader reader = new CapturingReader(new TdenginePage<>(List.of(), Optional.empty())); + TelemetryFieldHistoryController controller = new TelemetryFieldHistoryController(reader); + + controller.fields( + "JT808", + "location.speed_kmh", + null, + null, + "g7gps", + "2026-06-29", + "2026-06-29", + TdengineQueryOrder.DESC, + 50, + null, + null); + + assertThat(reader.query.vehicleKey()).isEqualTo("jt808:g7gps"); + } + + @Test + void rejectsMissingVehicleIdentity() { + TelemetryFieldHistoryController controller = new TelemetryFieldHistoryController( + new CapturingReader(new TdenginePage<>(List.of(), Optional.empty()))); + + assertThatThrownBy(() -> controller.fields( + "GB32960", + "vehicle.speed", + null, + null, + null, + "2026-06-29", + "2026-06-29", + TdengineQueryOrder.DESC, + 50, + null, + null)) + .isInstanceOfSatisfying(ResponseStatusException.class, ex -> + assertThat(ex.getStatusCode()).isEqualTo(HttpStatus.BAD_REQUEST)); + } + + private static TdengineTelemetryFieldRow row(String factId, String ts) { + Instant instant = Instant.parse(ts); + return new TdengineTelemetryFieldRow( + instant, + factId, + "frame-1", + instant.plusMillis(500), + "FUEL_CELL_V2016.hydrogenComsuxxx", + "DOUBLE", + "3.14", + 3.14, + null, + "kg/100km", + "GOOD", + "FUEL_CELL_V2016.hydrogenComsuxxx", + "archive://gb32960/frame-1.bin", + "{\"cmd\":\"REALTIME_REPORT\"}", + "GB32960", + "LNVFC000000000001", + "LNVFC000000000001", + ""); + } + + private static final class CapturingReader implements TdengineHistoryReader { + private final TdenginePage response; + private TdengineTelemetryFieldQuery query; + + private CapturingReader(TdenginePage response) { + this.response = response; + } + + @Override + public TdenginePage queryRawFrames(TdengineRawFrameQuery query) { + throw new UnsupportedOperationException(); + } + + @Override + public TdenginePage queryLocations(TdengineLocationQuery query) { + throw new UnsupportedOperationException(); + } + + @Override + public TdenginePage queryTelemetryFields(TdengineTelemetryFieldQuery query) + throws IOException { + this.query = query; + return response; + } + } +} diff --git a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java index 101a9074..baa29aa6 100644 --- a/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java +++ b/modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/config/EventHistoryAutoConfigurationTest.java @@ -8,6 +8,7 @@ import com.lingniu.ingest.eventhistory.EventHistoryEnvelopeIngestor; import com.lingniu.ingest.eventhistory.Gb32960DecodedFrameService; import com.lingniu.ingest.eventhistory.Gb32960FrameController; import com.lingniu.ingest.eventhistory.Jt808LocationHistoryController; +import com.lingniu.ingest.eventhistory.TelemetryFieldHistoryController; import com.lingniu.ingest.eventhistory.TelemetryEnvelopeRecordMapper; import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder; import com.lingniu.ingest.tdenginehistory.TdengineHistoryReader; @@ -73,6 +74,14 @@ class EventHistoryAutoConfigurationTest { .run(context -> assertThat(context).hasSingleBean(Jt808LocationHistoryController.class)); } + @Test + void createsTelemetryFieldHistoryControllerWhenTdengineReaderExists() { + contextRunner + .withPropertyValues("lingniu.ingest.event-history.enabled=true") + .withBean(TdengineHistoryReader.class, () -> mock(TdengineHistoryReader.class)) + .run(context -> assertThat(context).hasSingleBean(TelemetryFieldHistoryController.class)); + } + @Test void parsesFileUriArchiveRoot() { assertThat(EventHistoryAutoConfiguration.archiveRoot("file:///tmp/lingniu-test-archive"))