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
# 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