Algorithm Search⌘K

Mo's algorithm in Python

Sorts offline queries by blocks and moves a window across the array in O(n sqrt q) total.

File
mo_algorithm.py, 83 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
mo_algorithm.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/mo-algorithm/. The authors are listed in prov/S9FYEP.jsonl.

import math
from typing import Callable, Hashable, Sequence, TypeVar

T = TypeVar("T")


def default_block(n: int, q: int) -> int:
    """The block size max(1, min(n, ceil(n / sqrt(q))))."""
    if n <= 0 or q <= 0:
        return 1
    s = math.isqrt(q)
    if s * s < q:
        s += 1
    return max(1, min(n, -(-n // s)))


def mo_algorithm(
    n: int,
    queries: Sequence[tuple[int, int]],
    add: Callable[[int], None],
    remove: Callable[[int], None],
    answer: Callable[[], T],
    block: int = 0,
) -> list[T]:
    """Answer offline window queries [l, r) over positions 0..n-1.

    Precondition: 0 <= l <= r <= n for each query; add(i) puts position i
    into the window, remove(i) takes it out, and answer() reads the value
    of the current window. block <= 0 picks ceil(n / sqrt(q)).
    Cost: O(n sqrt(q)) calls of add and remove, plus O(q log q) to sort.
    """
    q = len(queries)
    b = block if block > 0 else default_block(n, q)
    order = sorted(range(q), key=lambda j: (queries[j][0] // b, queries[j][1]))  # @line L1
    cur_l = cur_r = 0  # @line L2
    ans: list[T] = [None] * q  # type: ignore[list-item]
    for j in order:  # @line L3
        l, r = queries[j]
        while cur_l > l:  # @line L4
            cur_l -= 1
            add(cur_l)
        while cur_r < r:  # @line L5
            add(cur_r)
            cur_r += 1
        while cur_l < l:  # @line L6
            remove(cur_l)
            cur_l += 1
        while cur_r > r:  # @line L7
            cur_r -= 1
            remove(cur_r)
        ans[j] = answer()  # @line L8
    return ans  # @line L9


def distinct_counts(a: Sequence[Hashable], queries: Sequence[tuple[int, int]], block: int = 0) -> list[int]:
    """The number of distinct values of a[l:r] for each query (l, r).

    Precondition: 0 <= l <= r <= len(a). Cost: O(n sqrt(q) + q log q)
    expected time with a hash map of counts, O(n + q) extra space.
    """
    cnt: dict[Hashable, int] = {}
    d = 0

    def add(i: int) -> None:
        nonlocal d
        c = cnt.get(a[i], 0)
        if c == 0:
            d += 1
        cnt[a[i]] = c + 1

    def remove(i: int) -> None:
        nonlocal d
        c = cnt[a[i]] - 1
        if c < 0:
            raise ValueError("remove of an item that is not in the window")
        if c == 0:
            d -= 1
        cnt[a[i]] = c

    return mo_algorithm(len(a), queries, add, remove, lambda: d, block)
test_mo_algorithm.py 73 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/mo-algorithm/. The authors are listed in prov/S9FYEP.jsonl.

import random

from mo_algorithm import default_block, distinct_counts, mo_algorithm


def test_example() -> None:
    a = [1, 2, 1, 3, 2, 2, 4, 1]
    qs = [(0, 4), (2, 7), (1, 3), (5, 8), (3, 6)]
    moves = []
    window: list[int] = []

    def add(i: int) -> None:
        moves.append(i)
        window.append(i)

    def remove(i: int) -> None:
        moves.append(i)
        window.remove(i)

    ans = mo_algorithm(8, qs, add, remove, lambda: len({a[i] for i in window}), 3)
    assert ans == [3, 4, 2, 3, 2]
    assert len(moves) == 17
    assert distinct_counts(a, qs, 3) == [3, 4, 2, 3, 2]


def test_edges() -> None:
    assert distinct_counts([], []) == []
    assert distinct_counts([], [(0, 0)]) == [0]
    assert distinct_counts([7], [(0, 1), (0, 0), (1, 1)]) == [1, 0, 0]
    assert distinct_counts([5, 5, 5], [(0, 3), (2, 3), (1, 1)]) == [1, 1, 0]
    assert default_block(0, 0) == 1
    assert default_block(100, 10000) == 1
    assert default_block(100, 4) == 50
    assert default_block(100, 1) == 100


def test_window_never_inverts() -> None:
    rng = random.Random(7)
    for _ in range(200):
        n = rng.randint(0, 30)
        qs = []
        for _ in range(rng.randint(0, 20)):
            l = rng.randint(0, n)
            qs.append((l, rng.randint(l, n)))
        inside: set[int] = set()

        def add(i: int) -> None:
            assert i not in inside
            inside.add(i)

        def remove(i: int) -> None:
            assert i in inside
            inside.remove(i)

        got = mo_algorithm(n, qs, add, remove, lambda: frozenset(inside), rng.randint(0, 6))
        assert got == [frozenset(range(l, r)) for l, r in qs]


def test_random_against_brute_force() -> None:
    rng = random.Random(2024)
    for _ in range(600):
        n = rng.randint(0, 40)
        a = [rng.randint(0, 6) for _ in range(n)]
        qs = []
        for _ in range(rng.randint(0, 25)):
            l = rng.randint(0, n)
            qs.append((l, rng.randint(l, n)))
        expected = [len(set(a[l:r])) for l, r in qs]
        assert distinct_counts(a, qs) == expected
        assert distinct_counts(a, qs, rng.randint(1, 8)) == expected

What the next level needs

  • the tests did not run in this build

Read the full page about Mo's algorithm