diff --git a/wizard-work/beam/intervals.py b/wizard-work/beam/intervals.py new file mode 100644 index 0000000..3bb5833 --- /dev/null +++ b/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 diff --git a/wizard-work/beam/test_intervals.py b/wizard-work/beam/test_intervals.py new file mode 100644 index 0000000..5947634 --- /dev/null +++ b/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)) \ No newline at end of file