beam: interval merge utility with property-tested semantics

wizard-work/beam/interval_merge.py

@@ -0,0 +1,67 @@
+"""Interval merge utilities.
+
+Semantics (half-open intervals [start, end)):
+- Intervals with start >= end are invalid (ValueError).
+- Two intervals overlap/touch if a.start <= b.end and b.start <= a.end? We define
+  "merge if they overlap or touch" (closed merging) using half-open touch rule:
+  intervals [a,b) and [b,c) DO merge (adjacent) because there is no gap.
+  intervals [a,b) and [b+1,c) do NOT merge.
+- Output is sorted, disjoint, merged, each as (start, end) tuple.
+- Empty input -> [].
+"""
+
+from typing import Iterable, List, Tuple
+
+Interval = Tuple[int, int]
+
+
+def _validate(iv: Interval) -> None:
+    if not isinstance(iv, (tuple, list)) or len(iv) != 2:
+        raise ValueError(f"interval must be 2-tuple, got {iv!r}")
+    s, e = iv
+    if not isinstance(s, int) or not isinstance(e, int) or isinstance(s, bool) or isinstance(e, bool):
+        raise ValueError(f"endpoints must be ints, got {iv!r}")
+    if s >= e:
+        raise ValueError(f"empty/invalid interval {iv!r}: start must be < end")
+
+
+def merge_intervals(intervals: Iterable[Interval]) -> List[Interval]:
+    ivs = [tuple(iv) for iv in intervals]
+    for iv in ivs:
+        _validate(iv)
+    if not ivs:
+        return []
+    ivs.sort()
+    out: List[Interval] = [ivs[0]]
+    for s, e in ivs[1:]:
+        ps, pe = out[-1]
+        if s <= pe:  # overlap or touch (half-open touch merges)
+            out[-1] = (ps, max(pe, e))
+        else:
+            out.append((s, e))
+    return out
+
+
+def total_covered(intervals: Iterable[Interval]) -> int:
+    return sum(e - s for s, e in merge_intervals(intervals))
+
+
+def gaps(intervals: Iterable[Interval], lo: int, hi: int) -> List[Interval]:
+    """Return merged gaps within [lo, hi) not covered by intervals."""
+    if lo >= hi:
+        raise ValueError("lo must be < hi")
+    merged = merge_intervals(intervals)
+    out: List[Interval] = []
+    cur = lo
+    for s, e in merged:
+        s = max(s, lo); e = min(e, hi)
+        if s >= e:
+            continue
+        if s > cur:
+            out.append((cur, s))
+        cur = max(cur, e)
+        if cur >= hi:
+            break
+    if cur < hi:
+        out.append((cur, hi))
+    return out

wizard-work/beam/test_interval.py

@@ -0,0 +1,63 @@
+import random
+import sys
+sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")
+from interval_merge import merge_intervals, total_covered, gaps
+
+def brute_merge(intervals):
+    pts = set()
+    # discrete integer points model for small ranges
+    pts_all = set()
+    for s, e in intervals:
+        for x in range(s, e):
+            pts_all.add(x)
+    if not pts_all:
+        return []
+    # build runs
+    out = []
+    for x in sorted(pts_all):
+        if not out or x != out[-1][1]:
+            out.append([x, x+1])
+        else:
+            out[-1][1] = x+1
+    return [tuple(r) for r in out]
+
+random.seed(42)
+for trial in range(2000):
+    n = random.randint(0, 8)
+    ivs = []
+    for _ in range(n):
+        s = random.randint(-5, 5)
+        e = s + random.randint(1, 6)
+        ivs.append((s, e))
+    got = merge_intervals(ivs)
+    exp = brute_merge(ivs)
+    assert got == exp, (ivs, got, exp)
+
+# total_covered
+assert total_covered([(1,4),(3,6),(10,12)]) == (6-1)+(12-10)  # 5+2=7? 1..6 =5, 10..12=2 =>7
+assert total_covered([]) == 0
+
+# gaps
+assert gaps([(2,4)], 0, 6) == [(0,2),(4,6)]
+assert gaps([(0,6)], 0, 6) == []
+assert gaps([], 0, 3) == [(0,3)]
+
+# validation
+for bad in [((1,1),(2,3)), (1,2,3), (True,5), (1,True), ("a",2), (2,1)]:
+    try:
+        merge_intervals([bad] if not isinstance(bad[0], tuple) or len(bad)!=2 or not isinstance(bad[0], int) else [bad])
+    except ValueError:
+        pass
+    else:
+        # some cases may be wrapped; ensure at least invalid ones raise
+        pass
+
+# explicit invalid
+for bad in [(1,1),(2,1),(True,3),(1,True)]:
+    try:
+        merge_intervals([bad])
+        raise AssertionError(f"no raise for {bad}")
+    except ValueError:
+        pass
+
+print("ALL TESTS PASS")