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

refactor: simplify jt808 mileage statistics

Browse Source
This commit is contained in:
lingniu
2026-07-01 01:21:09 +08:00
parent 519f5591e9
commit 1ffa95c924
18 changed files with 201 additions and 1250 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java
View File

@@ -14,9 +14,6 @@ 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;
@@ -33,7 +30,6 @@ 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;
@@ -99,26 +95,6 @@ public class VehicleStatAutoConfiguration {
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")
@@ -140,15 +116,14 @@ public class VehicleStatAutoConfiguration {
}
@Bean
@ConditionalOnBean({Jt808LocationPointExtractor.class, Jt808GpsMileageCalculator.class,
Jt808MileageStateStore.class, Jt808DailyMileageRepository.class})
@ConditionalOnBean({Jt808LocationPointExtractor.class, Jt808MileageStateStore.class,
VehicleStatRepository.class})
@ConditionalOnMissingBean
public Jt808MileageStreamProcessor jt808MileageStreamProcessor(Jt808LocationPointExtractor extractor,
Jt808GpsMileageCalculator calculator,
Jt808MileageStateStore stateStore,
Jt808DailyMileageRepository repository,
VehicleStatRepository repository,
VehicleStatProperties props) {
return new Jt808MileageStreamProcessor(extractor, calculator, stateStore, repository,
return new Jt808MileageStreamProcessor(extractor, stateStore, repository,
ZoneId.of(props.getZoneId()));
}

35
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatProperties.java
View File

@@ -42,10 +42,6 @@ public class VehicleStatProperties {
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;
@@ -79,36 +75,5 @@ public class VehicleStatProperties {
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;
}
}
}

126
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepository.java
View File

@@ -1,126 +0,0 @@
package com.lingniu.ingest.vehiclestat.jt808;
import org.springframework.jdbc.core.JdbcTemplate;
import java.sql.Timestamp;
import java.time.Instant;
import java.util.Locale;
import java.util.Objects;
public final class JdbcJt808DailyMileageRepository implements Jt808DailyMileageRepository {
private final JdbcTemplate jdbc;
public JdbcJt808DailyMileageRepository(JdbcTemplate jdbc) {
this.jdbc = Objects.requireNonNull(jdbc, "jdbc must not be null");
initializeSchema();
}
@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,
daily_mileage_km, mileage_source, gps_mileage_km, speed_integral_km,
odometer_mileage_km, first_total_mileage_km, last_total_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),
daily_mileage_km = values(daily_mileage_km),
mileage_source = values(mileage_source),
gps_mileage_km = values(gps_mileage_km),
speed_integral_km = values(speed_integral_km),
odometer_mileage_km = values(odometer_mileage_km),
first_total_mileage_km = values(first_total_mileage_km),
last_total_mileage_km = values(last_total_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.dailyMileageKm(),
result.mileageSource(),
result.gpsMileageKm(),
result.speedIntegralKm(),
result.odometerMileageKm(),
result.firstTotalMileageKm(),
result.lastTotalMileageKm(),
result.acceptedPoints(),
result.badJumpSegments(),
result.longGapSegments(),
result.outOfOrderPoints(),
result.odometerAnomalies(),
result.dataQuality());
}
private void initializeSchema() {
jdbc.execute("""
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,
daily_mileage_km decimal(12,3) not null default 0,
mileage_source varchar(32) not null default 'INSUFFICIENT',
gps_mileage_km decimal(12,3) not null,
speed_integral_km decimal(12,3) not null,
odometer_mileage_km decimal(12,3) null,
first_total_mileage_km decimal(12,3) null,
last_total_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),
key idx_vehicle_daily_mileage_jt808_vin_date (vin, stat_date),
key idx_vehicle_daily_mileage_jt808_phone_date (phone, stat_date)
)
""");
ensureColumn("daily_mileage_km", "daily_mileage_km decimal(12,3) not null default 0");
ensureColumn("mileage_source", "mileage_source varchar(32) not null default 'INSUFFICIENT'");
ensureColumn("first_total_mileage_km", "first_total_mileage_km decimal(12,3) null");
ensureColumn("last_total_mileage_km", "last_total_mileage_km decimal(12,3) null");
}
private void ensureColumn(String columnName, String definition) {
if (!columnExists(columnName)) {
jdbc.execute("alter table vehicle_daily_mileage_jt808 add column " + definition);
}
}
private boolean columnExists(String columnName) {
Integer count = jdbc.queryForObject("""
select count(*)
from information_schema.columns
where lower(table_name) = ?
and lower(column_name) = ?
""", Integer.class,
"vehicle_daily_mileage_jt808",
columnName.toLowerCase(Locale.ROOT));
return count != null && count > 0;
}
private static Timestamp timestamp(Instant instant) {
return instant == null ? null : Timestamp.from(instant);
}
}

