beam: interval merge/overlap/gaps utils with fuzz tests

wizard-work/beam/intervals.py

@@ -0,0 +1,82 @@
+"""Interval overlap/merge utilities with well-defined boundary semantics.
+
+Boundary convention: intervals are half-open [start, end). Two intervals
+overlap iff they share at least one point, i.e. max(s1,s2) < min(e1,e2).
+Touching at a point (e.g. [1,2) and [2,3)) does NOT overlap and is NOT
+merged (they remain distinct). This matches the common scheduling convention.
+"""
+from __future__ import annotations
+
+from typing import Iterable, List, Tuple
+
+Interval = Tuple[int, int]
+
+
+def _validate(iv: Interval) -> None:
+    if not (isinstance(iv, tuple) and len(iv) == 2):
+        raise ValueError(f"interval must be 2-tuple, got {iv!r}")
+    s, e = iv
+    if not (isinstance(s, int) and isinstance(e, int)):
+        raise ValueError(f"endpoints must be int, got {iv!r}")
+    if s > e:
+        raise ValueError(f"start > end in {iv!r}")
+
+
+def normalize(intervals: Iterable[Interval]) -> List[Interval]:
+    """Validate, drop empties, return sorted by (start, end)."""
+    out: List[Interval] = []
+    for iv in intervals:
+        _validate(iv)
+        s, e = iv
+        if s == e:
+            continue  # empty interval contributes nothing
+        out.append((s, e))
+    out.sort(key=lambda x: (x[0], x[1]))
+    return out
+
+
+def overlaps(a: Interval, b: Interval) -> bool:
+    _validate(a); _validate(b)
+    return max(a[0], b[0]) < min(a[1], b[1])
+
+
+def merge(intervals: Iterable[Interval]) -> List[Interval]:
+    """Merge overlapping intervals (half-open). Touching intervals stay separate."""
+    ivs = normalize(intervals)
+    if not ivs:
+        return []
+    merged: List[Interval] = [ivs[0]]
+    for s, e in ivs[1:]:
+        ls, le = merged[-1]
+        if s < le:  # strict: touching (s == le) does not merge
+            merged[-1] = (ls, max(le, e))
+        else:
+            merged.append((s, e))
+    return merged
+
+
+def total_covered(intervals: Iterable[Interval]) -> int:
+    """Total length of union of intervals (half-open length = end - start)."""
+    return sum(e - s for s, e in merge(intervals))
+
+
+def find_gaps(intervals: Iterable[Interval], lo: int, hi: int) -> List[Interval]:
+    """Gaps within [lo, hi) not covered by any interval."""
+    if lo > hi:
+        raise ValueError("lo > hi")
+    merged = merge(intervals)
+    gaps: List[Interval] = []
+    cur = lo
+    for s, e in merged:
+        if e <= cur or s >= hi:
+            continue
+        ss = max(s, lo)
+        ee = min(e, hi)
+        if ss > cur:
+            gaps.append((cur, ss))
+        cur = max(cur, ee)
+        if cur >= hi:
+            break
+    if cur < hi:
+        gaps.append((cur, hi))
+    return gaps

wizard-work/beam/test_intervals.py

@@ -0,0 +1,56 @@
+import random
+from intervals import merge, overlaps, total_covered, find_gaps, normalize
+
+def test_merge_basic():
+    assert merge([(1,3),(2,4)]) == [(1,4)]
+    assert merge([(1,2),(3,4)]) == [(1,2),(3,4)]  # touching-ish? no, gap
+    assert merge([(1,2),(2,3)]) == [(1,2),(2,3)]  # touching stays separate
+
+def test_merge_nested_unsorted():
+    assert merge([(5,9),(1,10),(2,3)]) == [(1,10)]
+
+def test_empty():
+    assert merge([]) == []
+    assert total_covered([]) == 0
+
+def test_total_covered():
+    assert total_covered([(1,4),(2,5)]) == 4
+    assert total_covered([(1,2),(2,3)]) == 2
+
+def test_overlaps():
+    assert overlaps((1,3),(2,4)) is True
+    assert overlaps((1,2),(2,3)) is False
+    assert overlaps((1,5),(2,3)) is True
+
+def test_find_gaps():
+    assert find_gaps([(1,3),(5,7)], 0, 8) == [(0,1),(3,5),(7,8)]
+    assert find_gaps([], 0, 5) == [(0,5)]
+    assert find_gaps([(0,5)], 0, 5) == []
+
+def test_validate():
+    try:
+        merge([(2,1)])
+        raise AssertionError("expected ValueError")
+    except ValueError:
+        pass
+
+def test_fuzz_against_brute():
+    rng = random.Random(42)
+    for _ in range(2000):
+        n = rng.randint(0, 8)
+        ivs = []
+        for _ in range(n):
+            a = rng.randint(-5, 5)
+            b = rng.randint(-5, 5)
+            ivs.append((min(a,b), max(a,b)))
+        # brute union length over integer points? use half-open lengths with integer grid
+        merged = merge(ivs)
+        # check no overlaps remain
+        for i in range(len(merged)-1):
+            assert merged[i][1] <= merged[i+1][0], (merged, ivs)
+        # check total length equals brute force over unit grid
+        pts = set()
+        for s,e in ivs:
+            for x in range(s, e):
+                pts.add(x)
+        assert total_covered(ivs) == len(pts), (ivs, merged, total_covered(ivs), len(pts))