Feature/ab#2252 implement pull to zero in times of no activity

README.MD

@@ -42,6 +42,23 @@ In a defined area this job counts how many vehicles are not moving and so consid
 It works by reconstructing and windowing object trajectories (length of analyzing window: `areaOccupancy.analyzingWindow`). Every new detection is added to its corresponding object trajectory. In regular intervals (`areaOccupancy.analyzingIntervalMs`) the trajectories are evaluated. The algorithm checks all objects that have trajectories of at least 80% the length of `areaOccupancy.analyzingWindow` if they have been stationary within that window. If yes, the object is added to the count.
 The stationary detection works by taking the entire (windowed) trajectory, calculating the average position and then checking if the average position is within the defined area and the 95-percentile of distances to that average point is smaller than the given threshold. For non-geo-referenced jobs the property `areaOccupancy.pxDistanceP95ThresholdScale` defines that threshold on the 95-percentile through a scaling factor on the individual bounding box diagonal (to account for objects being smaller if they're further away from the camera). For geo-referenced jobs the property `areaOccupancy.geoDistanceP95Threshold` directly defines the 95-percentile threshold in coordinate space (there is a known relationship to distance in meters). E.g. in northern Germany a distance of 0.0001 in geo-space translates to roughly 10m.
 
+### Flow-based Area Occupancy
+
+As an alternative to the detection-based approach above, area occupancy can also be maintained by counting in/out flow events (e.g. from a `LINE_CROSSING` job at an entrance). Each inbound event increments the count by one, each outbound event decrements it (clamped to zero). An optional maximum count can be configured per job to cap the value.
+
+Because events can be missed (e.g. due to occlusion or tracking gaps), the count may drift upward over time even when the area is empty. The **idle decrease** feature compensates for this by gradually pulling the count towards zero whenever a long gap between events is detected:
+
+| Property | Default | Description |
+|---|---|---|
+| `areaOccupancy.flowIdleDecreaseEnabled` | `false` | Enable/disable idle decrease |
+| `areaOccupancy.flowIdleDecreaseThreshold` | `1h` | Minimum gap before decrease kicks in |
+| `areaOccupancy.flowIdleDecreaseInterval` | `15m` | Count is reduced by one step per interval |
+| `areaOccupancy.flowIdleDecreaseFactor` | `1` | Amount subtracted per step |
+
+When a new flow event arrives and the elapsed time since the last recorded entry exceeds `flowIdleDecreaseThreshold`, the number of full `flowIdleDecreaseInterval` periods in that gap is computed and `steps × flowIdleDecreaseFactor` is subtracted from the count (after applying the in/out delta), clamped at zero.
+
+These properties can be set in `application.properties` or via the corresponding Helm values under `areaOccupancy.*`.
+
 ## How to Deploy/Install
 
 Helm is the preferred tool to install Observatory. Installation can be done with the following command:

application/src/main/resources/application.properties

@@ -48,6 +48,10 @@ spring.data.redis.active=true
 # areaOccupancy.analyzingWindow=5s
 # areaOccupancy.geoDistanceP95Threshold=0.001
 # areaOccupancy.pxDistanceP95ThresholdScale=0.1
+# areaOccupancy.flowIdleDecreaseEnabled=false
+# areaOccupancy.flowIdleDecreaseThreshold=1h
+# areaOccupancy.flowIdleDecreaseInterval=15m
+# areaOccupancy.flowIdleDecreaseFactor=1
 
 analytics.datasource.username=analytics
 analytics.datasource.password=analytics

deployment/helm/observatory/templates/statefulset.yaml

@@ -83,7 +83,14 @@ spec:
             - name: ANALYTICS_TIMESCALE_DATASOURCE_FLYWAY_LOCATIONS
               value: "classpath:db/migration/analytics-timescale"
             {{- end }}
-
+            - name: AREAOCCUPANCY_FLOWIDLEDECREASEENABLED
+              value: {{ .Values.areaOccupancy.flowIdleDecreaseEnabled | quote }}
+            - name: AREAOCCUPANCY_FLOWIDLEDECREASETHRESHOLD
+              value: {{ .Values.areaOccupancy.flowIdleDecreaseThreshold | quote }}
+            - name: AREAOCCUPANCY_FLOWIDLEDECREASEINTERVAL
+              value: {{ .Values.areaOccupancy.flowIdleDecreaseInterval | quote }}
+            - name: AREAOCCUPANCY_FLOWIDLEDECREASEFACTOR
+              value: {{ .Values.areaOccupancy.flowIdleDecreaseFactor | quote }}
             - name: SERVER_SERVLET_CONTEXT_PATH
               value: {{ .Values.app.context_path | quote }}
             {{- with .Values.extraEnv }}

deployment/helm/observatory/values.yaml

@@ -18,6 +18,12 @@ app:
 
 analyticsJobRunInterval: 2000
 
+areaOccupancy:
+  flowIdleDecreaseEnabled: false
+  flowIdleDecreaseThreshold: 1h
+  flowIdleDecreaseInterval: 15m
+  flowIdleDecreaseFactor: 1
+
 spring:
   data:
     redis:

persistence/src/main/java/de/starwit/persistence/analytics/repository/AreaOccupancyRepository.java

@@ -41,15 +41,13 @@ public List<AreaOccupancyEntity> findFirst100() {
         return getEntityManager().createNativeQuery(queryString).getResultList();
     }
 
-    public Integer findFirstByMetadataIdAndObjectClassIdOrderByOccupancytime(Long metadataId, Integer objectClassId) {
-        String queryString = "select a.count from areaoccupancy a where metadata_id = :metadataId and object_class_id = :objectClassId order by occupancy_time desc limit 1";
-        try {
-            return (Integer) getEntityManager().createNativeQuery(queryString)
-                    .setParameter("metadataId", metadataId)
-                    .setParameter("objectClassId", objectClassId)
-                    .getSingleResult();
-        } catch (Exception e) {
-            return 0;
-        }
+    public AreaOccupancyEntity findFirstByMetadataIdAndObjectClassIdOrderByOccupancytime(Long metadataId, Integer objectClassId) {
+        String queryString = "select * from areaoccupancy where metadata_id = :metadataId and object_class_id = :objectClassId order by occupancy_time desc limit 1";
+        List<AreaOccupancyEntity> result = getEntityManager()
+                .createNativeQuery(queryString, AreaOccupancyEntity.class)
+                .setParameter("metadataId", metadataId)
+                .setParameter("objectClassId", objectClassId)
+                .getResultList();
+        return result.isEmpty() ? null : result.get(0);
     }
 }

service/src/main/java/de/starwit/service/analytics/AreaOccupancyService.java

@@ -1,12 +1,14 @@
 package de.starwit.service.analytics;
 
+import java.time.Duration;
 import java.time.Instant;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -27,6 +29,18 @@ public class AreaOccupancyService {
     @Autowired
     private MetadataService metadataService;
 
+    @Value("${areaOccupancy.flowIdleDecreaseEnabled:false}")
+    private boolean FLOW_IDLE_DECREASE_ENABLED;
+
+    @Value("${areaOccupancy.flowIdleDecreaseThreshold:1h}")
+    private Duration FLOW_IDLE_DECREASE_THRESHOLD;
+
+    @Value("${areaOccupancy.flowIdleDecreaseInterval:15m}")
+    private Duration FLOW_IDLE_DECREASE_INTERVAL;
+
+    @Value("${areaOccupancy.flowIdleDecreaseFactor:1}")
+    private int FLOW_IDLE_DECREASE_FACTOR;
+
     @Transactional("analyticsTransactionManager")
     public void addEntry(ObservationJobEntity jobEntity, ZonedDateTime occupancyTime, Long count) {
         log.info("Detected {} objects of class {} in area (area={}, name={})", count, jobEntity.getDetectionClassId(),
@@ -64,26 +78,37 @@ public void updateCountFromFlow(ObservationJobEntity jobEntity, ZonedDateTime oc
 
         MetadataEntity metadata = metadataService.saveMetadataForJob(jobEntity);
 
-        Integer lastCount = 0;
-
-        lastCount = areaoccupancyRepository.findFirstByMetadataIdAndObjectClassIdOrderByOccupancytime(metadata.getId(),
-                jobEntity.getDetectionClassId());
-        AreaOccupancyEntity entity = new AreaOccupancyEntity();
+        AreaOccupancyEntity lastEntity = areaoccupancyRepository
+                .findFirstByMetadataIdAndObjectClassIdOrderByOccupancytime(metadata.getId(),
+                        jobEntity.getDetectionClassId());
+        int lastCount = lastEntity == null ? 0 : lastEntity.getCount();
 
+        int newCount;
         if (Direction.in.equals(direction)) {
             if (jobEntity.getMaxCount() == null || lastCount < jobEntity.getMaxCount()) {
-                entity.setCount(lastCount + 1);
+                newCount = lastCount + 1;
             } else {
                 log.info("Max count of {} for job {} reached. Resetting to max count.", jobEntity.getMaxCount(),
                         jobEntity.getName());
-                entity.setCount(jobEntity.getMaxCount());
+                newCount = jobEntity.getMaxCount();
             }
         } else if (lastCount > 0) {
-            entity.setCount(lastCount - 1);
+            newCount = lastCount - 1;
         } else {
-            entity.setCount(0);
+            newCount = 0;
         }
 
+        // Slowly pull count towards zero after times of no activity (assumption is that area is empty if no activity)
+        if (FLOW_IDLE_DECREASE_ENABLED && lastEntity != null) {
+            Duration elapsed = Duration.between(lastEntity.getOccupancyTime(), occupancyTime);
+            if (elapsed.compareTo(FLOW_IDLE_DECREASE_THRESHOLD) > 0) {
+                long steps = elapsed.dividedBy(FLOW_IDLE_DECREASE_INTERVAL);
+                newCount = Math.max(0, newCount - Math.toIntExact(steps * FLOW_IDLE_DECREASE_FACTOR));
+            }
+        }
+
+        AreaOccupancyEntity entity = new AreaOccupancyEntity();
+        entity.setCount(newCount);
         entity.setOccupancyTime(occupancyTime);
         entity.setObjectClassId(jobEntity.getDetectionClassId());
         entity.setMetadataId(metadata.getId());

service/src/test/java/de/starwit/service/analytics/AreaOccupancyServiceTest.java

@@ -0,0 +1,142 @@
+package de.starwit.service.analytics;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InjectMocks;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.test.context.bean.override.mockito.MockitoBean;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import de.starwit.persistence.analytics.entity.AreaOccupancyEntity;
+import de.starwit.persistence.analytics.entity.Direction;
+import de.starwit.persistence.analytics.entity.MetadataEntity;
+import de.starwit.persistence.analytics.repository.AreaOccupancyRepository;
+import de.starwit.persistence.observatory.entity.ObservationJobEntity;
+
+@ExtendWith(MockitoExtension.class)
+@ExtendWith(SpringExtension.class)
+public class AreaOccupancyServiceTest {
+
+    @MockitoBean
+    AreaOccupancyRepository areaoccupancyRepository;
+
+    @MockitoBean
+    MetadataService metadataService;
+
+    @InjectMocks
+    AreaOccupancyService areaOccupancyService;
+
+    static final ZonedDateTime TIME = Instant.parse("2026-06-16T00:00:00Z").atZone(ZoneOffset.UTC);
+
+    @BeforeEach
+    public void setupMocks() {
+        ReflectionTestUtils.setField(areaOccupancyService, "FLOW_IDLE_DECREASE_ENABLED", true);
+        ReflectionTestUtils.setField(areaOccupancyService, "FLOW_IDLE_DECREASE_THRESHOLD", Duration.ofHours(1));
+        ReflectionTestUtils.setField(areaOccupancyService, "FLOW_IDLE_DECREASE_INTERVAL", Duration.ofMinutes(15));
+        ReflectionTestUtils.setField(areaOccupancyService, "FLOW_IDLE_DECREASE_FACTOR", 1);
+
+        MetadataEntity metadata = new MetadataEntity();
+        metadata.setId(1L);
+        when(metadataService.saveMetadataForJob(any())).thenReturn(metadata);
+    }
+
+    @Test
+    public void testNoDecayOnFirstEntry() {
+        when(areaoccupancyRepository.findFirstByMetadataIdAndObjectClassIdOrderByOccupancytime(any(), any()))
+                .thenReturn(null);
+
+        areaOccupancyService.updateCountFromFlow(jobEntity(null), TIME, Direction.in);
+
+        assertThat(captureInsertedCount()).isEqualTo(1);
+    }
+
+    @Test
+    public void testShortGapNoDecay() {
+        when(areaoccupancyRepository.findFirstByMetadataIdAndObjectClassIdOrderByOccupancytime(any(), any()))
+                .thenReturn(lastEntry(5, TIME));
+
+        areaOccupancyService.updateCountFromFlow(jobEntity(null), TIME.plusMinutes(30), Direction.in);
+
+        assertThat(captureInsertedCount()).isEqualTo(6);
+    }
+
+    @Test
+    public void testLongGapAppliesExactStepDecay() {
+        when(areaoccupancyRepository.findFirstByMetadataIdAndObjectClassIdOrderByOccupancytime(any(), any()))
+                .thenReturn(lastEntry(10, TIME));
+
+        areaOccupancyService.updateCountFromFlow(jobEntity(null), TIME.plusHours(2), Direction.out);
+
+        // delta first: 10 - 1 = 9; decay: 7200s / 900s = 8 steps -> max(0, 9 - 8) = 1
+        assertThat(captureInsertedCount()).isEqualTo(1);
+    }
+
+    @Test
+    public void testDecayClampsAtZero() {
+        when(areaoccupancyRepository.findFirstByMetadataIdAndObjectClassIdOrderByOccupancytime(any(), any()))
+                .thenReturn(lastEntry(2, TIME));
+
+        areaOccupancyService.updateCountFromFlow(jobEntity(null), TIME.plusHours(3), Direction.out);
+
+        // delta first: 2 - 1 = 1; decay: 10800s / 900s = 12 steps -> max(0, 1 - 12) = 0
+        assertThat(captureInsertedCount()).isEqualTo(0);
+    }
+
+    @Test
+    public void testGapExactlyAtThresholdNoDecay() {
+        when(areaoccupancyRepository.findFirstByMetadataIdAndObjectClassIdOrderByOccupancytime(any(), any()))
+                .thenReturn(lastEntry(5, TIME));
+
+        areaOccupancyService.updateCountFromFlow(jobEntity(null), TIME.plusHours(1), Direction.in);
+
+        assertThat(captureInsertedCount()).isEqualTo(6);
+    }
+
+    @Test
+    public void testNoDecayWhenDisabled() {
+        ReflectionTestUtils.setField(areaOccupancyService, "FLOW_IDLE_DECREASE_ENABLED", false);
+        when(areaoccupancyRepository.findFirstByMetadataIdAndObjectClassIdOrderByOccupancytime(any(), any()))
+                .thenReturn(lastEntry(10, TIME));
+
+        areaOccupancyService.updateCountFromFlow(jobEntity(null), TIME.plusHours(2), Direction.out);
+
+        // delta only: 10 - 1 = 9; decay would otherwise apply since elapsed (2h) exceeds threshold (1h)
+        assertThat(captureInsertedCount()).isEqualTo(9);
+    }
+
+    private int captureInsertedCount() {
+        ArgumentCaptor<AreaOccupancyEntity> captor = ArgumentCaptor.forClass(AreaOccupancyEntity.class);
+        org.mockito.Mockito.verify(areaoccupancyRepository).insert(captor.capture());
+        return captor.getValue().getCount();
+    }
+
+    private static AreaOccupancyEntity lastEntry(int count, ZonedDateTime time) {
+        AreaOccupancyEntity entity = new AreaOccupancyEntity();
+        entity.setCount(count);
+        entity.setOccupancyTime(time);
+        entity.setObjectClassId(1);
+        entity.setMetadataId(1L);
+        return entity;
+    }
+
+    private static ObservationJobEntity jobEntity(Integer maxCount) {
+        ObservationJobEntity entity = new ObservationJobEntity();
+        entity.setName("job1");
+        entity.setDetectionClassId(1);
+        entity.setObservationAreaId(1L);
+        entity.setMaxCount(maxCount);
+        return entity;
+    }
+}