Add tachograph speeding extractor

README.md

@@ -828,6 +828,7 @@ CARD_POSITION      -> POSITION      / DRIVER_CARD / CardGnssAccumulatedDriving
 VU_POSITION        -> POSITION      / VEHICLE_UNIT / VUGnssAccumulatedDriving
 CARD_PLACE         -> PLACE         / DRIVER_CARD / CardPlaces
 VU_PLACE           -> PLACE         / VEHICLE_UNIT / VUPlaces
+SPEEDING_EVENTS    -> SPEEDING     / VEHICLE_UNIT / SpeedingEvents
 ```
 
 SQL resources:
@@ -841,6 +842,7 @@ src/main/resources/sql/tachograph/card-specific-condition.sql
 src/main/resources/sql/tachograph/card-vehicles-used.sql
 src/main/resources/sql/tachograph/card-activity.sql
 src/main/resources/sql/tachograph/iw-cycle.sql
+src/main/resources/sql/tachograph/speeding-events.sql
 src/main/resources/sql/tachograph/vu-border-crossing.sql
 src/main/resources/sql/tachograph/vu-load-unload.sql
 src/main/resources/sql/tachograph/vu-place.sql

src/main/java/at/procon/eventhub/service/EventDetailsFactory.java

@@ -75,11 +75,16 @@ public class EventDetailsFactory {
     }
 
     public EventDetailsDto speeding(BigDecimal speedKmh, BigDecimal permittedSpeedKmh) {
+        return speeding(null, speedKmh, permittedSpeedKmh);
+    }
+
+    public EventDetailsDto speeding(BigDecimal avgSpeedKmh, BigDecimal maxSpeedKmh, BigDecimal permittedSpeedKmh) {
         Map<String, Object> attributes = new LinkedHashMap<>();
-        put(attributes, "speedKmh", speedKmh);
+        put(attributes, "avgSpeedKmh", avgSpeedKmh);
+        put(attributes, "maxSpeedKmh", maxSpeedKmh);
         put(attributes, "permittedSpeedKmh", permittedSpeedKmh);
-        if (speedKmh != null && permittedSpeedKmh != null) {
+        if (maxSpeedKmh != null && permittedSpeedKmh != null) {
-            put(attributes, "overspeedKmh", speedKmh.subtract(permittedSpeedKmh));
+            put(attributes, "overspeedKmh", maxSpeedKmh.subtract(permittedSpeedKmh));
         }
         return new EventDetailsDto("SPEEDING", objectMapper.valueToTree(attributes));
     }

src/main/java/at/procon/eventhub/tachograph/service/SpeedingEventRowMapper.java

@@ -0,0 +1,200 @@
+package at.procon.eventhub.tachograph.service;
+
+import at.procon.eventhub.dto.DriverCardRefDto;
+import at.procon.eventhub.dto.DriverRefDto;
+import at.procon.eventhub.dto.EventDomain;
+import at.procon.eventhub.dto.EventHubEventDto;
+import at.procon.eventhub.dto.EventLifecycle;
+import at.procon.eventhub.dto.EventType;
+import at.procon.eventhub.dto.SourcePackageRefDto;
+import at.procon.eventhub.dto.VehicleRefDto;
+import at.procon.eventhub.dto.VehicleRegistrationRefDto;
+import at.procon.eventhub.importing.extraction.ExtractionContext;
+import at.procon.eventhub.importing.extraction.ExtractionRowMapper;
+import at.procon.eventhub.service.EventDetailsFactory;
+import at.procon.eventhub.tachograph.dto.TachographImportRequest;
+import java.math.BigDecimal;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.UUID;
+import org.springframework.stereotype.Component;
+
+@Component
+public class SpeedingEventRowMapper implements ExtractionRowMapper<TachographImportRequest> {
+
+    private final EventDetailsFactory detailsFactory;
+
+    public SpeedingEventRowMapper(EventDetailsFactory detailsFactory) {
+        this.detailsFactory = detailsFactory;
+    }
+
+    @Override
+    public EventHubEventDto map(ResultSet rs, int rowNum, ExtractionContext<TachographImportRequest> context) throws SQLException {
+        OffsetDateTime occurredAt = offsetDateTime(rs, "occurred_at");
+        SourcePackageRefDto sourcePackageRef = sourcePackageRef(rs);
+        DriverRefDto driverRef = driverRef(rs);
+        VehicleRefDto vehicleRef = vehicleRef(rs);
+        EventLifecycle lifecycle = lifecycle(rs);
+
+        String externalSourceEventId = string(rs, "external_source_event_id");
+        if (externalSourceEventId == null) {
+            externalSourceEventId = defaultExternalSourceEventId(context, rowNum, occurredAt, sourcePackageRef, driverRef, vehicleRef, lifecycle);
+        }
+
+        return new EventHubEventDto(
+                UUID.randomUUID(),
+                externalSourceEventId,
+                driverRef,
+                vehicleRef,
+                occurredAt,
+                offsetDateTime(rs, "received_partner_at"),
+                OffsetDateTime.now(),
+                EventDomain.SPEEDING,
+                EventType.SPEEDING,
+                lifecycle,
+                null,
+                null,
+                detailsFactory.speeding(decimal(rs, "avg_speed_kmh"), decimal(rs, "max_speed_kmh"), null),
+                sourcePackageRef != null && sourcePackageRef.hasAnyReference() ? sourcePackageRef : null,
+                detailsFactory.payloadFromMap(payload(rs)),
+                false,
+                context.packageInfo()
+        );
+    }
+
+    private DriverRefDto driverRef(ResultSet rs) throws SQLException {
+        DriverCardRefDto driverCard = null;
+        String cardNumber = string(rs, "driver_card_number");
+        if (cardNumber != null) {
+            driverCard = new DriverCardRefDto(string(rs, "driver_card_nation"), cardNumber);
+        }
+        DriverRefDto driverRef = new DriverRefDto(string(rs, "driver_source_entity_id"), driverCard);
+        return driverRef.hasAnyReference() ? driverRef : null;
+    }
+
+    private VehicleRefDto vehicleRef(ResultSet rs) throws SQLException {
+        VehicleRegistrationRefDto registration = null;
+        String registrationNumber = string(rs, "vehicle_registration_number");
+        if (registrationNumber != null) {
+            registration = new VehicleRegistrationRefDto(string(rs, "vehicle_registration_nation"), registrationNumber);
+        }
+        VehicleRefDto vehicleRef = new VehicleRefDto(
+                string(rs, "vehicle_source_entity_id"),
+                string(rs, "vehicle_vin"),
+                string(rs, "vehicle_registration_source_entity_id"),
+                registration
+        );
+        return vehicleRef.hasAnyReference() ? vehicleRef : null;
+    }
+
+    private SourcePackageRefDto sourcePackageRef(ResultSet rs) throws SQLException {
+        return new SourcePackageRefDto(
+                string(rs, "source_package_kind"),
+                string(rs, "source_package_id"),
+                string(rs, "source_package_entity_id"),
+                offsetDateTime(rs, "source_package_period_from"),
+                offsetDateTime(rs, "source_package_period_to"),
+                offsetDateTime(rs, "source_package_imported_at")
+        );
+    }
+
+    private Map<String, Object> payload(ResultSet rs) throws SQLException {
+        Map<String, Object> raw = new LinkedHashMap<>();
+        put(raw, "sourceRowId", string(rs, "source_row_id"));
+        return Map.of("raw", raw);
+    }
+
+    private String defaultExternalSourceEventId(
+            ExtractionContext<TachographImportRequest> context,
+            int rowNum,
+            OffsetDateTime occurredAt,
+            SourcePackageRefDto sourcePackageRef,
+            DriverRefDto driverRef,
+            VehicleRefDto vehicleRef,
+            EventLifecycle lifecycle
+    ) {
+        String sourcePackageId = sourcePackageRef == null || sourcePackageRef.sourcePackageId() == null
+                ? "NO_SOURCE_PACKAGE"
+                : sourcePackageRef.sourcePackageId();
+        String subject = driverRef != null && driverRef.hasAnyReference()
+                ? driverRef.stableKey()
+                : vehicleRef == null ? "NO_SUBJECT" : vehicleRef.stableKey();
+        return "TACHOGRAPH:" + context.planItem().extractionCode()
+                + ":" + sourcePackageId
+                + ":SPEEDING"
+                + ":" + lifecycle
+                + ":" + occurredAt
+                + ":" + subject
+                + ":ROW-" + rowNum;
+    }
+
+    private String string(ResultSet rs, String column) throws SQLException {
+        String value = rs.getString(column);
+        return value == null || value.isBlank() ? null : value.trim();
+    }
+
+    private OffsetDateTime offsetDateTime(ResultSet rs, String column) throws SQLException {
+        Object value = rs.getObject(column);
+        if (value == null) {
+            return null;
+        }
+        if (value instanceof OffsetDateTime offsetDateTime) {
+            return offsetDateTime.withOffsetSameInstant(ZoneOffset.UTC);
+        }
+        if (value instanceof Timestamp timestamp) {
+            return timestamp.toLocalDateTime().atOffset(ZoneOffset.UTC);
+        }
+        if (value instanceof LocalDateTime localDateTime) {
+            return localDateTime.atOffset(ZoneOffset.UTC);
+        }
+        String text = value.toString();
+        try {
+            return OffsetDateTime.parse(text).withOffsetSameInstant(ZoneOffset.UTC);
+        } catch (RuntimeException ignored) {
+            return LocalDateTime.parse(text).atOffset(ZoneOffset.UTC);
+        }
+    }
+
+    private BigDecimal decimal(ResultSet rs, String column) throws SQLException {
+        Object value = rs.getObject(column);
+        if (value == null) {
+            return null;
+        }
+        if (value instanceof BigDecimal bigDecimal) {
+            return bigDecimal;
+        }
+        if (value instanceof Number number) {
+            return BigDecimal.valueOf(number.doubleValue());
+        }
+        return new BigDecimal(value.toString());
+    }
+
+    private EventLifecycle lifecycle(ResultSet rs) throws SQLException {
+        return parseEnum(EventLifecycle.class, string(rs, "lifecycle"), EventLifecycle.BEGIN);
+    }
+
+    private <T extends Enum<T>> T parseEnum(Class<T> type, String value, T fallback) {
+        if (value == null) {
+            return fallback;
+        }
+        String normalized = value.trim().toUpperCase(Locale.ROOT).replace('-', '_').replace(' ', '_');
+        try {
+            return Enum.valueOf(type, normalized);
+        } catch (IllegalArgumentException ignored) {
+            return fallback;
+        }
+    }
+
+    private void put(Map<String, Object> target, String key, Object value) {
+        if (value != null) {
+            target.put(key, value);
+        }
+    }
+}

src/main/java/at/procon/eventhub/tachograph/service/TachographExtractionDefinitionRegistry.java

@@ -24,7 +24,8 @@ public class TachographExtractionDefinitionRegistry extends ExtractionDefinition
             VuPositionRowMapper vuPositionRowMapper,
             CardPositionRowMapper cardPositionRowMapper,
             VuPlaceRowMapper vuPlaceRowMapper,
-            CardPlaceRowMapper cardPlaceRowMapper
+            CardPlaceRowMapper cardPlaceRowMapper,
+            SpeedingEventRowMapper speedingEventRowMapper
     ) {
         super(List.of(
                 new ExtractionDefinition<>(
@@ -138,6 +139,14 @@ public class TachographExtractionDefinitionRegistry extends ExtractionDefinition
                         "DRIVER",
                         "classpath:sql/tachograph/card-place.sql",
                         cardPlaceRowMapper
+                ),
+                new ExtractionDefinition<>(
+                        "SPEEDING_EVENTS",
+                        EventFamily.SPEEDING,
+                        "VEHICLE_UNIT",
+                        "VEHICLE",
+                        "classpath:sql/tachograph/speeding-events.sql",
+                        speedingEventRowMapper
                 )
         ));
     }

src/main/resources/sql/tachograph/speeding-events.sql

@@ -0,0 +1,118 @@
+/*
+ * SpeedingEvents SPEEDING extraction for the bytebar tachograph schema.
+ *
+ * The source row is an interval. The normalized event contract is point-based,
+ * so this query emits a BEGIN point at BeginTime and an END point at EndTime.
+ */
+with OrgTree as (
+    select org.I_90021_OID
+    from dbo.GetOrganisationTree(null, :organisationId, 0, null) org
+    where :organisationId is not null
+)
+,
+Base as (
+    select
+        speeding.ID,
+        speeding.BeginTime,
+        speeding.EndTime,
+        speeding.AvgSpeed,
+        speeding.MaxSpeed,
+        speeding.ID_FileLog,
+        speeding.ID_VUInstallation,
+        c.ID as card_id,
+        c.ID_Driver as driver_id,
+        cn.AlphaCode as driver_card_nation,
+        c.CardNumber as driver_card_number,
+        vui.ID_FileLog as vui_filelog_id,
+        vui.ID_VehicleIdentification,
+        vi.ID as vehicle_identification_id,
+        vi.VIN as vehicle_vin,
+        coalesce(fl.DownloadDate, fl.OriginalDownloadDate, fl.TStamp, fl.CreationDate) as received_partner_at,
+        coalesce(fl.ID, speeding.ID_FileLog, vui.ID_FileLog, speeding.ID) as source_package_id_raw,
+        coalesce(fl.ID_VehicleIdentification, vui.ID_VehicleIdentification) as source_package_entity_id_raw,
+        coalesce(fl.DownloadFrom, speeding.BeginTime) as source_package_period_from,
+        coalesce(fl.DownloadTo, speeding.EndTime, speeding.BeginTime) as source_package_period_to,
+        coalesce(fl.CreationDate, fl.TStamp) as source_package_imported_at
+    from dbo.SpeedingEvents speeding
+    join dbo.VUInstallation vui on vui.ID = speeding.ID_VUInstallation
+    join dbo.VehicleIdentification vi on vi.ID = vui.ID_VehicleIdentification
+    left join dbo.Card c on c.ID = speeding.ID_Card
+    left join dbo.Nation cn on cn.ID = c.ID_Nation
+    left join dbo.FileLog fl on fl.ID = coalesce(speeding.ID_FileLog, vui.ID_FileLog)
+    where (
+          :occurredFrom is null
+          or speeding.BeginTime >= :occurredFrom
+          or speeding.EndTime >= :occurredFrom
+      )
+      and (
+          :occurredTo is null
+          or speeding.BeginTime < :occurredTo
+          or speeding.EndTime < :occurredTo
+      )
+)
+,
+Events as (
+    select
+        base.*,
+        base.BeginTime as occurred_at,
+        'BEGIN' as lifecycle
+    from Base base
+    union all
+    select
+        base.*,
+        base.EndTime as occurred_at,
+        'END' as lifecycle
+    from Base base
+)
+select
+    concat(cast(events.ID as varchar(128)), ':', events.lifecycle) as source_row_id,
+    concat('TACHOGRAPH:SPEEDING_EVENTS:', events.ID, ':', events.lifecycle) as external_source_event_id,
+
+    events.occurred_at,
+    events.received_partner_at,
+    cast(events.AvgSpeed as decimal(10, 2)) as avg_speed_kmh,
+    cast(events.MaxSpeed as decimal(10, 2)) as max_speed_kmh,
+    events.lifecycle,
+
+    cast(events.driver_id as varchar(128)) as driver_source_entity_id,
+    events.driver_card_nation,
+    events.driver_card_number,
+
+    cast(events.vehicle_identification_id as varchar(128)) as vehicle_source_entity_id,
+    events.vehicle_vin,
+    cast(v.ID as varchar(128)) as vehicle_registration_source_entity_id,
+    vn.AlphaCode as vehicle_registration_nation,
+    v.VRN as vehicle_registration_number,
+
+    'VEHICLE_UNIT' as source_package_kind,
+    cast(events.source_package_id_raw as varchar(128)) as source_package_id,
+    cast(events.source_package_entity_id_raw as varchar(128)) as source_package_entity_id,
+    events.source_package_period_from,
+    events.source_package_period_to,
+    events.source_package_imported_at
+from Events events
+outer apply (
+    select top 1 vehicle.ID,
+                 vehicle.VRN,
+                 vehicle.ID_Nation
+    from dbo.Vehicle vehicle
+    where vehicle.ID_VehicleIdentification = events.vehicle_identification_id
+      and (vehicle.ValidFrom is null or vehicle.ValidFrom <= events.occurred_at)
+      and (vehicle.ValidTo is null or vehicle.ValidTo > events.occurred_at)
+    order by
+        vehicle.ValidFrom desc,
+        vehicle.ID desc
+) v
+left join dbo.Nation vn on vn.ID = v.ID_Nation
+where (:occurredFrom is null or events.occurred_at >= :occurredFrom)
+  and (:occurredTo is null or events.occurred_at < :occurredTo)
+  and (
+      :organisationId is null
+      or exists (
+          select 1
+          from dbo.Vehicle_I_90021 rel
+          join OrgTree on OrgTree.I_90021_OID = rel.ID_I_90021
+          where rel.ID_Vehicle = v.ID
+            and rel.GILT_BIS is null
+      )
+  )

src/test/java/at/procon/eventhub/tachograph/service/TachographImportPlanServiceTest.java

@@ -13,7 +13,6 @@ import java.util.List;
 import org.junit.jupiter.api.Test;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 class TachographImportPlanServiceTest {
 
@@ -32,27 +31,10 @@ class TachographImportPlanServiceTest {
             new VuPositionRowMapper(detailsFactory),
             new CardPositionRowMapper(detailsFactory),
             new VuPlaceRowMapper(detailsFactory),
-            new CardPlaceRowMapper(detailsFactory)
+            new CardPlaceRowMapper(detailsFactory),
+            new SpeedingEventRowMapper(detailsFactory)
     );
 
-    @Test
-    void rejectsUnsupportedEventFamiliesWhenJdbcExtractionIsEnabled() {
-        TachographImportPlanService service = serviceWithJdbcExtractor();
-        TachographImportRequest request = requestForFamilies(EventFamily.SPEEDING);
-
-        assertThatThrownBy(() -> service.createPlan(request))
-                .isInstanceOf(UnsupportedTachographExtractionException.class)
-                .hasMessageContaining("SPEEDING_EVENTS")
-                .hasMessageContaining("Supported JDBC extraction codes")
-                .hasMessageContaining("DRIVER_ACTIVITY")
-                .hasMessageContaining("DRIVER_CARD")
-                .hasMessageContaining("BORDER_CROSSING")
-                .hasMessageContaining("LOAD_UNLOAD")
-                .hasMessageContaining("SPECIFIC_CONDITION")
-                .hasMessageContaining("POSITION")
-                .hasMessageContaining("PLACE");
-    }
-
     @Test
     void allowsSupportedDriverActivityFamilyWhenJdbcExtractionIsEnabled() {
         TachographImportPlanService service = serviceWithJdbcExtractor();
@@ -137,6 +119,18 @@ class TachographImportPlanServiceTest {
                 .containsExactlyInAnyOrder("VU_PLACE", "CARD_PLACE");
     }
 
+    @Test
+    void allowsSupportedSpeedingFamilyWhenJdbcExtractionIsEnabled() {
+        TachographImportPlanService service = serviceWithJdbcExtractor();
+        TachographImportRequest request = requestForFamilies(EventFamily.SPEEDING);
+
+        var plan = service.createPlan(request);
+
+        assertThat(plan.items())
+                .extracting(item -> item.extractionCode())
+                .containsExactly("SPEEDING_EVENTS");
+    }
+
     private TachographImportPlanService serviceWithJdbcExtractor() {
         EventHubProperties properties = new EventHubProperties();
         properties.getTachograph().getDatasource().setJdbcUrl("jdbc:sqlserver://tachograph-db");