Queue with minimum in Python
A queue that returns its current minimum in constant amortized time, built from two minimum stacks or a monotonic deque.
- File
min_queue.py, 109 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/min-queue/. The authors are listed in prov/VVXG5M.jsonl.
from collections import deque
from typing import Generic, Protocol, TypeVar
class Ordered(Protocol):
def __lt__(self, other: "Ordered", /) -> bool: ...
T = TypeVar("T", bound=Ordered)
class _MinStack(Generic[T]):
"""A stack of pairs (x, least item at or below x)."""
def __init__(self) -> None:
self.s: list[tuple[T, T]] = []
def __len__(self) -> int:
return len(self.s)
def push(self, x: T) -> None:
m = x if not self.s or x < self.s[-1][1] else self.s[-1][1]
self.s.append((x, m))
def pop(self) -> T:
return self.s.pop()[0]
def min(self) -> T:
return self.s[-1][1]
class MinQueue(Generic[T]):
"""A FIFO queue with a min query, built from two stacks with minimum.
The front of the queue is the top of `out`; the back is the top of
`inp`. enqueue, dequeue and min cost O(1) amortized: each item is pushed
and popped at most twice. dequeue costs O(n) in the worst case.
"""
def __init__(self) -> None:
self.inp: _MinStack[T] = _MinStack()
self.out: _MinStack[T] = _MinStack()
self.moves = 0 # items moved by line L4, for tests
def __len__(self) -> int:
return len(self.inp) + len(self.out)
def enqueue(self, x: T) -> None:
"""Add x at the back. Cost O(1)."""
self.inp.push(x) # @line L1
def dequeue(self) -> T:
"""Remove and return the front item. Precondition: not empty. Cost O(1) amortized."""
if len(self) == 0:
raise IndexError("dequeue from an empty queue")
if len(self.out) == 0: # @line L2
while len(self.inp) > 0: # @line L3
self.out.push(self.inp.pop()) # @line L4
self.moves += 1
return self.out.pop() # @line L5
def min(self) -> T:
"""Return the least item. Precondition: not empty. Cost O(1)."""
if len(self) == 0:
raise IndexError("min of an empty queue")
if len(self.inp) == 0: # @line L6
return self.out.min()
if len(self.out) == 0: # @line L7
return self.inp.min()
a, b = self.inp.min(), self.out.min() # @line L8
return b if b < a else a
def to_list(self) -> list[T]:
"""The items from the front to the back."""
return [x for x, _ in reversed(self.out.s)] + [x for x, _ in self.inp.s]
class MonotonicMinQueue(Generic[T]):
"""The deque variant: keep only the items that can still become the minimum.
Each entry is (x, t) with t the arrival number. The values in the deque
increase strictly from the front, so the front is the minimum.
enqueue costs O(1) amortized; dequeue and min cost O(1).
"""
def __init__(self) -> None:
self.d: deque[tuple[T, int]] = deque()
self.items: deque[T] = deque() # the full queue, to return the dequeued item
self.arrived = 0
self.left = 0
def enqueue(self, x: T) -> None:
while self.d and not self.d[-1][0] < x: # drop back items >= x
self.d.pop()
self.d.append((x, self.arrived))
self.items.append(x)
self.arrived += 1
def dequeue(self) -> T:
if self.d and self.d[0][1] == self.left:
self.d.popleft()
self.left += 1
return self.items.popleft()
def min(self) -> T:
return self.d[0][0]
test_min_queue.py 87 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/min-queue/. The authors are listed in prov/VVXG5M.jsonl.
import random
import pytest
from min_queue import MinQueue, MonotonicMinQueue
def test_page_example() -> None:
q: MinQueue[int] = MinQueue()
for x in [4, 2, 5]:
q.enqueue(x)
assert q.min() == 2
assert q.inp.s == [(4, 4), (2, 2), (5, 2)]
assert q.dequeue() == 4
assert q.moves == 3
assert q.out.s == [(5, 5), (2, 2)]
assert q.min() == 2
q.enqueue(1)
assert q.min() == 1
assert q.dequeue() == 2
assert q.min() == 1
assert q.dequeue() == 5
assert q.min() == 1
assert q.dequeue() == 1
assert q.moves == 4
assert len(q) == 0
def test_edges() -> None:
q: MinQueue[int] = MinQueue()
with pytest.raises(IndexError):
q.dequeue()
with pytest.raises(IndexError):
q.min()
for x in [3, 3, 3]:
q.enqueue(x)
assert q.dequeue() == 3 and q.min() == 3
def test_amortized_moves() -> None:
q: MinQueue[int] = MinQueue()
rng = random.Random(5)
enq = 0
for _ in range(10000):
if len(q) and rng.random() < 0.5:
q.dequeue()
else:
q.enqueue(rng.randint(0, 100))
enq += 1
assert q.moves <= enq
def test_sliding_window_minimum() -> None:
a = [5, 1, 4, 2, 6, 3, 0, 7]
q: MinQueue[int] = MinQueue()
out = []
for i, x in enumerate(a):
q.enqueue(x)
if i >= 3:
q.dequeue()
if i >= 2:
out.append(q.min())
assert out == [min(a[i : i + 3]) for i in range(len(a) - 2)]
def test_random_against_model() -> None:
rng = random.Random(2024)
for _ in range(500):
q: MinQueue[int] = MinQueue()
m: MonotonicMinQueue[int] = MonotonicMinQueue()
model: list[int] = []
for _ in range(60):
if model and rng.random() < 0.45:
want = model.pop(0)
assert q.dequeue() == want
assert m.dequeue() == want
else:
x = rng.randint(0, 6)
q.enqueue(x)
m.enqueue(x)
model.append(x)
assert q.to_list() == model
if model:
assert q.min() == min(model) == m.min()
What the next level needs
- the tests did not run in this build