fix(transform): split uncertain flood gaps at midpoint (#146)

Author: TimeToBuildBob

aw_transform/flood.py

@@ -9,11 +9,20 @@
 
 
 def flood(events: List[Event], pulsetime: float = 5) -> List[Event]:
-    """
-    Takes a list of events and "floods" any empty space between events by extending one of the surrounding events to cover the empty space.
+    """Fill short gaps between events and merge nearby equal-data events.
+
+    Events are ordered by timestamp and duration. Gaps no larger than
+    ``pulsetime`` are filled. Equal-data neighbours merge across the gap; when
+    their data differs, both events extend to the midpoint. Splitting an
+    uncertain interval evenly is independent of the surrounding event lengths
+    and matches the Rust implementation.
 
-    For more details on flooding, see this issue:
-     - https://github.com/ActivityWatch/activitywatch/issues/124
+    Overlapping equal-data events merge. For overlapping differing-data events,
+    final normalization gives the later event precedence so the result never
+    double-counts time.
+
+    See https://github.com/ActivityWatch/activitywatch/issues/124 for the data
+    collection uncertainty that flooding is intended to handle.
     """
     # Originally written in aw-research: https://github.com/ActivityWatch/aw-analysis/blob/7da1f2cd8552f866f643501de633d74cdecab168/aw_analysis/flood.py
     # NOTE: This algorithm has a lot of smaller details that need to be
@@ -61,27 +70,24 @@ def flood(events: List[Event], pulsetime: float = 5) -> List[Event]:
         elif -negative_gap_trim_thres < gap <= timedelta(seconds=pulsetime):
             e2_end = e2.timestamp + e2.duration
 
-            # Prioritize flooding from the longer event
-            if e1.duration >= e2.duration:
-                if e1.data == e2.data:
-                    # Extend e1 to the end of e2
-                    # Set duration of e2 to zero (mark to delete)
+            if e1.data == e2.data:
+                # Preserve the longer neighbour's extent while merging across
+                # the gap, matching the existing semantics for equal data.
+                if e1.duration >= e2.duration:
                     e1.duration = e2_end - e1.timestamp
                     e2.timestamp = e2_end
                     e2.duration = timedelta(0)
                 else:
-                    # Extend e1 to the start of e2
-                    e1.duration = e2.timestamp - e1.timestamp
-            else:
-                if e1.data == e2.data:
-                    # Extend e2 to the start of e1, discard e1
                     e2.timestamp = e1.timestamp
                     e2.duration = e2_end - e2.timestamp
                     e1.duration = timedelta(0)
-                else:
-                    # Extend e2 backwards to end of e1
-                    e2.timestamp = e1.timestamp + e1.duration
-                    e2.duration = e2_end - e2.timestamp
+            else:
+                # The gap is an interval of uncertainty: without evidence that
+                # either neighbour owns more of it, split it at the midpoint.
+                midpoint = e1.timestamp + e1.duration + gap / 2
+                e1.duration = midpoint - e1.timestamp
+                e2.timestamp = midpoint
+                e2.duration = e2_end - midpoint
 
     # Pairwise flooding can mutate an event after its previous pair has already
     # been processed. Normalize the final stream so downstream consumers never

tests/test_flood.py

@@ -8,15 +8,18 @@
 td1s = timedelta(seconds=1)
 
 
-def test_flood_forward():
+def test_flood_differing_data_meet_at_gap_midpoint():
     events = [
         Event(timestamp=now, duration=10, data={"a": 0}),
-        Event(timestamp=now + 15 * td1s, duration=5, data={"b": 1}),
+        Event(timestamp=now + 14 * td1s, duration=5, data={"b": 1}),
     ]
+
     flooded = flood(events)
-    assert (flooded[0].timestamp + flooded[0].duration) - flooded[
-        1
-    ].timestamp == timedelta(0)
+
+    assert flooded == [
+        Event(timestamp=now, duration=12, data={"a": 0}),
+        Event(timestamp=now + 12 * td1s, duration=7, data={"b": 1}),
+    ]
 
 
 def test_flood_forward_merge():
@@ -29,17 +32,6 @@ def test_flood_forward_merge():
     assert flooded[0].duration == timedelta(seconds=20)
 
 
-def test_flood_backward():
-    events = [
-        Event(timestamp=now, duration=5, data={"a": 0}),
-        Event(timestamp=now + 10 * td1s, duration=10, data={"b": 1}),
-    ]
-    flooded = flood(events)
-    assert (flooded[0].timestamp + flooded[0].duration) - flooded[
-        1
-    ].timestamp == timedelta(0)
-
-
 def test_flood_backward_merge():
     events = [
         Event(timestamp=now, duration=5),
@@ -97,8 +89,8 @@ def test_flood_normalization_preserves_non_overlapping_tail():
 
     assert flooded == [
         Event(timestamp=now, duration=5, data={"title": "first"}),
-        Event(timestamp=now + 5 * td1s, duration=7, data={"title": "second"}),
-        events[2],
+        Event(timestamp=now + 5 * td1s, duration=4.5, data={"title": "second"}),
+        Event(timestamp=now + 9.5 * td1s, duration=3.5, data={"title": "third"}),
     ]
 
 
@@ -137,11 +129,12 @@ def test_flood_with_custom_pulsetime():
     total_duration_default = sum((e.duration for e in flooded_default), timedelta(0))
     assert total_duration_default == timedelta(seconds=10)
 
-    # pulsetime=31: gap (30s) <= pulsetime, so event 1 extends to meet event 2
+    # pulsetime=31: gap (30s) <= pulsetime, so both events extend to midpoint
     flooded_custom = flood(events, pulsetime=31)
-    assert len(flooded_custom) == 2
-    total_duration_custom = sum((e.duration for e in flooded_custom), timedelta(0))
-    assert total_duration_custom == timedelta(seconds=40)
+    assert flooded_custom == [
+        Event(timestamp=now, duration=20, data={"a": 0}),
+        Event(timestamp=now + 20 * td1s, duration=20, data={"b": 1}),
+    ]
 
 
 def test_flood_idempotent():