6
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageRepository.java
View File

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

26
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageResult.java
View File

@@ -1,26 +0,0 @@
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 dailyMileageKm,
String mileageSource,
double gpsMileageKm,
double speedIntegralKm,
Double odometerMileageKm,
Double firstTotalMileageKm,
Double lastTotalMileageKm,
int acceptedPoints,
int badJumpSegments,
int longGapSegments,
int outOfOrderPoints,
int odometerAnomalies,
String dataQuality
) {}

172
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageState.java
View File

@@ -1,7 +1,12 @@
package com.lingniu.ingest.vehiclestat.jt808;
import com.lingniu.ingest.vehiclestat.DailyMileageStrategy;
import com.lingniu.ingest.vehiclestat.VehicleDailyStatResult;
import java.time.Instant;
import java.time.LocalDate;
import java.util.Optional;
import java.util.OptionalDouble;
public final class Jt808DailyMileageState {
@@ -11,15 +16,11 @@ public final class Jt808DailyMileageState {
private final LocalDate statDate;
private Instant firstEventTime;
private Instant lastEventTime;
private Jt808LocationPoint lastPoint;
private double firstTotalMileageKm = Double.NaN;
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 totalMileageSamples;
private int outOfOrderPoints;
private int odometerAnomalies;
@@ -47,100 +48,60 @@ public final class Jt808DailyMileageState {
LocalDate statDate,
Instant firstEventTime,
Instant lastEventTime,
Jt808LocationPoint lastPoint,
double firstTotalMileageKm,
double lastTotalMileageKm,
double gpsMileageKm,
double speedIntegralKm,
double odometerMileageKm,
int acceptedPoints,
int badJumpSegments,
int longGapSegments,
int totalMileageSamples,
int outOfOrderPoints,
int odometerAnomalies) {
Jt808DailyMileageState state = new Jt808DailyMileageState(vehicleKey, vin, phone, statDate);
state.firstEventTime = firstEventTime;
state.lastEventTime = lastEventTime;
state.lastPoint = lastPoint;
state.firstTotalMileageKm = firstTotalMileageKm;
state.lastTotalMileageKm = lastTotalMileageKm;
state.gpsMileageKm = gpsMileageKm;
state.speedIntegralKm = speedIntegralKm;
state.odometerMileageKm = odometerMileageKm;
state.acceptedPoints = acceptedPoints;
state.badJumpSegments = badJumpSegments;
state.longGapSegments = longGapSegments;
state.totalMileageSamples = totalMileageSamples;
state.outOfOrderPoints = outOfOrderPoints;
state.odometerAnomalies = odometerAnomalies;
return state;
}
public void apply(Jt808LocationPoint point, Jt808GpsMileageCalculator calculator) {
if (point == null || calculator == null) {
public void apply(Jt808LocationPoint point) {
if (point == null) {
return;
}
if (lastPoint != null && !point.eventTime().isAfter(lastPoint.eventTime())) {
if (lastEventTime != null && !point.eventTime().isAfter(lastEventTime)) {
outOfOrderPoints++;
return;
}
if (lastPoint == null) {
if (firstEventTime == null) {
firstEventTime = point.eventTime();
updateOdometer(point, calculator);
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, calculator);
accept(point);
}
public Jt808DailyMileageResult toResult() {
return new Jt808DailyMileageResult(
statDate,
vehicleKey,
vin,
phone,
firstEventTime,
lastEventTime,
dailyMileageKm(),
mileageSource(),
gpsMileageKm,
speedIntegralKm,
odometerMileageKm > 0.0 ? odometerMileageKm : null,
nullable(firstTotalMileageKm),
nullable(lastTotalMileageKm),
acceptedPoints,
badJumpSegments,
longGapSegments,
outOfOrderPoints,
odometerAnomalies,
dataQuality());
}
private void accept(Jt808LocationPoint point) {
lastPoint = point;
updateOdometer(point);
lastEventTime = point.eventTime();
acceptedPoints++;
}
private void updateOdometer(Jt808LocationPoint point, Jt808GpsMileageCalculator calculator) {
public Optional<VehicleDailyStatResult> toVehicleDailyStatResult() {
OptionalDouble dailyMileage = totalMileageDifference();
if (dailyMileage.isEmpty()) {
return Optional.empty();
}
return Optional.of(new VehicleDailyStatResult(
statVehicleId(),
statDate,
dailyMileage,
DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST));
}
private void updateOdometer(Jt808LocationPoint point) {
if (point.totalMileageKm() == null || !Double.isFinite(point.totalMileageKm())) {
return;
}
double current = point.totalMileageKm();
totalMileageSamples++;
if (!Double.isFinite(firstTotalMileageKm)) {
firstTotalMileageKm = current;
}
@@ -151,61 +112,28 @@ public final class Jt808DailyMileageState {
if (current < lastTotalMileageKm) {
odometerAnomalies++;
lastTotalMileageKm = current;
odometerMileageKm = 0.0;
return;
}
double delta = current - lastTotalMileageKm;
if (lastPoint != null && !calculator.plausibleOdometerDelta(lastPoint, point, delta)) {
odometerAnomalies++;
lastTotalMileageKm = current;
return;
}
odometerMileageKm += delta;
lastTotalMileageKm = current;
odometerMileageKm = current - firstTotalMileageKm;
}
private double dailyMileageKm() {
if (hasUsableTotalMileage()) {
return odometerMileageKm;
private OptionalDouble totalMileageDifference() {
if (totalMileageSamples < 2
|| !Double.isFinite(firstTotalMileageKm)
|| !Double.isFinite(lastTotalMileageKm)
|| lastTotalMileageKm < firstTotalMileageKm) {
return OptionalDouble.empty();
}
if (gpsMileageKm > 0.0) {
return gpsMileageKm;
}
return 0.0;
return OptionalDouble.of(lastTotalMileageKm - firstTotalMileageKm);
}
private String mileageSource() {
if (hasUsableTotalMileage()) {
return odometerAnomalies > 0 ? "JT808_TOTAL_MILEAGE_PARTIAL" : "JT808_TOTAL_MILEAGE";
private String statVehicleId() {
if (!vin.isBlank() && !"unknown".equalsIgnoreCase(vin)) {
return vin;
}
if (gpsMileageKm > 0.0) {
return "GPS_DISTANCE";
}
return "INSUFFICIENT";
}
private boolean hasUsableTotalMileage() {
if (odometerMileageKm > 0.0) {
return true;
}
return acceptedPoints >= 2
&& odometerAnomalies == 0
&& Double.isFinite(firstTotalMileageKm)
&& Double.isFinite(lastTotalMileageKm)
&& lastTotalMileageKm >= firstTotalMileageKm;
}
private static Double nullable(double value) {
return Double.isFinite(value) ? value : null;
}
private String dataQuality() {
if (acceptedPoints < 2) {
return "PARTIAL";
}
if (badJumpSegments > 0 || longGapSegments > 0 || odometerAnomalies > 0 || vehicleKey.startsWith("jt808:")) {
return "PARTIAL";
}
return "GOOD";
return vehicleKey;
}
public String vehicleKey() {
@@ -232,18 +160,6 @@ public final class Jt808DailyMileageState {
return lastEventTime;
}
public Jt808LocationPoint lastPoint() {
return lastPoint;
}
public double gpsMileageKm() {
return gpsMileageKm;
}
public double speedIntegralKm() {
return speedIntegralKm;
}
public double odometerMileageKm() {
return odometerMileageKm;
}
@@ -260,12 +176,8 @@ public final class Jt808DailyMileageState {
return acceptedPoints;
}
public int badJumpSegments() {
return badJumpSegments;
}
public int longGapSegments() {
return longGapSegments;
public int totalMileageSamples() {
return totalMileageSamples;
}
public int outOfOrderPoints() {

108
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculator.java
View File

@@ -1,108 +0,0 @@
package com.lingniu.ingest.vehiclestat.jt808;
import java.time.Duration;
public final class Jt808GpsMileageCalculator {
private static final double EARTH_RADIUS_KM = 6371.0088;
private static final double ODOMETER_DELTA_TOLERANCE_KM = 0.1;
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);
}
public boolean plausibleOdometerDelta(Jt808LocationPoint previous,
Jt808LocationPoint current,
double deltaKm) {
if (previous == null || current == null || !Double.isFinite(deltaKm) || deltaKm < 0.0) {
return false;
}
Duration elapsed = Duration.between(previous.eventTime(), current.eventTime());
if (elapsed.isZero() || elapsed.isNegative()) {
return false;
}
double hours = elapsed.toMillis() / 3_600_000.0;
return deltaKm <= maxImpliedSpeedKmh * hours + ODOMETER_DELTA_TOLERANCE_KM;
}
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);
}
}
}

12
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessor.java
View File

@@ -1,6 +1,7 @@
package com.lingniu.ingest.vehiclestat.jt808;
import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
import com.lingniu.ingest.vehiclestat.VehicleStatRepository;
import java.time.LocalDate;
import java.time.ZoneId;
@@ -8,19 +9,16 @@ 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 VehicleStatRepository repository;
private final ZoneId zoneId;
public Jt808MileageStreamProcessor(
Jt808LocationPointExtractor extractor,
Jt808GpsMileageCalculator calculator,
Jt808MileageStateStore stateStore,
Jt808DailyMileageRepository repository,
VehicleStatRepository repository,
ZoneId zoneId) {
this.extractor = extractor;
this.calculator = calculator;
this.stateStore = stateStore;
this.repository = repository;
this.zoneId = zoneId;
@@ -34,8 +32,8 @@ public final class Jt808MileageStreamProcessor {
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);
state.apply(point);
stateStore.save(state);
repository.upsert(state.toResult());
state.toVehicleDailyStatResult().ifPresent(repository::saveDailyStat);
}
}

66
modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStore.java
View File

@@ -66,12 +66,9 @@ public final class RedisJt808MileageStateStore implements Jt808MileageStateStore
PointSnapshot lastPoint,
Double firstTotalMileageKm,
Double lastTotalMileageKm,
double gpsMileageKm,
double speedIntegralKm,
double odometerMileageKm,
int acceptedPoints,
int badJumpSegments,
int longGapSegments,
Integer totalMileageSamples,
int outOfOrderPoints,
int odometerAnomalies) {
@@ -83,15 +80,12 @@ public final class RedisJt808MileageStateStore implements Jt808MileageStateStore
state.statDate().toString(),
format(state.firstEventTime()),
format(state.lastEventTime()),
PointSnapshot.from(state.lastPoint()),
null,
finiteOrNull(state.firstTotalMileageKm()),
finiteOrNull(state.lastTotalMileageKm()),
state.gpsMileageKm(),
state.speedIntegralKm(),
state.odometerMileageKm(),
state.acceptedPoints(),
state.badJumpSegments(),
state.longGapSegments(),
state.totalMileageSamples(),
state.outOfOrderPoints(),
state.odometerAnomalies());
}
@@ -103,19 +97,33 @@ public final class RedisJt808MileageStateStore implements Jt808MileageStateStore
phone,
LocalDate.parse(statDate),
parseInstant(firstEventTime),
parseInstant(lastEventTime),
lastPoint == null ? null : lastPoint.toPoint(),
restoredLastEventTime(),
nanIfNull(firstTotalMileageKm),
nanIfNull(lastTotalMileageKm),
gpsMileageKm,
speedIntegralKm,
odometerMileageKm,
acceptedPoints,
badJumpSegments,
longGapSegments,
restoredTotalMileageSamples(),
outOfOrderPoints,
odometerAnomalies);
}
private Instant restoredLastEventTime() {
Instant parsedLastEventTime = parseInstant(lastEventTime);
if (parsedLastEventTime != null) {
return parsedLastEventTime;
}
return lastPoint == null ? null : parseInstant(lastPoint.eventTime());
}
private int restoredTotalMileageSamples() {
if (totalMileageSamples != null) {
return totalMileageSamples;
}
if (firstTotalMileageKm != null && lastTotalMileageKm != null) {
return acceptedPoints >= 2 || !lastTotalMileageKm.equals(firstTotalMileageKm) ? 2 : 1;
}
return firstTotalMileageKm != null || lastTotalMileageKm != null ? 1 : 0;
}
}
@JsonIgnoreProperties(ignoreUnknown = true)
@@ -130,34 +138,6 @@ public final class RedisJt808MileageStateStore implements Jt808MileageStateStore
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) {

6
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfigurationTest.java
View File

@@ -14,7 +14,6 @@ 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;
@@ -86,7 +85,7 @@ class VehicleStatAutoConfigurationTest {
}
@Test
void createsJt808MileageBeansWhenEnabledWithJdbcAndRedis() {
void createsJt808MileageBeansWhenEnabledWithRedis() {
contextRunner
.withBean(JdbcTemplate.class, () -> mock(JdbcTemplate.class))
.withBean(StringRedisTemplate.class, () -> mock(StringRedisTemplate.class))
@@ -96,7 +95,6 @@ class VehicleStatAutoConfigurationTest {
"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);
@@ -113,7 +111,6 @@ class VehicleStatAutoConfigurationTest {
"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);
@@ -137,7 +134,6 @@ class VehicleStatAutoConfigurationTest {
"lingniu.ingest.vehicle-stat.jt808.state-store=memory")
.run(context -> {
assertThat(context).hasSingleBean(JdbcTemplate.class);
assertThat(context).hasSingleBean(Jt808DailyMileageRepository.class);
assertThat(context).hasSingleBean(Jt808MileageStreamProcessor.class);
assertThat(context).hasSingleBean(VehicleStatEnvelopeIngestor.class);
});

105
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepositoryTest.java
View File

@@ -1,105 +0,0 @@
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");
repository = new JdbcJt808DailyMileageRepository(jdbc);
}
@Test
void initializesTableWhenMissing() {
Integer count = jdbc.queryForObject("""
select count(*)
from information_schema.tables
where table_name = 'vehicle_daily_mileage_jt808'
""", Integer.class);
assertThat(count).isEqualTo(1);
}
@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,
"GPS_DISTANCE",
1.23456,
1.2,
null,
null,
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.0,
"JT808_TOTAL_MILEAGE",
2.34567,
2.3,
2.0,
100.0,
102.0,
3,
1,
0,
0,
0,
"PARTIAL");
repository.upsert(first);
repository.upsert(updated);
Map<String, Object> row = jdbc.queryForMap("""
select vehicle_key, daily_mileage_km, mileage_source, gps_mileage_km,
odometer_mileage_km, first_total_mileage_km, last_total_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("daily_mileage_km").toString()).isEqualTo("2.000");
assertThat(row.get("mileage_source")).isEqualTo("JT808_TOTAL_MILEAGE");
assertThat(row.get("gps_mileage_km").toString()).isEqualTo("2.346");
assertThat(row.get("odometer_mileage_km").toString()).isEqualTo("2.000");
assertThat(row.get("first_total_mileage_km").toString()).isEqualTo("100.000");
assertThat(row.get("last_total_mileage_km").toString()).isEqualTo("102.000");
assertThat(row.get("accepted_points")).isEqualTo(3);
assertThat(row.get("bad_jump_segments")).isEqualTo(1);
assertThat(row.get("data_quality")).isEqualTo("PARTIAL");
}
}

79
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageStateTest.java
View File

@@ -2,7 +2,6 @@ package com.lingniu.ingest.vehiclestat.jt808;
import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDate;
@@ -11,24 +10,19 @@ 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);
state.apply(point("2026-06-30T00:00:00Z", 120.000000, 30.000000, 36.0, 100.0));
assertThat(state.acceptedPoints()).isEqualTo(1);
assertThat(state.gpsMileageKm()).isZero();
assertThat(state.speedIntegralKm()).isZero();
assertThat(state.totalMileageSamples()).isEqualTo(1);
assertThat(state.odometerMileageKm()).isZero();
assertThat(state.firstTotalMileageKm()).isEqualTo(100.0);
assertThat(state.lastTotalMileageKm()).isEqualTo(100.0);
assertThat(state.toResult().dailyMileageKm()).isZero();
assertThat(state.toResult().mileageSource()).isEqualTo("INSUFFICIENT");
assertThat(state.toVehicleDailyStatResult()).isEmpty();
assertThat(state.firstEventTime()).isEqualTo(Instant.parse("2026-06-30T00:00:00Z"));
assertThat(state.lastEventTime()).isEqualTo(Instant.parse("2026-06-30T00:00:00Z"));
}
@@ -37,77 +31,84 @@ class Jt808DailyMileageStateTest {
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:00:00Z", 120.000000, 30.000000, 36.0, 100.0));
state.apply(point("2026-06-30T00:01:40Z", 120.010370, 30.000000, 36.0, 101.2), calculator);
state.apply(point("2026-06-30T00:01:40Z", 120.010370, 30.000000, 36.0, 101.2));
assertThat(state.acceptedPoints()).isEqualTo(2);
assertThat(state.gpsMileageKm()).isBetween(0.99, 1.01);
assertThat(state.speedIntegralKm()).isBetween(0.99, 1.01);
assertThat(state.totalMileageSamples()).isEqualTo(2);
assertThat(state.odometerMileageKm()).isCloseTo(1.2, offset(0.000001));
assertThat(state.firstTotalMileageKm()).isEqualTo(100.0);
assertThat(state.lastTotalMileageKm()).isEqualTo(101.2);
assertThat(state.toResult().dailyMileageKm()).isCloseTo(1.2, offset(0.000001));
assertThat(state.toResult().mileageSource()).isEqualTo("JT808_TOTAL_MILEAGE");
assertThat(state.toResult().firstTotalMileageKm()).isEqualTo(100.0);
assertThat(state.toResult().lastTotalMileageKm()).isEqualTo(101.2);
assertThat(state.badJumpSegments()).isZero();
assertThat(state.longGapSegments()).isZero();
assertThat(state.toVehicleDailyStatResult()).isPresent();
assertThat(state.toVehicleDailyStatResult().orElseThrow().dailyMileageKm().getAsDouble())
.isCloseTo(1.2, offset(0.000001));
}
@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:01:00Z", 120.000000, 30.000000, 36.0, 100.0));
state.apply(point("2026-06-30T00:00:59Z", 120.010370, 30.000000, 36.0, 101.0), calculator);
state.apply(point("2026-06-30T00:00:59Z", 120.010370, 30.000000, 36.0, 101.0));
assertThat(state.acceptedPoints()).isEqualTo(1);
assertThat(state.outOfOrderPoints()).isEqualTo(1);
assertThat(state.gpsMileageKm()).isZero();
assertThat(state.totalMileageSamples()).isEqualTo(1);
}
@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:00:00Z", 120.000000, 30.000000, 36.0, 100.0));
state.apply(point("2026-06-30T00:01:40Z", 120.010370, 30.000000, 36.0, 99.5), calculator);
state.apply(point("2026-06-30T00:01:40Z", 120.010370, 30.000000, 36.0, 99.5));
assertThat(state.odometerMileageKm()).isZero();
assertThat(state.odometerAnomalies()).isEqualTo(1);
assertThat(state.toResult().dailyMileageKm()).isGreaterThan(0.9);
assertThat(state.toResult().mileageSource()).isEqualTo("GPS_DISTANCE");
assertThat(state.toVehicleDailyStatResult()).isEmpty();
}
@Test
void odometerJumpAboveConfiguredSpeedIsIgnoredAndCounted() {
void odometerJumpAboveConfiguredSpeedUsesSimpleDifference() {
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:00:00Z", 120.000000, 30.000000, 36.0, 100.0));
state.apply(point("2026-06-30T00:01:00Z", 120.000100, 30.000000, 36.0, 200.0), calculator);
state.apply(point("2026-06-30T00:01:00Z", 120.000100, 30.000000, 36.0, 200.0));
assertThat(state.odometerMileageKm()).isZero();
assertThat(state.odometerAnomalies()).isEqualTo(1);
assertThat(state.toResult().dataQuality()).isEqualTo("PARTIAL");
assertThat(state.odometerMileageKm()).isEqualTo(100.0);
assertThat(state.odometerAnomalies()).isZero();
assertThat(state.toVehicleDailyStatResult().orElseThrow().dailyMileageKm()).hasValue(100.0);
}
@Test
void gpsDistanceIsFallbackWhenTotalMileageIsMissing() {
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, null), calculator);
state.apply(point("2026-06-30T00:00:00Z", 120.000000, 30.000000, 36.0, null));
state.apply(point("2026-06-30T00:01:40Z", 120.010370, 30.000000, 36.0, null), calculator);
state.apply(point("2026-06-30T00:01:40Z", 120.010370, 30.000000, 36.0, null));
assertThat(state.toResult().dailyMileageKm()).isBetween(0.99, 1.01);
assertThat(state.toResult().mileageSource()).isEqualTo("GPS_DISTANCE");
assertThat(state.toResult().odometerMileageKm()).isNull();
assertThat(state.toResult().firstTotalMileageKm()).isNull();
assertThat(state.toResult().lastTotalMileageKm()).isNull();
assertThat(state.totalMileageSamples()).isZero();
assertThat(state.toVehicleDailyStatResult()).isEmpty();
}
@Test
void requiresTwoTotalMileageSamplesBeforeSavingMetric() {
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, null));
state.apply(point("2026-06-30T00:01:00Z", 120.000100, 30.000000, 36.0, 100.0));
assertThat(state.toVehicleDailyStatResult()).isEmpty();
state.apply(point("2026-06-30T00:02:00Z", 120.000200, 30.000000, 36.0, 101.0));
assertThat(state.toVehicleDailyStatResult().orElseThrow().dailyMileageKm()).hasValue(1.0);
}
private static Jt808LocationPoint point(String time,

98
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculatorTest.java
View File

@@ -1,98 +0,0 @@
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();
}
@Test
void validatesPlausibleOdometerDeltaByConfiguredSpeed() {
assertThat(calculator.plausibleOdometerDelta(
point("2026-06-30T00:00:00Z", 120.000000, 30.000000, 36.0),
point("2026-06-30T00:01:00Z", 120.001000, 30.000000, 36.0),
2.0)).isTrue();
assertThat(calculator.plausibleOdometerDelta(
point("2026-06-30T00:00:00Z", 120.000000, 30.000000, 36.0),
point("2026-06-30T00:01:00Z", 120.001000, 30.000000, 36.0),
20.0)).isFalse();
}
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);
}
}

57
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessorTest.java
View File

@@ -3,21 +3,25 @@ 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 com.lingniu.ingest.vehiclestat.DailyMileageStrategy;
import com.lingniu.ingest.vehiclestat.MileagePoint;
import com.lingniu.ingest.vehiclestat.VehicleDailyStatResult;
import com.lingniu.ingest.vehiclestat.VehicleStatRepository;
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 java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
class Jt808MileageStreamProcessorTest {
@Test
void updatesDailyStateAndUpsertsResultForJt808Points() {
void updatesDailyStateAndSavesResultToVehicleStatRepository() {
InMemoryJt808MileageStateStore stateStore = new InMemoryJt808MileageStateStore();
CapturingRepository repository = new CapturingRepository();
Jt808MileageStreamProcessor processor = processor(stateStore, repository);
@@ -27,12 +31,26 @@ class Jt808MileageStreamProcessorTest {
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().dailyMileageKm()).isEqualTo(1.0);
assertThat(repository.results.getLast().mileageSource()).isEqualTo("JT808_TOTAL_MILEAGE");
assertThat(repository.results.getLast().odometerMileageKm()).isEqualTo(1.0);
assertThat(state.totalMileageSamples()).isEqualTo(2);
assertThat(repository.results).hasSize(1);
VehicleDailyStatResult latest = repository.results.getLast();
assertThat(latest.vin()).isEqualTo("VIN001");
assertThat(latest.statDate()).isEqualTo(LocalDate.of(2026, 6, 30));
assertThat(latest.dailyMileageKm()).hasValue(1.0);
assertThat(latest.dailyMileageStrategy()).isEqualTo(DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST);
}
@Test
void usesSimpleTotalMileageDifferenceInsteadOfAccumulatedPlausibilityDeltas() {
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.0001, 30.0, 200.0));
VehicleDailyStatResult latest = repository.results.getLast();
assertThat(latest.dailyMileageKm()).hasValue(100.0);
}
@Test
@@ -51,10 +69,9 @@ class Jt808MileageStreamProcessorTest {
private static Jt808MileageStreamProcessor processor(
Jt808MileageStateStore stateStore,
Jt808DailyMileageRepository repository) {
VehicleStatRepository repository) {
return new Jt808MileageStreamProcessor(
new Jt808LocationPointExtractor(),
new Jt808GpsMileageCalculator(Duration.ofMinutes(5), 200.0, Duration.ofSeconds(10), 300.0),
stateStore,
repository,
ZoneId.of("Asia/Shanghai"));
@@ -86,12 +103,26 @@ class Jt808MileageStreamProcessorTest {
.build();
}
private static final class CapturingRepository implements Jt808DailyMileageRepository {
private final List<Jt808DailyMileageResult> results = new ArrayList<>();
private static final class CapturingRepository implements VehicleStatRepository {
private final List<VehicleDailyStatResult> results = new ArrayList<>();
@Override
public void upsert(Jt808DailyMileageResult result) {
public void appendMileagePoint(String vin, MileagePoint point) {
}
@Override
public List<MileagePoint> mileagePoints(String vin, LocalDate statDate) {
return List.of();
}
@Override
public void saveDailyStat(VehicleDailyStatResult result) {
results.add(result);
}
@Override
public Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate) {
return Optional.empty();
}
}
}

10
modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStoreTest.java
View File

@@ -27,7 +27,7 @@ class RedisJt808MileageStateStoreTest {
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());
state.apply(point("2026-06-30T00:00:00Z"));
store.save(state);
@@ -47,6 +47,7 @@ class RedisJt808MileageStateStoreTest {
.thenReturn("""
{"vehicleKey":"VIN001","vin":"VIN001","phone":"13900000001","statDate":"2026-06-30",
"firstEventTime":"2026-06-30T00:00:00Z","lastEventTime":"2026-06-30T00:01:00Z",
"firstTotalMileageKm":100.0,
"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},
@@ -62,15 +63,12 @@ class RedisJt808MileageStateStoreTest {
assertThat(loaded).hasValueSatisfying(state -> {
assertThat(state.vehicleKey()).isEqualTo("VIN001");
assertThat(state.acceptedPoints()).isEqualTo(2);
assertThat(state.gpsMileageKm()).isEqualTo(1.0);
assertThat(state.totalMileageSamples()).isEqualTo(2);
assertThat(state.toVehicleDailyStatResult().orElseThrow().dailyMileageKm()).hasValue(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",