From 3cc7ac9669c2b68f684d7832d95346bc6ece1c53 Mon Sep 17 00:00:00 2001 From: lingniu Date: Tue, 30 Jun 2026 17:02:54 +0800 Subject: [PATCH] feat: add jt808 streaming daily mileage --- deploy/portainer/docker-compose.yml | 10 + .../operations/jt808-daily-mileage-runbook.md | 49 ++ .../2026-06-30-kafka-streaming-mileage.md | 457 ++++++++++++++++++ .../src/main/resources/application.yml | 28 +- modules/services/vehicle-stat-service/pom.xml | 17 + .../VehicleStatEnvelopeIngestor.java | 16 + .../config/VehicleStatAutoConfiguration.java | 79 ++- .../config/VehicleStatProperties.java | 85 ++++ .../jt808/InMemoryJt808MileageStateStore.java | 25 + .../JdbcJt808DailyMileageRepository.java | 61 +++ .../jt808/Jt808DailyMileageRepository.java | 6 + .../jt808/Jt808DailyMileageResult.java | 22 + .../jt808/Jt808DailyMileageState.java | 223 +++++++++ .../jt808/Jt808GpsMileageCalculator.java | 93 ++++ .../vehiclestat/jt808/Jt808LocationPoint.java | 27 ++ .../jt808/Jt808LocationPointExtractor.java | 96 ++++ .../jt808/Jt808MileageStateStore.java | 11 + .../jt808/Jt808MileageStreamProcessor.java | 41 ++ .../jt808/RedisJt808MileageStateStore.java | 167 +++++++ .../VehicleStatEnvelopeIngestorTest.java | 17 + .../VehicleStatAutoConfigurationTest.java | 43 ++ .../JdbcJt808DailyMileageRepositoryTest.java | 102 ++++ .../jt808/Jt808DailyMileageStateTest.java | 89 ++++ .../jt808/Jt808GpsMileageCalculatorTest.java | 86 ++++ .../Jt808LocationPointExtractorTest.java | 84 ++++ .../Jt808MileageStreamProcessorTest.java | 95 ++++ .../RedisJt808MileageStateStoreTest.java | 86 ++++ 27 files changed, 2109 insertions(+), 6 deletions(-) create mode 100644 docs/operations/jt808-daily-mileage-runbook.md create mode 100644 docs/superpowers/plans/2026-06-30-kafka-streaming-mileage.md create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/InMemoryJt808MileageStateStore.java create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepository.java create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageRepository.java create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageResult.java create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageState.java create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculator.java create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPoint.java create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPointExtractor.java create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStateStore.java create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessor.java create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStore.java create mode 100644 modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepositoryTest.java create mode 100644 modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageStateTest.java create mode 100644 modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculatorTest.java create mode 100644 modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPointExtractorTest.java create mode 100644 modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessorTest.java create mode 100644 modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStoreTest.java diff --git a/deploy/portainer/docker-compose.yml b/deploy/portainer/docker-compose.yml index 8ebfc1d1..9180db08 100644 --- a/deploy/portainer/docker-compose.yml +++ b/deploy/portainer/docker-compose.yml @@ -235,6 +235,16 @@ services: VEHICLE_STAT_ENABLED: ${VEHICLE_STAT_ENABLED:-true} VEHICLE_STAT_FILE_PATH: /vehicle-stat/ VEHICLE_STAT_ZONE_ID: ${VEHICLE_STAT_ZONE_ID:-Asia/Shanghai} + VEHICLE_STAT_JT808_MILEAGE_ENABLED: ${VEHICLE_STAT_JT808_MILEAGE_ENABLED:-true} + VEHICLE_STAT_JT808_STATE_STORE: ${VEHICLE_STAT_JT808_STATE_STORE:-redis} + VEHICLE_STAT_JT808_STATE_TTL_DAYS: ${VEHICLE_STAT_JT808_STATE_TTL_DAYS:-3} + MYSQL_JDBC_URL: ${MYSQL_JDBC_URL:-} + MYSQL_USERNAME: ${MYSQL_USERNAME:-} + MYSQL_PASSWORD: ${MYSQL_PASSWORD:-} + REDIS_HOST: ${REDIS_HOST:-} + REDIS_PORT: ${REDIS_PORT:-6379} + REDIS_DATABASE: ${REDIS_DATABASE:-50} + REDIS_PASSWORD: ${REDIS_PASSWORD:-} ports: - "${VEHICLE_ANALYTICS_HTTP_PORT:-20300}:20300" volumes: diff --git a/docs/operations/jt808-daily-mileage-runbook.md b/docs/operations/jt808-daily-mileage-runbook.md new file mode 100644 index 00000000..5637e6ff --- /dev/null +++ b/docs/operations/jt808-daily-mileage-runbook.md @@ -0,0 +1,49 @@ +# JT808 Daily Mileage Streaming + +## Scope + +The analytics app can calculate daily mileage from JT808 telemetry only. It consumes `vehicle.event.jt808.v1`, keeps the rolling per-vehicle daily state in Redis or memory, and upserts the current daily result into MySQL. + +## Required MySQL Table + +```sql +create table if not exists vehicle_daily_mileage_jt808 ( + stat_date date not null, + vehicle_key varchar(64) not null, + vin varchar(32) null, + phone varchar(32) null, + first_event_time timestamp null, + last_event_time timestamp null, + gps_mileage_km decimal(12,3) not null, + speed_integral_km decimal(12,3) not null, + odometer_mileage_km decimal(12,3) null, + accepted_points int not null, + bad_jump_segments int not null, + long_gap_segments int not null, + out_of_order_points int not null, + odometer_anomalies int not null, + data_quality varchar(16) not null, + updated_at timestamp not null default current_timestamp on update current_timestamp, + primary key (stat_date, vehicle_key), + key idx_vehicle_daily_mileage_jt808_vehicle_date (vehicle_key, stat_date) +); +``` + +## Runtime Settings + +Set these in Portainer or Nacos, without committing secrets: + +```text +KAFKA_TOPIC_JT808_EVENT=vehicle.event.jt808.v1 +VEHICLE_STAT_JT808_MILEAGE_ENABLED=true +VEHICLE_STAT_JT808_STATE_STORE=redis +MYSQL_JDBC_URL=jdbc:mysql://:3306/?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai +MYSQL_USERNAME= +MYSQL_PASSWORD= +REDIS_HOST= +REDIS_PORT=6379 +REDIS_DATABASE=50 +REDIS_PASSWORD= +``` + +Algorithm defaults are Asia/Shanghai daily boundary, 5 minute max segment gap, 200 km/h max implied speed, and short-gap jump guard of 300 meters within 10 seconds. diff --git a/docs/superpowers/plans/2026-06-30-kafka-streaming-mileage.md b/docs/superpowers/plans/2026-06-30-kafka-streaming-mileage.md new file mode 100644 index 00000000..ee6f8c9b --- /dev/null +++ b/docs/superpowers/plans/2026-06-30-kafka-streaming-mileage.md @@ -0,0 +1,457 @@ +# JT808 Kafka Streaming Mileage Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Consume only JT808 Kafka location events, calculate daily mileage in real time, keep rolling state in Redis or memory, and upsert final/current daily results into MySQL. + +**Architecture:** Reuse `vehicle-analytics-app` and `vehicle-stat-service` instead of adding a new runtime. The Kafka consumer subscribes to `vehicle.event.jt808.v1`, extracts JT808 location points from `VehicleEnvelope.telemetry_snapshot`, updates one rolling state per `(vehicleKey, statDate)`, stores that state in Redis DB 50 when available, and writes the current daily aggregate to MySQL table `vehicle_daily_mileage_jt808`. No GB32960 comparison and no historical backfill are required for v1. + +**Tech Stack:** Java 21, Spring Boot auto-configuration, existing `sink-mq` Kafka consumer, protobuf `VehicleEnvelope`, Spring Data Redis `StringRedisTemplate`, Spring JDBC `JdbcTemplate`, MySQL `INSERT ... ON DUPLICATE KEY UPDATE`. + +--- + +## Confirmed Runtime Inputs + +- Kafka topic: `vehicle.event.jt808.v1` +- Statistics source: JT808 only +- MySQL database: reuse the configured production vehicle database +- MySQL user: use the configured production vehicle database user +- Redis endpoint: configure through `REDIS_HOST` and `REDIS_PORT` +- Redis DB: use `50` by default; `50-60` are available +- Date zone: `Asia/Shanghai` +- Segment rules: + - max segment gap: `PT5M` + - max implied speed: `200 km/h` + - short-gap jump filter: `PT10S` and `300 m` +- Historical backfill: not needed for v1 +- Vehicle key: + - if VIN is present and not `unknown`, use VIN + - otherwise use `jt808:{phone}` + +## Configuration To Add + +Add these deployment properties to `vehicle-analytics-app`: + +```yaml +spring: + data: + redis: + host: ${REDIS_HOST:} + port: ${REDIS_PORT:6379} + password: ${REDIS_PASSWORD:} + database: ${REDIS_DATABASE:50} + +lingniu: + ingest: + sink: + mq: + consumer: + bindings: + vehicleStatEnvelopeConsumerProcessor: + topics: + - ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1} + vehicle-stat: + jt808-mileage: + enabled: ${JT808_MILEAGE_ENABLED:true} + mysql: + jdbc-url: ${JT808_MILEAGE_MYSQL_JDBC_URL:} + username: ${JT808_MILEAGE_MYSQL_USERNAME:} + password: ${JT808_MILEAGE_MYSQL_PASSWORD:} + table-name: ${JT808_MILEAGE_MYSQL_TABLE:vehicle_daily_mileage_jt808} + initialize-schema: ${JT808_MILEAGE_MYSQL_INITIALIZE_SCHEMA:true} + state: + store: ${JT808_MILEAGE_STATE_STORE:redis} + redis-key-prefix: ${JT808_MILEAGE_REDIS_KEY_PREFIX:vehicle:mileage:jt808:daily:} + ttl: ${JT808_MILEAGE_REDIS_TTL:P3D} + max-segment-gap: ${JT808_MILEAGE_MAX_SEGMENT_GAP:PT5M} + max-implied-speed-kmh: ${JT808_MILEAGE_MAX_IMPLIED_SPEED_KMH:200} + short-gap: ${JT808_MILEAGE_SHORT_GAP:PT10S} + short-gap-jump-meters: ${JT808_MILEAGE_SHORT_GAP_JUMP_METERS:300} +``` + +Do not commit real passwords to `application.yml`; pass them through Nacos or environment variables. + +## MySQL Table + +Create automatically when `initialize-schema=true`: + +```sql +CREATE TABLE IF NOT EXISTS vehicle_daily_mileage_jt808 ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + stat_date DATE NOT NULL, + vehicle_key VARCHAR(64) NOT NULL, + vin VARCHAR(32) NOT NULL DEFAULT '', + phone VARCHAR(32) NOT NULL DEFAULT '', + first_event_time DATETIME(3) NULL, + last_event_time DATETIME(3) NULL, + gps_mileage_km DECIMAL(12,3) NOT NULL DEFAULT 0, + speed_integral_km DECIMAL(12,3) NOT NULL DEFAULT 0, + odometer_mileage_km DECIMAL(12,3) NULL, + accepted_points INT NOT NULL DEFAULT 0, + bad_jump_segments INT NOT NULL DEFAULT 0, + long_gap_segments INT NOT NULL DEFAULT 0, + out_of_order_points INT NOT NULL DEFAULT 0, + odometer_anomalies INT NOT NULL DEFAULT 0, + data_quality VARCHAR(16) NOT NULL DEFAULT 'PARTIAL', + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uk_jt808_daily_mileage (stat_date, vehicle_key), + KEY idx_jt808_daily_mileage_vin_date (vin, stat_date), + KEY idx_jt808_daily_mileage_phone_date (phone, stat_date) +); +``` + +## File Structure + +- Modify `modules/services/vehicle-stat-service/pom.xml` + - Add `spring-boot-starter-data-redis`, `spring-jdbc`, and `mysql-connector-j` if not already present through module dependencies. +- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageProperties.java` + - Configuration properties for MySQL, Redis state, and GPS filters. +- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPoint.java` + - Normalized JT808 point. +- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPointExtractor.java` + - Extracts only JT808 location snapshots from Kafka envelopes. +- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculator.java` + - Haversine segment calculator with filtering. +- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageState.java` + - Rolling mutable daily state. +- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageResult.java` + - Persisted result model. +- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStateStore.java` + - State load/save abstraction. +- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStore.java` + - Redis JSON state store using `StringRedisTemplate`. +- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/InMemoryJt808MileageStateStore.java` + - Local fallback state store. +- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepository.java` + - MySQL schema initialization and upsert. +- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessor.java` + - One-envelope processing flow. +- Modify `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java` + - Fan out to existing odometer processor and optional JT808 mileage stream processor. +- Modify `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java` + - Wire the new JT808 mileage beans. +- Modify `modules/apps/vehicle-analytics-app/src/main/resources/application.yml` + - Change stat binding topic to `vehicle.event.jt808.v1` when JT808 mileage is enabled. + +## Task 1: JT808 Point Extraction + +**Files:** +- Create: `.../jt808/Jt808LocationPoint.java` +- Create: `.../jt808/Jt808LocationPointExtractor.java` +- Test: `.../Jt808LocationPointExtractorTest.java` + +- [ ] **Step 1: Write failing tests** + +Test cases: +- Extracts a `source=JT808`, `eventType=LOCATION` envelope with `longitude`, `latitude`, `speed_kmh`, `location_status_raw`, and `total_mileage_km`. +- Returns empty for `source=GB32960`. +- Returns empty when longitude or latitude is missing. +- Uses VIN as `vehicleKey` when VIN is not blank and not `unknown`. +- Uses `jt808:{phone}` when VIN is `unknown`. + +- [ ] **Step 2: Implement extractor** + +The normalized record should contain: + +```java +public record Jt808LocationPoint( + String vehicleKey, + String vin, + String phone, + Instant eventTime, + double longitude, + double latitude, + Double speedKmh, + Long statusFlag, + Double totalMileageKm +) {} +``` + +- [ ] **Step 3: Verify** + +Run: + +```bash +mvn -pl :vehicle-stat-service -Dtest=Jt808LocationPointExtractorTest test +``` + +Expected: tests pass. + +## Task 2: GPS Segment Calculator + +**Files:** +- Create: `.../jt808/Jt808GpsMileageCalculator.java` +- Test: `.../Jt808GpsMileageCalculatorTest.java` + +- [ ] **Step 1: Write failing tests** + +Cover: +- Valid adjacent points accumulate Haversine distance. +- `dt <= 0` increments out-of-order/non-positive counter. +- `dt > PT5M` increments long-gap counter and does not accumulate. +- implied speed `> 200 km/h` increments bad-jump counter and does not accumulate. +- speed integral uses adjacent average speed. + +- [ ] **Step 2: Implement calculator** + +Return a segment result with: +- `gpsKm` +- `speedIntegralKm` +- `used` +- `badJump` +- `longGap` +- `outOfOrder` + +- [ ] **Step 3: Verify** + +Run: + +```bash +mvn -pl :vehicle-stat-service -Dtest=Jt808GpsMileageCalculatorTest test +``` + +Expected: tests pass. + +## Task 3: Rolling State In Redis/Memory + +**Files:** +- Create: `.../jt808/Jt808DailyMileageState.java` +- Create: `.../jt808/Jt808MileageStateStore.java` +- Create: `.../jt808/RedisJt808MileageStateStore.java` +- Create: `.../jt808/InMemoryJt808MileageStateStore.java` +- Test: `.../Jt808DailyMileageStateTest.java` +- Test: `.../RedisJt808MileageStateStoreTest.java` + +- [ ] **Step 1: Write failing state tests** + +Cover: +- First point initializes state without distance. +- Ordered point updates GPS and speed mileage. +- Duplicate/older point is ignored. +- Monotonic `totalMileageKm` updates `odometerMileageKm`. +- Odometer rollback increments anomaly counter. + +- [ ] **Step 2: Implement state key** + +Use Redis key: + +```text +vehicle:mileage:jt808:daily:{statDate}:{vehicleKey} +``` + +Example: + +```text +vehicle:mileage:jt808:daily:2026-06-30:LKLG7C4E7NA774755 +vehicle:mileage:jt808:daily:2026-06-30:jt808:40692934289 +``` + +- [ ] **Step 3: Implement Redis store** + +Use JSON value and TTL `P3D`. Use DB selection through Spring Redis database config, not by issuing `SELECT` manually. + +- [ ] **Step 4: Verify** + +Run: + +```bash +mvn -pl :vehicle-stat-service -Dtest=Jt808DailyMileageStateTest,RedisJt808MileageStateStoreTest test +``` + +Expected: tests pass. + +## Task 4: MySQL Result Repository + +**Files:** +- Create: `.../jt808/Jt808DailyMileageResult.java` +- Create: `.../jt808/JdbcJt808DailyMileageRepository.java` +- Test: `.../JdbcJt808DailyMileageRepositoryTest.java` + +- [ ] **Step 1: Write failing repository tests** + +Use H2 MySQL mode. Cover: +- Schema initialization creates `vehicle_daily_mileage_jt808`. +- Upsert inserts first result. +- Upsert updates same `(stat_date, vehicle_key)`. +- Query by date and vehicle key returns latest result. + +- [ ] **Step 2: Implement repository** + +Use `JdbcTemplate` and: + +```sql +INSERT INTO vehicle_daily_mileage_jt808 (...) +VALUES (...) +ON DUPLICATE KEY UPDATE + vin = VALUES(vin), + phone = VALUES(phone), + first_event_time = VALUES(first_event_time), + last_event_time = VALUES(last_event_time), + gps_mileage_km = VALUES(gps_mileage_km), + speed_integral_km = VALUES(speed_integral_km), + odometer_mileage_km = VALUES(odometer_mileage_km), + accepted_points = VALUES(accepted_points), + bad_jump_segments = VALUES(bad_jump_segments), + long_gap_segments = VALUES(long_gap_segments), + out_of_order_points = VALUES(out_of_order_points), + odometer_anomalies = VALUES(odometer_anomalies), + data_quality = VALUES(data_quality) +``` + +- [ ] **Step 3: Verify** + +Run: + +```bash +mvn -pl :vehicle-stat-service -Dtest=JdbcJt808DailyMileageRepositoryTest test +``` + +Expected: tests pass. + +## Task 5: Kafka Processor Fan-Out + +**Files:** +- Create: `.../jt808/Jt808MileageStreamProcessor.java` +- Modify: `.../VehicleStatEnvelopeIngestor.java` +- Test: `.../Jt808MileageStreamProcessorTest.java` +- Test: `.../VehicleStatEnvelopeIngestorTest.java` + +- [ ] **Step 1: Write failing processor tests** + +Cover: +- Two JT808 location envelopes update state and upsert MySQL result. +- GB32960 envelope is ignored. +- VIN unknown uses phone key. +- Different natural days create different states. + +- [ ] **Step 2: Implement processor flow** + +Flow: + +```text +VehicleEnvelope bytes +-> Jt808LocationPointExtractor +-> statDate from eventTime at Asia/Shanghai +-> stateStore.load(key) or new state +-> state.apply(point, calculator) +-> stateStore.save(state) +-> mysqlRepository.upsert(state.toResult()) +``` + +- [ ] **Step 3: Modify ingestor** + +Keep the old `VehicleStatEventProcessor` path for compatibility, but add optional `Jt808MileageStreamProcessor`. The new processor ignores non-JT808 envelopes internally. + +- [ ] **Step 4: Verify** + +Run: + +```bash +mvn -pl :vehicle-stat-service -Dtest=Jt808MileageStreamProcessorTest,VehicleStatEnvelopeIngestorTest test +``` + +Expected: tests pass. + +## Task 6: Auto-Configuration And App Properties + +**Files:** +- Modify: `.../config/VehicleStatAutoConfiguration.java` +- Modify: `modules/apps/vehicle-analytics-app/src/main/resources/application.yml` +- Test: `.../config/VehicleStatAutoConfigurationTest.java` + +- [ ] **Step 1: Write failing auto-config tests** + +Cover: +- JT808 mileage beans are present when `lingniu.ingest.vehicle-stat.jt808-mileage.enabled=true`. +- Redis state store is used when `state.store=redis` and `StringRedisTemplate` exists. +- Memory state store is used when `state.store=memory`. +- MySQL repository initializes when config is present. + +- [ ] **Step 2: Wire beans** + +Create beans for: +- `Jt808LocationPointExtractor` +- `Jt808GpsMileageCalculator` +- `Jt808MileageStateStore` +- `JdbcJt808DailyMileageRepository` +- `Jt808MileageStreamProcessor` + +- [ ] **Step 3: Update analytics app Kafka binding** + +For `vehicleStatEnvelopeConsumerProcessor`, topic must be: + +```yaml +- ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1} +``` + +- [ ] **Step 4: Verify** + +Run: + +```bash +mvn -pl :vehicle-stat-service,:vehicle-analytics-app -am test +``` + +Expected: tests pass. + +## Task 7: Deployment Smoke Test + +**Files:** +- Modify: `deploy/portainer/docker-compose.yml` +- Modify: `docs/operations/gb32960-service-split-runbook.md` + +- [ ] **Step 1: Add environment variables** + +Add to `vehicle-analytics-app`: + +```yaml +REDIS_HOST: ${REDIS_HOST:-} +REDIS_PORT: ${REDIS_PORT:-6379} +REDIS_DATABASE: ${REDIS_DATABASE:-50} +REDIS_PASSWORD: ${REDIS_PASSWORD:-} +KAFKA_TOPIC_JT808_EVENT: ${KAFKA_TOPIC_JT808_EVENT:-vehicle.event.jt808.v1} +JT808_MILEAGE_ENABLED: ${JT808_MILEAGE_ENABLED:-true} +JT808_MILEAGE_STATE_STORE: ${JT808_MILEAGE_STATE_STORE:-redis} +JT808_MILEAGE_MYSQL_JDBC_URL: ${JT808_MILEAGE_MYSQL_JDBC_URL:-} +JT808_MILEAGE_MYSQL_USERNAME: ${JT808_MILEAGE_MYSQL_USERNAME:-} +JT808_MILEAGE_MYSQL_PASSWORD: ${JT808_MILEAGE_MYSQL_PASSWORD:-} +``` + +- [ ] **Step 2: Run live smoke** + +Start analytics app and verify: +- Redis keys appear under `vehicle:mileage:jt808:daily:*`. +- MySQL table `vehicle_daily_mileage_jt808` receives/upserts rows. +- Kafka consumer group `vehicle-stat` advances on `vehicle.event.jt808.v1`. + +- [ ] **Step 3: Check one vehicle** + +For a known active 808 vehicle, query: + +```sql +SELECT * +FROM vehicle_daily_mileage_jt808 +WHERE stat_date = CURRENT_DATE +ORDER BY gps_mileage_km DESC +LIMIT 20; +``` + +Expected: rows show increasing `accepted_points`, `last_event_time`, and mileage during the day. + +## Remaining Needed Support + +- MySQL password should be supplied through Nacos/env as `JT808_MILEAGE_MYSQL_PASSWORD`; do not commit it. +- Redis password value should be supplied through Nacos/env as `REDIS_PASSWORD`; do not commit it. + +## Risks And Decisions + +- Redis is rolling state, not the source of truth. MySQL is the durable daily result. +- Memory state is acceptable only for local/dev or emergency Redis outage; restart loses segment continuity until the next point. +- No historical backfill means the first deployment day only calculates from the consumer start offset unless Kafka group is reset. +- JT808 GPS-only mileage is an estimate. Rows with high `bad_jump_segments`, high `long_gap_segments`, or no VIN should be marked `PARTIAL`. + +## Self-Review + +- Spec coverage: JT808-only Kafka streaming, Redis/memory state, MySQL result persistence, confirmed topic, confirmed daily algorithm defaults, and no backfill are covered. +- Placeholder scan: The only open item is deliberately listed under Remaining Needed Support: MySQL connection identity if not already inherited from deployment. +- Type consistency: The plan uses the `Jt808*` class family throughout and removes cross-protocol verifier scope. diff --git a/modules/apps/vehicle-analytics-app/src/main/resources/application.yml b/modules/apps/vehicle-analytics-app/src/main/resources/application.yml index e1ad81e4..ac3bab4d 100644 --- a/modules/apps/vehicle-analytics-app/src/main/resources/application.yml +++ b/modules/apps/vehicle-analytics-app/src/main/resources/application.yml @@ -17,6 +17,17 @@ spring: threads: virtual: enabled: true + datasource: + url: ${MYSQL_JDBC_URL:} + username: ${MYSQL_USERNAME:} + password: ${MYSQL_PASSWORD:} + driver-class-name: com.mysql.cj.jdbc.Driver + data: + redis: + host: ${REDIS_HOST:} + port: ${REDIS_PORT:6379} + password: ${REDIS_PASSWORD:} + database: ${REDIS_DATABASE:50} server: port: ${HTTP_PORT:20300} @@ -37,8 +48,8 @@ lingniu: type: kafka bootstrap-servers: ${KAFKA_BROKERS:114.55.58.251:9092} topics: - realtime: ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} - dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1} + realtime: ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1} + dlq: ${KAFKA_TOPIC_JT808_DLQ:vehicle.dlq.jt808.v1} consumer: enabled: ${KAFKA_CONSUMER_ENABLED:true} client-id-prefix: ${KAFKA_CONSUMER_CLIENT_ID_PREFIX:vehicle-analytics} @@ -49,12 +60,12 @@ lingniu: enabled: ${VEHICLE_STATE_ENABLED:false} group-id: ${KAFKA_GROUP_STATE:vehicle-state} topics: - - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} + - ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1} vehicleStatEnvelopeConsumerProcessor: enabled: ${VEHICLE_STAT_ENABLED:true} group-id: ${KAFKA_GROUP_STAT:vehicle-stat} topics: - - ${KAFKA_TOPIC_GB32960_EVENT:vehicle.event.gb32960.v1} + - ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1} archive: enabled: false event-file-store: @@ -67,6 +78,15 @@ lingniu: enabled: ${VEHICLE_STAT_ENABLED:true} file-path: ${VEHICLE_STAT_FILE_PATH:./target/vehicle-stat/} zone-id: ${VEHICLE_STAT_ZONE_ID:Asia/Shanghai} + jt808: + enabled: ${VEHICLE_STAT_JT808_MILEAGE_ENABLED:false} + state-store: ${VEHICLE_STAT_JT808_STATE_STORE:redis} + redis-key-prefix: ${VEHICLE_STAT_JT808_REDIS_KEY_PREFIX:vehicle:mileage:jt808:daily:} + state-ttl-days: ${VEHICLE_STAT_JT808_STATE_TTL_DAYS:3} + max-segment-gap-seconds: ${VEHICLE_STAT_JT808_MAX_SEGMENT_GAP_SECONDS:300} + max-implied-speed-kmh: ${VEHICLE_STAT_JT808_MAX_IMPLIED_SPEED_KMH:200} + short-gap-seconds: ${VEHICLE_STAT_JT808_SHORT_GAP_SECONDS:10} + short-gap-jump-meters: ${VEHICLE_STAT_JT808_SHORT_GAP_JUMP_METERS:300} management: endpoints: diff --git a/modules/services/vehicle-stat-service/pom.xml b/modules/services/vehicle-stat-service/pom.xml index 231d2038..1324eba4 100644 --- a/modules/services/vehicle-stat-service/pom.xml +++ b/modules/services/vehicle-stat-service/pom.xml @@ -28,6 +28,18 @@ org.springframework.boot spring-boot-starter-web + + org.springframework.boot + spring-boot-starter-data-redis + + + org.springframework + spring-jdbc + + + com.mysql + mysql-connector-j + org.springframework.boot spring-boot-configuration-processor @@ -53,5 +65,10 @@ spring-boot-starter-test test + + com.h2database + h2 + test + diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java index aa8d2b39..a7bef3f6 100644 --- a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java @@ -4,16 +4,24 @@ import com.google.protobuf.InvalidProtocolBufferException; import com.lingniu.ingest.api.consumer.EnvelopeIngestResult; import com.lingniu.ingest.api.consumer.EnvelopeIngestor; import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; +import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStreamProcessor; public final class VehicleStatEnvelopeIngestor implements EnvelopeIngestor { private final VehicleStatEventProcessor processor; + private final Jt808MileageStreamProcessor jt808MileageProcessor; public VehicleStatEnvelopeIngestor(VehicleStatEventProcessor processor) { + this(processor, null); + } + + public VehicleStatEnvelopeIngestor(VehicleStatEventProcessor processor, + Jt808MileageStreamProcessor jt808MileageProcessor) { if (processor == null) { throw new IllegalArgumentException("processor must not be null"); } this.processor = processor; + this.jt808MileageProcessor = jt808MileageProcessor; } public void ingest(byte[] kafkaValue) { @@ -22,6 +30,7 @@ public final class VehicleStatEnvelopeIngestor implements EnvelopeIngestor { return; } processor.process(envelope); + processJt808Mileage(envelope); } @Override @@ -34,6 +43,7 @@ public final class VehicleStatEnvelopeIngestor implements EnvelopeIngestor { envelope.getEventId(), envelope.getVin(), "envelope telemetry_snapshot is required"); } processor.process(envelope); + processJt808Mileage(envelope); return EnvelopeIngestResult.processed(envelope.getEventId(), envelope.getVin()); } catch (IllegalArgumentException ex) { return envelope == null @@ -42,6 +52,12 @@ public final class VehicleStatEnvelopeIngestor implements EnvelopeIngestor { } } + private void processJt808Mileage(VehicleEnvelope envelope) { + if (jt808MileageProcessor != null) { + jt808MileageProcessor.process(envelope); + } + } + private static VehicleEnvelope parse(byte[] kafkaValue) { if (kafkaValue == null || kafkaValue.length == 0) { throw new IllegalArgumentException("VehicleEnvelope bytes must not be empty"); diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java index af098c28..12ae756a 100644 --- a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java @@ -13,14 +13,27 @@ import com.lingniu.ingest.vehiclestat.VehicleStatEventProcessor; import com.lingniu.ingest.vehiclestat.VehicleStatRepository; import com.lingniu.ingest.vehiclestat.VehicleStatRule; import com.lingniu.ingest.vehiclestat.VehicleStatRuleRepository; +import com.lingniu.ingest.vehiclestat.jt808.InMemoryJt808MileageStateStore; +import com.lingniu.ingest.vehiclestat.jt808.JdbcJt808DailyMileageRepository; +import com.lingniu.ingest.vehiclestat.jt808.Jt808DailyMileageRepository; +import com.lingniu.ingest.vehiclestat.jt808.Jt808GpsMileageCalculator; +import com.lingniu.ingest.vehiclestat.jt808.Jt808LocationPointExtractor; +import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStateStore; +import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStreamProcessor; +import com.lingniu.ingest.vehiclestat.jt808.RedisJt808MileageStateStore; +import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.context.annotation.Bean; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.jdbc.core.JdbcTemplate; import java.nio.file.Path; +import java.time.Duration; import java.time.ZoneId; @AutoConfiguration @@ -66,8 +79,70 @@ public class VehicleStatAutoConfiguration { @Bean @ConditionalOnBean(VehicleStatEventProcessor.class) @ConditionalOnMissingBean - public VehicleStatEnvelopeIngestor vehicleStatEnvelopeIngestor(VehicleStatEventProcessor processor) { - return new VehicleStatEnvelopeIngestor(processor); + public VehicleStatEnvelopeIngestor vehicleStatEnvelopeIngestor( + VehicleStatEventProcessor processor, + ObjectProvider jt808MileageProcessor) { + return new VehicleStatEnvelopeIngestor(processor, jt808MileageProcessor.getIfAvailable()); + } + + @Bean + @ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat.jt808", name = "enabled", havingValue = "true") + @ConditionalOnMissingBean + public Jt808LocationPointExtractor jt808LocationPointExtractor() { + return new Jt808LocationPointExtractor(); + } + + @Bean + @ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat.jt808", name = "enabled", havingValue = "true") + @ConditionalOnMissingBean + public Jt808GpsMileageCalculator jt808GpsMileageCalculator(VehicleStatProperties props) { + VehicleStatProperties.Jt808 jt808 = props.getJt808(); + return new Jt808GpsMileageCalculator( + Duration.ofSeconds(jt808.getMaxSegmentGapSeconds()), + jt808.getMaxImpliedSpeedKmh(), + Duration.ofSeconds(jt808.getShortGapSeconds()), + jt808.getShortGapJumpMeters()); + } + + @Bean + @ConditionalOnBean(JdbcTemplate.class) + @ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat.jt808", name = "enabled", havingValue = "true") + @ConditionalOnMissingBean + public Jt808DailyMileageRepository jt808DailyMileageRepository(JdbcTemplate jdbcTemplate) { + return new JdbcJt808DailyMileageRepository(jdbcTemplate); + } + + @Bean + @ConditionalOnBean({StringRedisTemplate.class, ObjectMapper.class}) + @ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat.jt808", name = "enabled", havingValue = "true") + @ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat.jt808", name = "state-store", havingValue = "redis") + @ConditionalOnMissingBean(Jt808MileageStateStore.class) + public Jt808MileageStateStore redisJt808MileageStateStore(StringRedisTemplate redis, + ObjectMapper objectMapper, + VehicleStatProperties props) { + VehicleStatProperties.Jt808 jt808 = props.getJt808(); + return new RedisJt808MileageStateStore(redis, objectMapper, + jt808.getRedisKeyPrefix(), Duration.ofDays(jt808.getStateTtlDays())); + } + + @Bean + @ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat.jt808", name = "enabled", havingValue = "true") + @ConditionalOnMissingBean(Jt808MileageStateStore.class) + public Jt808MileageStateStore inMemoryJt808MileageStateStore() { + return new InMemoryJt808MileageStateStore(); + } + + @Bean + @ConditionalOnBean({Jt808LocationPointExtractor.class, Jt808GpsMileageCalculator.class, + Jt808MileageStateStore.class, Jt808DailyMileageRepository.class}) + @ConditionalOnMissingBean + public Jt808MileageStreamProcessor jt808MileageStreamProcessor(Jt808LocationPointExtractor extractor, + Jt808GpsMileageCalculator calculator, + Jt808MileageStateStore stateStore, + Jt808DailyMileageRepository repository, + VehicleStatProperties props) { + return new Jt808MileageStreamProcessor(extractor, calculator, stateStore, repository, + ZoneId.of(props.getZoneId())); } @Bean diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatProperties.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatProperties.java index 9cde4bb1..ff20cfc4 100644 --- a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatProperties.java +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatProperties.java @@ -11,6 +11,8 @@ public class VehicleStatProperties { /** 统计自然日口径，默认按国内业务使用东八区。 */ private String zoneId = "Asia/Shanghai"; + private Jt808 jt808 = new Jt808(); + public String getFilePath() { return filePath; } @@ -26,4 +28,87 @@ public class VehicleStatProperties { public void setZoneId(String zoneId) { this.zoneId = zoneId; } + + public Jt808 getJt808() { + return jt808; + } + + public void setJt808(Jt808 jt808) { + this.jt808 = jt808; + } + + public static class Jt808 { + private boolean enabled; + private String stateStore = "memory"; + private String redisKeyPrefix = "vehicle:mileage:jt808:daily:"; + private long stateTtlDays = 3; + private long maxSegmentGapSeconds = 300; + private double maxImpliedSpeedKmh = 200.0; + private long shortGapSeconds = 10; + private double shortGapJumpMeters = 300.0; + + public boolean isEnabled() { + return enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + public String getStateStore() { + return stateStore; + } + + public void setStateStore(String stateStore) { + this.stateStore = stateStore; + } + + public String getRedisKeyPrefix() { + return redisKeyPrefix; + } + + public void setRedisKeyPrefix(String redisKeyPrefix) { + this.redisKeyPrefix = redisKeyPrefix; + } + + public long getStateTtlDays() { + return stateTtlDays; + } + + public void setStateTtlDays(long stateTtlDays) { + this.stateTtlDays = stateTtlDays; + } + + public long getMaxSegmentGapSeconds() { + return maxSegmentGapSeconds; + } + + public void setMaxSegmentGapSeconds(long maxSegmentGapSeconds) { + this.maxSegmentGapSeconds = maxSegmentGapSeconds; + } + + public double getMaxImpliedSpeedKmh() { + return maxImpliedSpeedKmh; + } + + public void setMaxImpliedSpeedKmh(double maxImpliedSpeedKmh) { + this.maxImpliedSpeedKmh = maxImpliedSpeedKmh; + } + + public long getShortGapSeconds() { + return shortGapSeconds; + } + + public void setShortGapSeconds(long shortGapSeconds) { + this.shortGapSeconds = shortGapSeconds; + } + + public double getShortGapJumpMeters() { + return shortGapJumpMeters; + } + + public void setShortGapJumpMeters(double shortGapJumpMeters) { + this.shortGapJumpMeters = shortGapJumpMeters; + } + } } diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/InMemoryJt808MileageStateStore.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/InMemoryJt808MileageStateStore.java new file mode 100644 index 00000000..f305e642 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/InMemoryJt808MileageStateStore.java @@ -0,0 +1,25 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import java.time.LocalDate; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; + +public final class InMemoryJt808MileageStateStore implements Jt808MileageStateStore { + + private final Map states = new ConcurrentHashMap<>(); + + @Override + public Optional load(String vehicleKey, LocalDate statDate) { + return Optional.ofNullable(states.get(key(vehicleKey, statDate))); + } + + @Override + public void save(Jt808DailyMileageState state) { + states.put(key(state.vehicleKey(), state.statDate()), state); + } + + private static String key(String vehicleKey, LocalDate statDate) { + return statDate + ":" + vehicleKey; + } +} diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepository.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepository.java new file mode 100644 index 00000000..dccc3149 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepository.java @@ -0,0 +1,61 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import org.springframework.jdbc.core.JdbcTemplate; + +import java.sql.Timestamp; +import java.time.Instant; + +public final class JdbcJt808DailyMileageRepository implements Jt808DailyMileageRepository { + + private final JdbcTemplate jdbc; + + public JdbcJt808DailyMileageRepository(JdbcTemplate jdbc) { + this.jdbc = jdbc; + } + + @Override + public void upsert(Jt808DailyMileageResult result) { + jdbc.update(""" + insert into vehicle_daily_mileage_jt808 ( + stat_date, vehicle_key, vin, phone, first_event_time, last_event_time, + gps_mileage_km, speed_integral_km, odometer_mileage_km, + accepted_points, bad_jump_segments, long_gap_segments, out_of_order_points, + odometer_anomalies, data_quality + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + on duplicate key update + vin = values(vin), + phone = values(phone), + first_event_time = values(first_event_time), + last_event_time = values(last_event_time), + gps_mileage_km = values(gps_mileage_km), + speed_integral_km = values(speed_integral_km), + odometer_mileage_km = values(odometer_mileage_km), + accepted_points = values(accepted_points), + bad_jump_segments = values(bad_jump_segments), + long_gap_segments = values(long_gap_segments), + out_of_order_points = values(out_of_order_points), + odometer_anomalies = values(odometer_anomalies), + data_quality = values(data_quality), + updated_at = current_timestamp + """, + result.statDate(), + result.vehicleKey(), + result.vin(), + result.phone(), + timestamp(result.firstEventTime()), + timestamp(result.lastEventTime()), + result.gpsMileageKm(), + result.speedIntegralKm(), + result.odometerMileageKm(), + result.acceptedPoints(), + result.badJumpSegments(), + result.longGapSegments(), + result.outOfOrderPoints(), + result.odometerAnomalies(), + result.dataQuality()); + } + + private static Timestamp timestamp(Instant instant) { + return instant == null ? null : Timestamp.from(instant); + } +} diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageRepository.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageRepository.java new file mode 100644 index 00000000..c1ce0ea1 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageRepository.java @@ -0,0 +1,6 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +public interface Jt808DailyMileageRepository { + + void upsert(Jt808DailyMileageResult result); +} diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageResult.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageResult.java new file mode 100644 index 00000000..43b4b4b1 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageResult.java @@ -0,0 +1,22 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import java.time.Instant; +import java.time.LocalDate; + +public record Jt808DailyMileageResult( + LocalDate statDate, + String vehicleKey, + String vin, + String phone, + Instant firstEventTime, + Instant lastEventTime, + double gpsMileageKm, + double speedIntegralKm, + Double odometerMileageKm, + int acceptedPoints, + int badJumpSegments, + int longGapSegments, + int outOfOrderPoints, + int odometerAnomalies, + String dataQuality +) {} diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageState.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageState.java new file mode 100644 index 00000000..83d9259f --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageState.java @@ -0,0 +1,223 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import java.time.Instant; +import java.time.LocalDate; + +public final class Jt808DailyMileageState { + + private final String vehicleKey; + private final String vin; + private final String phone; + private final LocalDate statDate; + private Instant firstEventTime; + private Instant lastEventTime; + private Jt808LocationPoint lastPoint; + private double lastTotalMileageKm = Double.NaN; + private double gpsMileageKm; + private double speedIntegralKm; + private double odometerMileageKm; + private int acceptedPoints; + private int badJumpSegments; + private int longGapSegments; + private int outOfOrderPoints; + private int odometerAnomalies; + + private Jt808DailyMileageState(String vehicleKey, String vin, String phone, LocalDate statDate) { + if (vehicleKey == null || vehicleKey.isBlank()) { + throw new IllegalArgumentException("vehicleKey must not be blank"); + } + if (statDate == null) { + throw new IllegalArgumentException("statDate must not be null"); + } + this.vehicleKey = vehicleKey; + this.vin = vin == null ? "" : vin; + this.phone = phone == null ? "" : phone; + this.statDate = statDate; + } + + public static Jt808DailyMileageState empty(String vehicleKey, String vin, String phone, LocalDate statDate) { + return new Jt808DailyMileageState(vehicleKey, vin, phone, statDate); + } + + public static Jt808DailyMileageState restore( + String vehicleKey, + String vin, + String phone, + LocalDate statDate, + Instant firstEventTime, + Instant lastEventTime, + Jt808LocationPoint lastPoint, + double lastTotalMileageKm, + double gpsMileageKm, + double speedIntegralKm, + double odometerMileageKm, + int acceptedPoints, + int badJumpSegments, + int longGapSegments, + int outOfOrderPoints, + int odometerAnomalies) { + Jt808DailyMileageState state = new Jt808DailyMileageState(vehicleKey, vin, phone, statDate); + state.firstEventTime = firstEventTime; + state.lastEventTime = lastEventTime; + state.lastPoint = lastPoint; + state.lastTotalMileageKm = lastTotalMileageKm; + state.gpsMileageKm = gpsMileageKm; + state.speedIntegralKm = speedIntegralKm; + state.odometerMileageKm = odometerMileageKm; + state.acceptedPoints = acceptedPoints; + state.badJumpSegments = badJumpSegments; + state.longGapSegments = longGapSegments; + state.outOfOrderPoints = outOfOrderPoints; + state.odometerAnomalies = odometerAnomalies; + return state; + } + + public void apply(Jt808LocationPoint point, Jt808GpsMileageCalculator calculator) { + if (point == null || calculator == null) { + return; + } + if (lastPoint != null && !point.eventTime().isAfter(lastPoint.eventTime())) { + outOfOrderPoints++; + return; + } + if (lastPoint == null) { + firstEventTime = point.eventTime(); + updateOdometer(point); + accept(point); + return; + } + + Jt808GpsMileageCalculator.SegmentResult segment = calculator.calculateSegment(lastPoint, point); + if (segment.outOfOrder()) { + outOfOrderPoints++; + return; + } + if (segment.longGap()) { + longGapSegments++; + } else if (segment.badJump()) { + badJumpSegments++; + } else if (segment.used()) { + gpsMileageKm += segment.gpsKm(); + speedIntegralKm += segment.speedIntegralKm(); + } + updateOdometer(point); + accept(point); + } + + public Jt808DailyMileageResult toResult() { + return new Jt808DailyMileageResult( + statDate, + vehicleKey, + vin, + phone, + firstEventTime, + lastEventTime, + gpsMileageKm, + speedIntegralKm, + odometerMileageKm > 0.0 ? odometerMileageKm : null, + acceptedPoints, + badJumpSegments, + longGapSegments, + outOfOrderPoints, + odometerAnomalies, + dataQuality()); + } + + private void accept(Jt808LocationPoint point) { + lastPoint = point; + lastEventTime = point.eventTime(); + acceptedPoints++; + } + + private void updateOdometer(Jt808LocationPoint point) { + if (point.totalMileageKm() == null || !Double.isFinite(point.totalMileageKm())) { + return; + } + double current = point.totalMileageKm(); + if (!Double.isFinite(lastTotalMileageKm)) { + lastTotalMileageKm = current; + return; + } + if (current < lastTotalMileageKm) { + odometerAnomalies++; + lastTotalMileageKm = current; + return; + } + odometerMileageKm += current - lastTotalMileageKm; + lastTotalMileageKm = current; + } + + private String dataQuality() { + if (acceptedPoints < 2) { + return "PARTIAL"; + } + if (badJumpSegments > 0 || longGapSegments > 0 || vehicleKey.startsWith("jt808:")) { + return "PARTIAL"; + } + return "GOOD"; + } + + public String vehicleKey() { + return vehicleKey; + } + + public String vin() { + return vin; + } + + public String phone() { + return phone; + } + + public LocalDate statDate() { + return statDate; + } + + public Instant firstEventTime() { + return firstEventTime; + } + + public Instant lastEventTime() { + return lastEventTime; + } + + public Jt808LocationPoint lastPoint() { + return lastPoint; + } + + public double gpsMileageKm() { + return gpsMileageKm; + } + + public double speedIntegralKm() { + return speedIntegralKm; + } + + public double odometerMileageKm() { + return odometerMileageKm; + } + + public double lastTotalMileageKm() { + return lastTotalMileageKm; + } + + public int acceptedPoints() { + return acceptedPoints; + } + + public int badJumpSegments() { + return badJumpSegments; + } + + public int longGapSegments() { + return longGapSegments; + } + + public int outOfOrderPoints() { + return outOfOrderPoints; + } + + public int odometerAnomalies() { + return odometerAnomalies; + } +} diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculator.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculator.java new file mode 100644 index 00000000..d25c9739 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculator.java @@ -0,0 +1,93 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import java.time.Duration; + +public final class Jt808GpsMileageCalculator { + + private static final double EARTH_RADIUS_KM = 6371.0088; + + private final Duration maxSegmentGap; + private final double maxImpliedSpeedKmh; + private final Duration shortGap; + private final double shortGapJumpMeters; + + public Jt808GpsMileageCalculator(Duration maxSegmentGap, + double maxImpliedSpeedKmh, + Duration shortGap, + double shortGapJumpMeters) { + if (maxSegmentGap == null || maxSegmentGap.isNegative() || maxSegmentGap.isZero()) { + throw new IllegalArgumentException("maxSegmentGap must be positive"); + } + if (shortGap == null || shortGap.isNegative() || shortGap.isZero()) { + throw new IllegalArgumentException("shortGap must be positive"); + } + if (!Double.isFinite(maxImpliedSpeedKmh) || maxImpliedSpeedKmh <= 0) { + throw new IllegalArgumentException("maxImpliedSpeedKmh must be positive"); + } + if (!Double.isFinite(shortGapJumpMeters) || shortGapJumpMeters <= 0) { + throw new IllegalArgumentException("shortGapJumpMeters must be positive"); + } + this.maxSegmentGap = maxSegmentGap; + this.maxImpliedSpeedKmh = maxImpliedSpeedKmh; + this.shortGap = shortGap; + this.shortGapJumpMeters = shortGapJumpMeters; + } + + public SegmentResult calculateSegment(Jt808LocationPoint previous, Jt808LocationPoint current) { + if (previous == null || current == null) { + return SegmentResult.rejectedOutOfOrder(); + } + Duration elapsed = Duration.between(previous.eventTime(), current.eventTime()); + if (elapsed.isZero() || elapsed.isNegative()) { + return SegmentResult.rejectedOutOfOrder(); + } + if (elapsed.compareTo(maxSegmentGap) > 0) { + return SegmentResult.rejectedLongGap(); + } + + double gpsKm = haversineKm(previous.longitude(), previous.latitude(), current.longitude(), current.latitude()); + double hours = elapsed.toMillis() / 3_600_000.0; + double impliedSpeed = gpsKm / hours; + if (impliedSpeed > maxImpliedSpeedKmh + || (elapsed.compareTo(shortGap) <= 0 && gpsKm * 1000.0 > shortGapJumpMeters)) { + return SegmentResult.rejectedBadJump(); + } + + double speedIntegralKm = 0.0; + if (previous.speedKmh() != null && current.speedKmh() != null + && Double.isFinite(previous.speedKmh()) && Double.isFinite(current.speedKmh())) { + speedIntegralKm = ((previous.speedKmh() + current.speedKmh()) / 2.0) * hours; + } + return new SegmentResult(gpsKm, speedIntegralKm, true, false, false, false); + } + + private static double haversineKm(double lon1, double lat1, double lon2, double lat2) { + double latRad1 = Math.toRadians(lat1); + double latRad2 = Math.toRadians(lat2); + double deltaLat = Math.toRadians(lat2 - lat1); + double deltaLon = Math.toRadians(lon2 - lon1); + double a = Math.sin(deltaLat / 2.0) * Math.sin(deltaLat / 2.0) + + Math.cos(latRad1) * Math.cos(latRad2) + * Math.sin(deltaLon / 2.0) * Math.sin(deltaLon / 2.0); + return 2.0 * EARTH_RADIUS_KM * Math.asin(Math.min(1.0, Math.sqrt(a))); + } + + public record SegmentResult(double gpsKm, + double speedIntegralKm, + boolean used, + boolean badJump, + boolean longGap, + boolean outOfOrder) { + private static SegmentResult rejectedBadJump() { + return new SegmentResult(0.0, 0.0, false, true, false, false); + } + + private static SegmentResult rejectedLongGap() { + return new SegmentResult(0.0, 0.0, false, false, true, false); + } + + private static SegmentResult rejectedOutOfOrder() { + return new SegmentResult(0.0, 0.0, false, false, false, true); + } + } +} diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPoint.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPoint.java new file mode 100644 index 00000000..dcccdb5f --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPoint.java @@ -0,0 +1,27 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import java.time.Instant; + +public record Jt808LocationPoint( + String vehicleKey, + String vin, + String phone, + Instant eventTime, + double longitude, + double latitude, + Double speedKmh, + Long statusFlag, + Double totalMileageKm +) { + + public Jt808LocationPoint { + if (vehicleKey == null || vehicleKey.isBlank()) { + throw new IllegalArgumentException("vehicleKey must not be blank"); + } + if (eventTime == null) { + throw new IllegalArgumentException("eventTime must not be null"); + } + vin = vin == null ? "" : vin; + phone = phone == null ? "" : phone; + } +} diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPointExtractor.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPointExtractor.java new file mode 100644 index 00000000..afbf57e0 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPointExtractor.java @@ -0,0 +1,96 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import com.lingniu.ingest.sink.mq.proto.TelemetryField; +import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; + +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Optional; + +public final class Jt808LocationPointExtractor { + + public Optional extract(VehicleEnvelope envelope) { + if (envelope == null + || !"JT808".equalsIgnoreCase(envelope.getSource()) + || !envelope.hasTelemetrySnapshot()) { + return Optional.empty(); + } + Map fields = fields(envelope); + Double longitude = decimal(fields.get("longitude")); + Double latitude = decimal(fields.get("latitude")); + if (longitude == null || latitude == null || !validCoordinate(longitude, latitude)) { + return Optional.empty(); + } + + String vin = clean(envelope.getVin()); + String phone = clean(envelope.getMetadataMap().getOrDefault("phone", "")); + String vehicleKey = vehicleKey(vin, phone); + if (vehicleKey.isBlank()) { + return Optional.empty(); + } + return Optional.of(new Jt808LocationPoint( + vehicleKey, + vin, + phone, + Instant.ofEpochMilli(envelope.getEventTimeMs()), + longitude, + latitude, + decimal(fields.get("speed_kmh")), + integer(fields.get("location_status_raw")), + decimal(fields.get("total_mileage_km")))); + } + + private static Map fields(VehicleEnvelope envelope) { + Map out = new LinkedHashMap<>(); + for (TelemetryField field : envelope.getTelemetrySnapshot().getFieldsList()) { + if (!field.getKey().isBlank()) { + out.put(field.getKey(), field.getValue()); + } + } + return out; + } + + private static String vehicleKey(String vin, String phone) { + if (!vin.isBlank() && !"unknown".equalsIgnoreCase(vin)) { + return vin; + } + return phone.isBlank() ? "" : "jt808:" + phone; + } + + private static boolean validCoordinate(double longitude, double latitude) { + return Double.isFinite(longitude) + && Double.isFinite(latitude) + && longitude >= -180.0 + && longitude <= 180.0 + && latitude >= -90.0 + && latitude <= 90.0; + } + + private static Double decimal(String raw) { + if (raw == null || raw.isBlank()) { + return null; + } + try { + double value = Double.parseDouble(raw); + return Double.isFinite(value) ? value : null; + } catch (NumberFormatException ignored) { + return null; + } + } + + private static Long integer(String raw) { + if (raw == null || raw.isBlank()) { + return null; + } + try { + return Long.parseLong(raw); + } catch (NumberFormatException ignored) { + return null; + } + } + + private static String clean(String value) { + return value == null ? "" : value.trim(); + } +} diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStateStore.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStateStore.java new file mode 100644 index 00000000..a9e3c43e --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStateStore.java @@ -0,0 +1,11 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import java.time.LocalDate; +import java.util.Optional; + +public interface Jt808MileageStateStore { + + Optional load(String vehicleKey, LocalDate statDate); + + void save(Jt808DailyMileageState state); +} diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessor.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessor.java new file mode 100644 index 00000000..cc6286ae --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessor.java @@ -0,0 +1,41 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; + +import java.time.LocalDate; +import java.time.ZoneId; + +public final class Jt808MileageStreamProcessor { + + private final Jt808LocationPointExtractor extractor; + private final Jt808GpsMileageCalculator calculator; + private final Jt808MileageStateStore stateStore; + private final Jt808DailyMileageRepository repository; + private final ZoneId zoneId; + + public Jt808MileageStreamProcessor( + Jt808LocationPointExtractor extractor, + Jt808GpsMileageCalculator calculator, + Jt808MileageStateStore stateStore, + Jt808DailyMileageRepository repository, + ZoneId zoneId) { + this.extractor = extractor; + this.calculator = calculator; + this.stateStore = stateStore; + this.repository = repository; + this.zoneId = zoneId; + } + + public synchronized void process(VehicleEnvelope envelope) { + extractor.extract(envelope).ifPresent(this::processPoint); + } + + private void processPoint(Jt808LocationPoint point) { + LocalDate statDate = LocalDate.ofInstant(point.eventTime(), zoneId); + Jt808DailyMileageState state = stateStore.load(point.vehicleKey(), statDate) + .orElseGet(() -> Jt808DailyMileageState.empty(point.vehicleKey(), point.vin(), point.phone(), statDate)); + state.apply(point, calculator); + stateStore.save(state); + repository.upsert(state.toResult()); + } +} diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStore.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStore.java new file mode 100644 index 00000000..0c6c6002 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStore.java @@ -0,0 +1,167 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.time.Duration; +import java.time.Instant; +import java.time.LocalDate; +import java.util.Optional; + +public final class RedisJt808MileageStateStore implements Jt808MileageStateStore { + + private final StringRedisTemplate redis; + private final ObjectMapper objectMapper; + private final String keyPrefix; + private final Duration ttl; + + public RedisJt808MileageStateStore( + StringRedisTemplate redis, + ObjectMapper objectMapper, + String keyPrefix, + Duration ttl) { + this.redis = redis; + this.objectMapper = objectMapper; + this.keyPrefix = keyPrefix; + this.ttl = ttl; + } + + @Override + public Optional load(String vehicleKey, LocalDate statDate) { + String json = redis.opsForValue().get(key(vehicleKey, statDate)); + if (json == null || json.isBlank()) { + return Optional.empty(); + } + try { + return Optional.of(objectMapper.readValue(json, StateSnapshot.class).toState()); + } catch (JsonProcessingException e) { + throw new IllegalStateException("Failed to deserialize JT808 mileage state", e); + } + } + + @Override + public void save(Jt808DailyMileageState state) { + try { + redis.opsForValue().set(key(state.vehicleKey(), state.statDate()), + objectMapper.writeValueAsString(StateSnapshot.from(state)), ttl); + } catch (JsonProcessingException e) { + throw new IllegalStateException("Failed to serialize JT808 mileage state", e); + } + } + + private String key(String vehicleKey, LocalDate statDate) { + return keyPrefix + statDate + ":" + vehicleKey; + } + + @JsonIgnoreProperties(ignoreUnknown = true) + record StateSnapshot( + String vehicleKey, + String vin, + String phone, + String statDate, + String firstEventTime, + String lastEventTime, + PointSnapshot lastPoint, + double lastTotalMileageKm, + double gpsMileageKm, + double speedIntegralKm, + double odometerMileageKm, + int acceptedPoints, + int badJumpSegments, + int longGapSegments, + int outOfOrderPoints, + int odometerAnomalies) { + + static StateSnapshot from(Jt808DailyMileageState state) { + return new StateSnapshot( + state.vehicleKey(), + state.vin(), + state.phone(), + state.statDate().toString(), + format(state.firstEventTime()), + format(state.lastEventTime()), + PointSnapshot.from(state.lastPoint()), + state.lastTotalMileageKm(), + state.gpsMileageKm(), + state.speedIntegralKm(), + state.odometerMileageKm(), + state.acceptedPoints(), + state.badJumpSegments(), + state.longGapSegments(), + state.outOfOrderPoints(), + state.odometerAnomalies()); + } + + Jt808DailyMileageState toState() { + return Jt808DailyMileageState.restore( + vehicleKey, + vin, + phone, + LocalDate.parse(statDate), + parseInstant(firstEventTime), + parseInstant(lastEventTime), + lastPoint == null ? null : lastPoint.toPoint(), + lastTotalMileageKm, + gpsMileageKm, + speedIntegralKm, + odometerMileageKm, + acceptedPoints, + badJumpSegments, + longGapSegments, + outOfOrderPoints, + odometerAnomalies); + } + } + + @JsonIgnoreProperties(ignoreUnknown = true) + record PointSnapshot( + String vehicleKey, + String vin, + String phone, + String eventTime, + double longitude, + double latitude, + Double speedKmh, + Long statusFlag, + Double totalMileageKm) { + + static PointSnapshot from(Jt808LocationPoint point) { + if (point == null) { + return null; + } + return new PointSnapshot( + point.vehicleKey(), + point.vin(), + point.phone(), + format(point.eventTime()), + point.longitude(), + point.latitude(), + point.speedKmh(), + point.statusFlag(), + point.totalMileageKm()); + } + + Jt808LocationPoint toPoint() { + return new Jt808LocationPoint( + vehicleKey, + vin, + phone, + Instant.parse(eventTime), + longitude, + latitude, + speedKmh, + statusFlag, + totalMileageKm); + } + } + + private static String format(Instant instant) { + return instant == null ? null : instant.toString(); + } + + private static Instant parseInstant(String value) { + return value == null || value.isBlank() ? null : Instant.parse(value); + } +} diff --git a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestorTest.java b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestorTest.java index efceed2e..ac8a12a8 100644 --- a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestorTest.java +++ b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestorTest.java @@ -6,6 +6,7 @@ import com.lingniu.ingest.sink.mq.proto.RawArchiveRef; import com.lingniu.ingest.sink.mq.proto.TelemetryField; import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot; import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; +import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStreamProcessor; import org.junit.jupiter.api.Test; import java.time.LocalDate; @@ -14,6 +15,8 @@ import java.util.Optional; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; class VehicleStatEnvelopeIngestorTest { @@ -31,6 +34,20 @@ class VehicleStatEnvelopeIngestorTest { .containsExactly(123.45); } + @Test + void fansOutTelemetryEnvelopeToOptionalJt808MileageProcessor() { + InMemoryVehicleStatRepository repository = new InMemoryVehicleStatRepository(); + Jt808MileageStreamProcessor jt808Processor = mock(Jt808MileageStreamProcessor.class); + VehicleEnvelope envelope = envelope(); + VehicleStatEnvelopeIngestor ingestor = new VehicleStatEnvelopeIngestor( + new VehicleStatEventProcessor(repository), + jt808Processor); + + ingestor.ingest(envelope.toByteArray()); + + verify(jt808Processor).process(envelope); + } + @Test void rejectsInvalidEnvelopeBytes() { VehicleStatEnvelopeIngestor ingestor = diff --git a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfigurationTest.java b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfigurationTest.java index e4a39a78..a94029d0 100644 --- a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfigurationTest.java +++ b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfigurationTest.java @@ -14,15 +14,23 @@ import com.lingniu.ingest.vehiclestat.VehicleStatEventSink; import com.lingniu.ingest.vehiclestat.VehicleStatEventProcessor; import com.lingniu.ingest.vehiclestat.VehicleStatRepository; import com.lingniu.ingest.vehiclestat.VehicleStatRuleRepository; +import com.lingniu.ingest.vehiclestat.jt808.Jt808DailyMileageRepository; +import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStateStore; +import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStreamProcessor; +import com.lingniu.ingest.vehiclestat.jt808.RedisJt808MileageStateStore; +import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.jdbc.core.JdbcTemplate; import java.time.LocalDate; import java.util.List; import java.util.Optional; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; class VehicleStatAutoConfigurationTest { @@ -74,6 +82,41 @@ class VehicleStatAutoConfigurationTest { }); } + @Test + void createsJt808MileageBeansWhenEnabledWithJdbcAndRedis() { + contextRunner + .withBean(JdbcTemplate.class, () -> mock(JdbcTemplate.class)) + .withBean(StringRedisTemplate.class, () -> mock(StringRedisTemplate.class)) + .withBean(ObjectMapper.class, ObjectMapper::new) + .withPropertyValues( + "lingniu.ingest.vehicle-stat.enabled=true", + "lingniu.ingest.vehicle-stat.jt808.enabled=true", + "lingniu.ingest.vehicle-stat.jt808.state-store=redis") + .run(context -> { + assertThat(context).hasSingleBean(Jt808DailyMileageRepository.class); + assertThat(context).hasSingleBean(Jt808MileageStateStore.class); + assertThat(context).hasSingleBean(RedisJt808MileageStateStore.class); + assertThat(context).hasSingleBean(Jt808MileageStreamProcessor.class); + assertThat(context).hasSingleBean(VehicleStatEnvelopeIngestor.class); + }); + } + + @Test + void fallsBackToInMemoryJt808StateStoreWhenRedisIsNotSelected() { + contextRunner + .withBean(JdbcTemplate.class, () -> mock(JdbcTemplate.class)) + .withPropertyValues( + "lingniu.ingest.vehicle-stat.enabled=true", + "lingniu.ingest.vehicle-stat.jt808.enabled=true", + "lingniu.ingest.vehicle-stat.jt808.state-store=memory") + .run(context -> { + assertThat(context).hasSingleBean(Jt808DailyMileageRepository.class); + assertThat(context).hasSingleBean(Jt808MileageStateStore.class); + assertThat(context).doesNotHaveBean(RedisJt808MileageStateStore.class); + assertThat(context).hasSingleBean(Jt808MileageStreamProcessor.class); + }); + } + private static final class InMemoryVehicleStatRepository implements VehicleStatRepository { @Override public void appendMileagePoint(String vin, MileagePoint point) {} @Override public List mileagePoints(String vin, LocalDate statDate) { return List.of(); } diff --git a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepositoryTest.java b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepositoryTest.java new file mode 100644 index 00000000..62ec0dcd --- /dev/null +++ b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepositoryTest.java @@ -0,0 +1,102 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DriverManagerDataSource; + +import java.time.Instant; +import java.time.LocalDate; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +class JdbcJt808DailyMileageRepositoryTest { + + private JdbcTemplate jdbc; + private JdbcJt808DailyMileageRepository repository; + + @BeforeEach + void setUp() { + DriverManagerDataSource dataSource = new DriverManagerDataSource( + "jdbc:h2:mem:jt808_mileage;MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1", + "sa", + ""); + jdbc = new JdbcTemplate(dataSource); + jdbc.execute("drop table if exists vehicle_daily_mileage_jt808"); + jdbc.execute(""" + create table vehicle_daily_mileage_jt808 ( + stat_date date not null, + vehicle_key varchar(64) not null, + vin varchar(32) null, + phone varchar(32) null, + first_event_time timestamp null, + last_event_time timestamp null, + gps_mileage_km decimal(12,3) not null, + speed_integral_km decimal(12,3) not null, + odometer_mileage_km decimal(12,3) null, + accepted_points int not null, + bad_jump_segments int not null, + long_gap_segments int not null, + out_of_order_points int not null, + odometer_anomalies int not null, + data_quality varchar(16) not null, + updated_at timestamp not null default current_timestamp, + primary key (stat_date, vehicle_key) + ) + """); + repository = new JdbcJt808DailyMileageRepository(jdbc); + } + + @Test + void upsertsDailyMileageResult() { + Jt808DailyMileageResult first = new Jt808DailyMileageResult( + LocalDate.of(2026, 6, 30), + "VIN001", + "VIN001", + "13900000001", + Instant.parse("2026-06-30T00:00:00Z"), + Instant.parse("2026-06-30T00:01:00Z"), + 1.23456, + 1.2, + null, + 2, + 0, + 0, + 0, + 0, + "GOOD"); + Jt808DailyMileageResult updated = new Jt808DailyMileageResult( + LocalDate.of(2026, 6, 30), + "VIN001", + "VIN001", + "13900000001", + Instant.parse("2026-06-30T00:00:00Z"), + Instant.parse("2026-06-30T00:02:00Z"), + 2.34567, + 2.3, + 2.0, + 3, + 1, + 0, + 0, + 0, + "PARTIAL"); + + repository.upsert(first); + repository.upsert(updated); + + Map row = jdbc.queryForMap(""" + select vehicle_key, gps_mileage_km, odometer_mileage_km, accepted_points, + bad_jump_segments, data_quality + from vehicle_daily_mileage_jt808 + where stat_date = '2026-06-30' and vehicle_key = 'VIN001' + """); + assertThat(row.get("vehicle_key")).isEqualTo("VIN001"); + assertThat(row.get("gps_mileage_km").toString()).isEqualTo("2.346"); + assertThat(row.get("odometer_mileage_km").toString()).isEqualTo("2.000"); + assertThat(row.get("accepted_points")).isEqualTo(3); + assertThat(row.get("bad_jump_segments")).isEqualTo(1); + assertThat(row.get("data_quality")).isEqualTo("PARTIAL"); + } +} diff --git a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageStateTest.java b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageStateTest.java new file mode 100644 index 00000000..033afe10 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageStateTest.java @@ -0,0 +1,89 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.time.Instant; +import java.time.LocalDate; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.data.Offset.offset; + +class Jt808DailyMileageStateTest { + + private final Jt808GpsMileageCalculator calculator = + new Jt808GpsMileageCalculator(Duration.ofMinutes(5), 200.0, Duration.ofSeconds(10), 300.0); + + @Test + void firstPointInitializesStateWithoutDistance() { + Jt808DailyMileageState state = Jt808DailyMileageState.empty("VIN001", "VIN001", "13900000001", + LocalDate.of(2026, 6, 30)); + + state.apply(point("2026-06-30T00:00:00Z", 120.000000, 30.000000, 36.0, 100.0), calculator); + + assertThat(state.acceptedPoints()).isEqualTo(1); + assertThat(state.gpsMileageKm()).isZero(); + assertThat(state.speedIntegralKm()).isZero(); + assertThat(state.odometerMileageKm()).isZero(); + assertThat(state.firstEventTime()).isEqualTo(Instant.parse("2026-06-30T00:00:00Z")); + assertThat(state.lastEventTime()).isEqualTo(Instant.parse("2026-06-30T00:00:00Z")); + } + + @Test + void orderedPointUpdatesGpsSpeedAndOdometerMileage() { + Jt808DailyMileageState state = Jt808DailyMileageState.empty("VIN001", "VIN001", "13900000001", + LocalDate.of(2026, 6, 30)); + state.apply(point("2026-06-30T00:00:00Z", 120.000000, 30.000000, 36.0, 100.0), calculator); + + state.apply(point("2026-06-30T00:01:40Z", 120.010370, 30.000000, 36.0, 101.2), calculator); + + assertThat(state.acceptedPoints()).isEqualTo(2); + assertThat(state.gpsMileageKm()).isBetween(0.99, 1.01); + assertThat(state.speedIntegralKm()).isBetween(0.99, 1.01); + assertThat(state.odometerMileageKm()).isCloseTo(1.2, offset(0.000001)); + assertThat(state.badJumpSegments()).isZero(); + assertThat(state.longGapSegments()).isZero(); + } + + @Test + void duplicateOrOlderPointIsIgnored() { + Jt808DailyMileageState state = Jt808DailyMileageState.empty("VIN001", "VIN001", "13900000001", + LocalDate.of(2026, 6, 30)); + state.apply(point("2026-06-30T00:01:00Z", 120.000000, 30.000000, 36.0, 100.0), calculator); + + state.apply(point("2026-06-30T00:00:59Z", 120.010370, 30.000000, 36.0, 101.0), calculator); + + assertThat(state.acceptedPoints()).isEqualTo(1); + assertThat(state.outOfOrderPoints()).isEqualTo(1); + assertThat(state.gpsMileageKm()).isZero(); + } + + @Test + void odometerRollbackIsIgnoredAndCounted() { + Jt808DailyMileageState state = Jt808DailyMileageState.empty("VIN001", "VIN001", "13900000001", + LocalDate.of(2026, 6, 30)); + state.apply(point("2026-06-30T00:00:00Z", 120.000000, 30.000000, 36.0, 100.0), calculator); + + state.apply(point("2026-06-30T00:01:40Z", 120.010370, 30.000000, 36.0, 99.5), calculator); + + assertThat(state.odometerMileageKm()).isZero(); + assertThat(state.odometerAnomalies()).isEqualTo(1); + } + + private static Jt808LocationPoint point(String time, + double longitude, + double latitude, + Double speedKmh, + Double totalMileageKm) { + return new Jt808LocationPoint( + "VIN001", + "VIN001", + "13900000001", + Instant.parse(time), + longitude, + latitude, + speedKmh, + 3L, + totalMileageKm); + } +} diff --git a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculatorTest.java b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculatorTest.java new file mode 100644 index 00000000..868c739e --- /dev/null +++ b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculatorTest.java @@ -0,0 +1,86 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.time.Instant; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.data.Offset.offset; + +class Jt808GpsMileageCalculatorTest { + + private final Jt808GpsMileageCalculator calculator = + new Jt808GpsMileageCalculator(Duration.ofMinutes(5), 200.0, Duration.ofSeconds(10), 300.0); + + @Test + void accumulatesValidHaversineDistanceAndSpeedIntegral() { + var result = calculator.calculateSegment( + point("2026-06-30T00:00:00Z", 120.000000, 30.000000, 36.0), + point("2026-06-30T00:01:40Z", 120.010370, 30.000000, 36.0)); + + assertThat(result.used()).isTrue(); + assertThat(result.gpsKm()).isCloseTo(1.0, offset(0.01)); + assertThat(result.speedIntegralKm()).isCloseTo(1.0, offset(0.001)); + assertThat(result.badJump()).isFalse(); + assertThat(result.longGap()).isFalse(); + assertThat(result.outOfOrder()).isFalse(); + } + + @Test + void rejectsNonPositiveTime() { + var result = calculator.calculateSegment( + point("2026-06-30T00:01:00Z", 120.000000, 30.000000, 36.0), + point("2026-06-30T00:01:00Z", 120.001000, 30.000000, 36.0)); + + assertThat(result.used()).isFalse(); + assertThat(result.outOfOrder()).isTrue(); + assertThat(result.gpsKm()).isZero(); + } + + @Test + void rejectsLongGap() { + var result = calculator.calculateSegment( + point("2026-06-30T00:00:00Z", 120.000000, 30.000000, 36.0), + point("2026-06-30T00:06:00Z", 120.001000, 30.000000, 36.0)); + + assertThat(result.used()).isFalse(); + assertThat(result.longGap()).isTrue(); + assertThat(result.gpsKm()).isZero(); + } + + @Test + void rejectsBadJumpByImpliedSpeed() { + var result = calculator.calculateSegment( + point("2026-06-30T00:00:00Z", 120.000000, 30.000000, 36.0), + point("2026-06-30T00:01:00Z", 120.100000, 30.000000, 36.0)); + + assertThat(result.used()).isFalse(); + assertThat(result.badJump()).isTrue(); + assertThat(result.gpsKm()).isZero(); + } + + @Test + void rejectsShortGapJumpByDistance() { + var result = calculator.calculateSegment( + point("2026-06-30T00:00:00Z", 120.000000, 30.000000, 36.0), + point("2026-06-30T00:00:05Z", 120.004000, 30.000000, 36.0)); + + assertThat(result.used()).isFalse(); + assertThat(result.badJump()).isTrue(); + assertThat(result.gpsKm()).isZero(); + } + + private static Jt808LocationPoint point(String time, double longitude, double latitude, Double speedKmh) { + return new Jt808LocationPoint( + "VIN001", + "VIN001", + "13900000001", + Instant.parse(time), + longitude, + latitude, + speedKmh, + 3L, + null); + } +} diff --git a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPointExtractorTest.java b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPointExtractorTest.java new file mode 100644 index 00000000..de3c5a77 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPointExtractorTest.java @@ -0,0 +1,84 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import com.lingniu.ingest.sink.mq.proto.TelemetryField; +import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot; +import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; +import org.junit.jupiter.api.Test; + +import java.time.Instant; + +import static org.assertj.core.api.Assertions.assertThat; + +class Jt808LocationPointExtractorTest { + + private final Jt808LocationPointExtractor extractor = new Jt808LocationPointExtractor(); + + @Test + void extractsJt808LocationPointWithVin() { + var point = extractor.extract(envelope("JT808", "VIN001", "13900000001", + field("longitude", "118.901034"), + field("latitude", "31.946325"), + field("speed_kmh", "52.3"), + field("location_status_raw", "3"), + field("total_mileage_km", "1234.5"))).orElseThrow(); + + assertThat(point.vehicleKey()).isEqualTo("VIN001"); + assertThat(point.vin()).isEqualTo("VIN001"); + assertThat(point.phone()).isEqualTo("13900000001"); + assertThat(point.eventTime()).isEqualTo(Instant.ofEpochMilli(1_782_112_400_000L)); + assertThat(point.longitude()).isEqualTo(118.901034); + assertThat(point.latitude()).isEqualTo(31.946325); + assertThat(point.speedKmh()).isEqualTo(52.3); + assertThat(point.statusFlag()).isEqualTo(3L); + assertThat(point.totalMileageKm()).isEqualTo(1234.5); + } + + @Test + void usesPhoneFallbackWhenVinIsUnknown() { + var point = extractor.extract(envelope("JT808", "unknown", "40692934289", + field("longitude", "118.901034"), + field("latitude", "31.946325"))).orElseThrow(); + + assertThat(point.vehicleKey()).isEqualTo("jt808:40692934289"); + assertThat(point.vin()).isEqualTo("unknown"); + assertThat(point.phone()).isEqualTo("40692934289"); + } + + @Test + void ignoresNonJt808Envelope() { + assertThat(extractor.extract(envelope("GB32960", "VIN001", "", + field("longitude", "118.901034"), + field("latitude", "31.946325")))).isEmpty(); + } + + @Test + void ignoresEnvelopeWithoutCoordinates() { + assertThat(extractor.extract(envelope("JT808", "VIN001", "13900000001", + field("longitude", "118.901034")))).isEmpty(); + } + + private static VehicleEnvelope envelope(String source, String vin, String phone, TelemetryField... fields) { + TelemetrySnapshot.Builder snapshot = TelemetrySnapshot.newBuilder().setEventType("LOCATION"); + for (TelemetryField field : fields) { + snapshot.addFields(field); + } + return VehicleEnvelope.newBuilder() + .setEventId("event-1") + .setVin(vin) + .setSource(source) + .putMetadata("phone", phone) + .setEventTimeMs(1_782_112_400_000L) + .setIngestTimeMs(1_782_112_401_000L) + .setTelemetrySnapshot(snapshot) + .build(); + } + + private static TelemetryField field(String key, String value) { + return TelemetryField.newBuilder() + .setKey(key) + .setValueType("DOUBLE") + .setValue(value) + .setQuality("GOOD") + .build(); + } +} diff --git a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessorTest.java b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessorTest.java new file mode 100644 index 00000000..8b5c5ae0 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessorTest.java @@ -0,0 +1,95 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import com.lingniu.ingest.sink.mq.proto.TelemetryField; +import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot; +import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.time.Instant; +import java.time.LocalDate; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +class Jt808MileageStreamProcessorTest { + + @Test + void updatesDailyStateAndUpsertsResultForJt808Points() { + InMemoryJt808MileageStateStore stateStore = new InMemoryJt808MileageStateStore(); + CapturingRepository repository = new CapturingRepository(); + Jt808MileageStreamProcessor processor = processor(stateStore, repository); + + processor.process(envelope("2026-06-30T00:00:00Z", 120.0, 30.0, 100.0)); + processor.process(envelope("2026-06-30T00:01:00Z", 120.01, 30.0, 101.0)); + + Jt808DailyMileageState state = stateStore.load("VIN001", LocalDate.of(2026, 6, 30)).orElseThrow(); + assertThat(state.acceptedPoints()).isEqualTo(2); + assertThat(state.gpsMileageKm()).isGreaterThan(0.9); + assertThat(repository.results).hasSize(2); + assertThat(repository.results.getLast().vehicleKey()).isEqualTo("VIN001"); + assertThat(repository.results.getLast().odometerMileageKm()).isEqualTo(1.0); + } + + @Test + void skipsNonJt808Envelope() { + CapturingRepository repository = new CapturingRepository(); + Jt808MileageStreamProcessor processor = processor(new InMemoryJt808MileageStateStore(), repository); + VehicleEnvelope envelope = envelope("2026-06-30T00:00:00Z", 120.0, 30.0, 100.0) + .toBuilder() + .setSource("GB32960") + .build(); + + processor.process(envelope); + + assertThat(repository.results).isEmpty(); + } + + private static Jt808MileageStreamProcessor processor( + Jt808MileageStateStore stateStore, + Jt808DailyMileageRepository repository) { + return new Jt808MileageStreamProcessor( + new Jt808LocationPointExtractor(), + new Jt808GpsMileageCalculator(Duration.ofMinutes(5), 200.0, Duration.ofSeconds(10), 300.0), + stateStore, + repository, + ZoneId.of("Asia/Shanghai")); + } + + private static VehicleEnvelope envelope(String eventTime, double longitude, double latitude, double totalMileageKm) { + long eventTimeMs = Instant.parse(eventTime).toEpochMilli(); + return VehicleEnvelope.newBuilder() + .setEventId("event-" + eventTimeMs) + .setVin("VIN001") + .setSource("JT808") + .setEventTimeMs(eventTimeMs) + .putMetadata("phone", "13900000001") + .setTelemetrySnapshot(TelemetrySnapshot.newBuilder() + .setEventType("LOCATION") + .addFields(field("longitude", longitude)) + .addFields(field("latitude", latitude)) + .addFields(field("speed_kmh", 36.0)) + .addFields(field("total_mileage_km", totalMileageKm))) + .build(); + } + + private static TelemetryField field(String key, double value) { + return TelemetryField.newBuilder() + .setKey(key) + .setValueType("DOUBLE") + .setValue(Double.toString(value)) + .setQuality("GOOD") + .build(); + } + + private static final class CapturingRepository implements Jt808DailyMileageRepository { + private final List results = new ArrayList<>(); + + @Override + public void upsert(Jt808DailyMileageResult result) { + results.add(result); + } + } +} diff --git a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStoreTest.java b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStoreTest.java new file mode 100644 index 00000000..a08cf2c8 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStoreTest.java @@ -0,0 +1,86 @@ +package com.lingniu.ingest.vehiclestat.jt808; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.ValueOperations; + +import java.time.Duration; +import java.time.Instant; +import java.time.LocalDate; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class RedisJt808MileageStateStoreTest { + + @Test + void savesStateAsJsonWithTtl() { + StringRedisTemplate redis = mock(StringRedisTemplate.class); + @SuppressWarnings("unchecked") + ValueOperations values = mock(ValueOperations.class); + when(redis.opsForValue()).thenReturn(values); + RedisJt808MileageStateStore store = new RedisJt808MileageStateStore( + redis, new ObjectMapper(), "vehicle:mileage:jt808:daily:", Duration.ofDays(3)); + Jt808DailyMileageState state = Jt808DailyMileageState.empty("VIN001", "VIN001", "13900000001", + LocalDate.of(2026, 6, 30)); + state.apply(point("2026-06-30T00:00:00Z"), calculator()); + + store.save(state); + + verify(values).set( + org.mockito.ArgumentMatchers.eq("vehicle:mileage:jt808:daily:2026-06-30:VIN001"), + org.mockito.ArgumentMatchers.contains("\"vehicleKey\":\"VIN001\""), + org.mockito.ArgumentMatchers.eq(Duration.ofDays(3))); + } + + @Test + void loadsStateFromJson() { + StringRedisTemplate redis = mock(StringRedisTemplate.class); + @SuppressWarnings("unchecked") + ValueOperations values = mock(ValueOperations.class); + when(redis.opsForValue()).thenReturn(values); + when(values.get("vehicle:mileage:jt808:daily:2026-06-30:VIN001")) + .thenReturn(""" + {"vehicleKey":"VIN001","vin":"VIN001","phone":"13900000001","statDate":"2026-06-30", + "firstEventTime":"2026-06-30T00:00:00Z","lastEventTime":"2026-06-30T00:01:00Z", + "lastPoint":{"vehicleKey":"VIN001","vin":"VIN001","phone":"13900000001", + "eventTime":"2026-06-30T00:01:00Z","longitude":120.0,"latitude":30.0, + "speedKmh":36.0,"statusFlag":3,"totalMileageKm":101.0}, + "lastTotalMileageKm":101.0,"gpsMileageKm":1.0,"speedIntegralKm":1.0, + "odometerMileageKm":1.0,"acceptedPoints":2,"badJumpSegments":0, + "longGapSegments":0,"outOfOrderPoints":0,"odometerAnomalies":0} + """); + RedisJt808MileageStateStore store = new RedisJt808MileageStateStore( + redis, new ObjectMapper(), "vehicle:mileage:jt808:daily:", Duration.ofDays(3)); + + Optional loaded = store.load("VIN001", LocalDate.of(2026, 6, 30)); + + assertThat(loaded).hasValueSatisfying(state -> { + assertThat(state.vehicleKey()).isEqualTo("VIN001"); + assertThat(state.acceptedPoints()).isEqualTo(2); + assertThat(state.gpsMileageKm()).isEqualTo(1.0); + assertThat(state.lastEventTime()).isEqualTo(Instant.parse("2026-06-30T00:01:00Z")); + }); + } + + private static Jt808GpsMileageCalculator calculator() { + return new Jt808GpsMileageCalculator(Duration.ofMinutes(5), 200.0, Duration.ofSeconds(10), 300.0); + } + + private static Jt808LocationPoint point(String time) { + return new Jt808LocationPoint( + "VIN001", + "VIN001", + "13900000001", + Instant.parse(time), + 120.0, + 30.0, + 36.0, + 3L, + 100.0); + } +}