Event queue in Python
Priority queue that yields the next event point in sweep order.
- File
event_queue.py, 64 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/event-queue/. The authors are listed in prov/324RJK.jsonl.
import heapq
from typing import Generic, Hashable, TypeVar
P = TypeVar("P", bound=Hashable) # an event point, ordered by <, for example (x, y)
A = TypeVar("A") # a payload, for example a segment that starts at the point
class EventQueue(Generic[P, A]):
"""The event queue of a sweep: a min-priority queue of distinct event points.
An insert of a point that is already present does not make a second
event: its payload joins the list of the existing event. A point is
stored once in a binary heap and once as a key of a hash map from the
point to its list of payloads. With m distinct points present, insert
and pop_min do O(log m) comparisons of points, plus O(1) expected time
for the hash map. The space is O(m + the number of payloads).
"""
def __init__(self) -> None:
self._heap: list[P] = []
self._index: dict[P, list[A]] = {}
def insert(self, p: P, payload: A | None = None) -> None:
"""Add event point p. If payload is not None, append it to the payloads of p."""
if p in self._index: # @line L1
if payload is not None:
self._index[p].append(payload) # @line L2
else:
self._index[p] = [] if payload is None else [payload] # @line L3
heapq.heappush(self._heap, p) # @line L4
def pop_min(self) -> tuple[P, list[A]]:
"""Remove the smallest event point and return it with all its payloads.
Precondition: the queue is not empty.
"""
if not self._heap:
raise IndexError("pop_min from an empty event queue")
p = heapq.heappop(self._heap) # @line L5
payloads = self._index.pop(p) # @line L6
return p, payloads # @line L7
def contains(self, p: P) -> bool:
"""Return True if event point p is in the queue."""
return p in self._index # @line L8
def is_empty(self) -> bool:
"""Return True if the queue holds no event point."""
return not self._heap # @line L9
def __len__(self) -> int:
"""Return the number of distinct event points in the queue."""
return len(self._heap)
def check_invariant(self) -> None:
"""Check that the heap and the map hold the same distinct points and that the heap is ordered."""
h = self._heap
assert len(h) == len(self._index), "heap and index differ in size"
assert set(h) == set(self._index), "heap and index hold different points"
for i in range(1, len(h)):
assert not (h[i] < h[(i - 1) // 2]), "heap order is broken"
test_event_queue.py 92 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/event-queue/. The authors are listed in prov/324RJK.jsonl.
import random
import pytest
from event_queue import EventQueue
def test_example() -> None:
# Segments a: (0,0)-(10,10), b: (0,10)-(10,0), c: (1,5)-(3,5), as on the page.
q: EventQueue[tuple[int, int], str] = EventQueue()
q.insert((0, 0), "a")
q.insert((0, 10), "b")
q.insert((1, 5), "c")
q.insert((10, 10))
q.insert((10, 0))
q.insert((3, 5))
q.check_invariant()
assert len(q) == 6
assert q.pop_min() == ((0, 0), ["a"])
assert q.pop_min() == ((0, 10), ["b"])
q.insert((5, 5)) # a and b are neighbors and cross at (5, 5)
assert len(q) == 5
assert q.pop_min() == ((1, 5), ["c"])
assert q.pop_min() == ((3, 5), [])
assert q.contains((5, 5))
q.insert((5, 5)) # a and b are neighbors again: the same event
assert len(q) == 3
q.check_invariant()
popped = []
while not q.is_empty():
popped.append(q.pop_min())
assert popped == [((5, 5), []), ((10, 0), []), ((10, 10), [])]
def test_merge_payloads() -> None:
q: EventQueue[tuple[int, int], str] = EventQueue()
q.insert((4, 4), "s1 ends")
q.insert((2, 9), "x")
q.insert((4, 4), "s3 starts")
q.insert((4, 4))
assert len(q) == 2
assert q.pop_min() == ((2, 9), ["x"])
assert q.pop_min() == ((4, 4), ["s1 ends", "s3 starts"])
assert q.is_empty()
def test_empty_and_one() -> None:
q: EventQueue[tuple[int, int], int] = EventQueue()
assert q.is_empty() and len(q) == 0
assert not q.contains((0, 0))
with pytest.raises(IndexError):
q.pop_min()
q.insert((-(2**63), 2**63 - 1), 7)
q.check_invariant()
assert q.pop_min() == ((-(2**63), 2**63 - 1), [7])
assert q.is_empty()
def test_lexicographic_order() -> None:
q: EventQueue[tuple[int, int], int] = EventQueue()
for p in [(1, 5), (1, -3), (0, 100), (1, 0)]:
q.insert(p, 0)
assert [q.pop_min()[0] for _ in range(4)] == [(0, 100), (1, -3), (1, 0), (1, 5)]
def test_random_against_model() -> None:
rng = random.Random(324)
for _ in range(600):
q: EventQueue[tuple[int, int], int] = EventQueue()
model: dict[tuple[int, int], list[int]] = {}
for step in range(rng.randrange(1, 60)):
op = rng.random()
if op < 0.6:
p = (rng.randrange(5), rng.randrange(5))
a = rng.choice([None, step])
q.insert(p, a)
model.setdefault(p, [])
if a is not None:
model[p].append(a)
elif model:
p = min(model)
assert q.pop_min() == (p, model.pop(p))
else:
assert q.is_empty()
q.check_invariant()
assert len(q) == len(model)
assert q.is_empty() == (not model)
probe = (rng.randrange(5), rng.randrange(5))
assert q.contains(probe) == (probe in model)
What the next level needs
- the tests did not run in this build