beam: add min_meeting_rooms (sweep + heap, zero-length safe)

wizard-work/beam/min_meeting_rooms.py

@@ -0,0 +1,48 @@
+"""Minimum meeting rooms.
+
+Given a list of intervals [start, end) (half-open), return the minimum
+number of concurrent "rooms" needed so no two overlapping intervals share
+a room. Touching endpoints (end == start) do NOT overlap.
+
+Approach: sweep-line over events. A start at time t needs a room before an
+end at the same time t frees one, so process ends before starts at equal t.
+O(n log n) time, O(n) space.
+
+Also provided: min_rooms_heap(intervals) using a min-heap of end times, for
+cross-checking.
+"""
+from __future__ import annotations
+import heapq
+
+
+def min_meeting_rooms(intervals):
+    """Return minimum rooms needed for intervals [(s,e), ...]."""
+    events = []
+    for s, e in intervals:
+        if e < s:
+            raise ValueError("end < start: %r" % ((s, e),))
+        if s == e:
+            continue  # empty half-open interval
+        events.append((s, 1))   # start
+        events.append((e, -1))  # end
+    # sort by time; at equal time, ends (-1) before starts (+1)
+    events.sort(key=lambda x: (x[0], x[1]))
+    cur = 0
+    best = 0
+    for _, delta in events:
+        cur += delta
+        if cur > best:
+            best = cur
+    return best
+
+
+def min_rooms_heap(intervals):
+    ends = []
+    for s, e in sorted(intervals):
+        if s >= e:
+            continue  # empty half-open interval needs no room
+        if ends and ends[0] <= s:
+            heapq.heapreplace(ends, e)
+        else:
+            heapq.heappush(ends, e)
+    return len(ends)

wizard-work/beam/test_min_meeting_rooms.py

@@ -0,0 +1,68 @@
+import random
+import unittest
+
+from min_meeting_rooms import min_meeting_rooms, min_rooms_heap
+
+
+class TestMinMeetingRooms(unittest.TestCase):
+    def test_empty(self):
+        self.assertEqual(min_meeting_rooms([]), 0)
+        self.assertEqual(min_rooms_heap([]), 0)
+
+    def test_single(self):
+        self.assertEqual(min_meeting_rooms([(1, 5)]), 1)
+
+    def test_no_overlap(self):
+        self.assertEqual(min_meeting_rooms([(1, 2), (2, 3), (3, 4)]), 1)
+
+    def test_all_overlap(self):
+        self.assertEqual(min_meeting_rooms([(1, 5), (2, 6), (3, 7)]), 3)
+
+    def test_touching_not_overlap(self):
+        self.assertEqual(min_meeting_rooms([(1, 2), (2, 3)]), 1)
+
+    def test_unsorted(self):
+        self.assertEqual(min_meeting_rooms([(5, 9), (1, 3), (4, 6)]), 2)
+
+    def test_bad_interval_raises(self):
+        with self.assertRaises(ValueError):
+            min_meeting_rooms([(5, 3)])
+
+    def test_random_vs_heap(self):
+        rng = random.Random(123)
+        for _ in range(3000):
+            n = rng.randint(0, 8)
+            ivs = []
+            for _ in range(n):
+                s = rng.randint(0, 6)
+                e = s + rng.randint(0, 6)
+                ivs.append((s, e))
+            a = min_meeting_rooms(ivs)
+            b = min_rooms_heap(ivs)
+            self.assertEqual(a, b, (ivs, a, b))
+
+    def test_random_bruteforce(self):
+        # brute force via per-time max concurrency on integer grid
+        rng = random.Random(999)
+        for _ in range(2000):
+            n = rng.randint(0, 7)
+            ivs = []
+            for _ in range(n):
+                s = rng.randint(0, 5)
+                e = s + rng.randint(1, 5)
+                ivs.append((s, e))
+            # brute
+            times = set()
+            for s, e in ivs:
+                for t in range(s, e):
+                    times.add(t)
+            best = 0
+            for t in times:
+                c = sum(1 for s, e in ivs if s <= t < e)
+                if c > best:
+                    best = c
+            self.assertEqual(min_meeting_rooms(ivs), best, (ivs,))
+
+
+if __name__ == "__main__":
+    unittest.main()