feat: add jt808 streaming daily mileage

This commit is contained in:
lingniu
2026-06-30 17:02:54 +08:00
parent e47e341034
commit 3cc7ac9669
27 changed files with 2109 additions and 6 deletions

View File

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

View File

@@ -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://<host>:3306/<database>?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai
MYSQL_USERNAME=<user>
MYSQL_PASSWORD=<password>
REDIS_HOST=<host>
REDIS_PORT=6379
REDIS_DATABASE=50
REDIS_PASSWORD=<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.

View File

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

View File

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

View File

@@ -28,6 +28,18 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
</dependency>
<dependency>
<groupId>com.mysql</groupId>
<artifactId>mysql-connector-j</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
@@ -53,5 +65,10 @@
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

View File

@@ -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");

View File

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

View File

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

View File

@@ -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<String, Jt808DailyMileageState> states = new ConcurrentHashMap<>();
@Override
public Optional<Jt808DailyMileageState> 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;
}
}

View File

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

View File

@@ -0,0 +1,6 @@
package com.lingniu.ingest.vehiclestat.jt808;
public interface Jt808DailyMileageRepository {
void upsert(Jt808DailyMileageResult result);
}

View File

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

View File

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

View File

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

View File

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

View File

@@ -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<Jt808LocationPoint> extract(VehicleEnvelope envelope) {
if (envelope == null
|| !"JT808".equalsIgnoreCase(envelope.getSource())
|| !envelope.hasTelemetrySnapshot()) {
return Optional.empty();
}
Map<String, String> 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<String, String> fields(VehicleEnvelope envelope) {
Map<String, String> 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();
}
}

View File

@@ -0,0 +1,11 @@
package com.lingniu.ingest.vehiclestat.jt808;
import java.time.LocalDate;
import java.util.Optional;
public interface Jt808MileageStateStore {
Optional<Jt808DailyMileageState> load(String vehicleKey, LocalDate statDate);
void save(Jt808DailyMileageState state);
}

View File

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

View File

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

View File

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

View File

@@ -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<MileagePoint> mileagePoints(String vin, LocalDate statDate) { return List.of(); }

View File

@@ -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<String, Object> 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");
}
}

View File

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

View File

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

View File

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

View File

@@ -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<Jt808DailyMileageResult> results = new ArrayList<>();
@Override
public void upsert(Jt808DailyMileageResult result) {
results.add(result);
}
}
}

View File

@@ -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<String, String> 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<String, String> 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<Jt808DailyMileageState> 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);
}
}