Algorithm Search⌘K

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
min_queue.pyRawGitHub
# 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

Read the full page about Queue with minimum