Segment tree beats in Python
Segment tree variant for range chmin, chmax and modulo updates with amortized logarithmic bounds.
- File
segment_tree_beats.py, 132 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/segment-tree-beats/. The authors are listed in prov/4S5W6M.jsonl.
from collections.abc import Sequence
NEG = float("-inf")
class SegmentTreeBeats:
"""Segment tree beats for range chmin, range sum and range max on integers.
Node 1 covers A[0 .. n-1]; node v with range [lo, hi] has the children 2v
and 2v + 1. Each node keeps the maximum mx1, the strict second maximum mx2
(NEG if all values of the range are equal), the number cnt of items equal
to mx1, and the sum. A pending chmin at a node is mx1 itself: the true
value of an item is the minimum of mx1 over the nodes on its root path.
chmin costs O(log n) amortized; the queries cost O(log n).
"""
def __init__(self, values: Sequence[int]) -> None:
self.n = len(values)
if self.n == 0:
raise ValueError("the array must not be empty")
size = 4 * self.n
self.mx1: list[float] = [NEG] * size
self.mx2: list[float] = [NEG] * size
self.cnt = [0] * size
self.sum = [0] * size
self.visits = 0 # nodes entered by chmin, for the amortized bound
self._build(1, 0, self.n - 1, values)
def _build(self, v: int, lo: int, hi: int, a: Sequence[int]) -> None:
if lo == hi: # @line L1
self.mx1[v], self.mx2[v], self.cnt[v], self.sum[v] = a[lo], NEG, 1, a[lo]
return
mid = (lo + hi) // 2 # @line L2
self._build(2 * v, lo, mid, a)
self._build(2 * v + 1, mid + 1, hi, a)
self._pull(v) # @line L3
def _pull(self, v: int) -> None:
a, b = 2 * v, 2 * v + 1 # @line L4
mx1, mx2, cnt = self.mx1, self.mx2, self.cnt
self.sum[v] = self.sum[a] + self.sum[b] # @line L5
if mx1[a] == mx1[b]: # @line L6
mx1[v], cnt[v], mx2[v] = mx1[a], cnt[a] + cnt[b], max(mx2[a], mx2[b])
elif mx1[a] > mx1[b]: # @line L7
mx1[v], cnt[v], mx2[v] = mx1[a], cnt[a], max(mx2[a], mx1[b])
else: # @line L8
mx1[v], cnt[v], mx2[v] = mx1[b], cnt[b], max(mx1[a], mx2[b])
def _apply(self, v: int, x: int) -> None:
if x < self.mx1[v]: # @line L9
self.sum[v] -= self.cnt[v] * (self.mx1[v] - x) # type: ignore[assignment]
self.mx1[v] = x
def _push(self, v: int) -> None:
self._apply(2 * v, self.mx1[v]) # type: ignore[arg-type] # @line L10
self._apply(2 * v + 1, self.mx1[v]) # type: ignore[arg-type]
def _chmin(self, v: int, lo: int, hi: int, l: int, r: int, x: int) -> None:
self.visits += 1
if r < lo or hi < l or self.mx1[v] <= x: # @line L11
return
if l <= lo and hi <= r and self.mx2[v] < x: # @line L12
self._apply(v, x)
return
self._push(v) # @line L13
mid = (lo + hi) // 2
self._chmin(2 * v, lo, mid, l, r, x) # @line L14
self._chmin(2 * v + 1, mid + 1, hi, l, r, x)
self._pull(v) # @line L15
def chmin(self, l: int, r: int, x: int) -> None:
"""Set A[i] = min(A[i], x) for l <= i <= r."""
self._check(l, r)
self._chmin(1, 0, self.n - 1, l, r, x)
def _query(self, v: int, lo: int, hi: int, l: int, r: int, want_sum: bool) -> float:
if r < lo or hi < l: # @line L16
return 0 if want_sum else NEG
if l <= lo and hi <= r: # @line L17
return self.sum[v] if want_sum else self.mx1[v]
self._push(v) # @line L18
mid = (lo + hi) // 2
a = self._query(2 * v, lo, mid, l, r, want_sum)
b = self._query(2 * v + 1, mid + 1, hi, l, r, want_sum)
return a + b if want_sum else max(a, b)
def query_sum(self, l: int, r: int) -> int:
"""Return A[l] + ... + A[r]."""
self._check(l, r)
return int(self._query(1, 0, self.n - 1, l, r, True))
def query_max(self, l: int, r: int) -> int:
"""Return max(A[l], ..., A[r])."""
self._check(l, r)
return int(self._query(1, 0, self.n - 1, l, r, False))
def values(self) -> list[int]:
"""Return the current array, in O(n log n) time."""
return [self.query_max(i, i) for i in range(self.n)]
def _check(self, l: int, r: int) -> None:
if not (0 <= l <= r < self.n):
raise IndexError("range out of bounds")
def check_invariant(self) -> None:
"""Assert that every node describes the capped values of its range.
The value of item i seen at node v is the minimum of the leaf value
and mx1 of the nodes from the leaf up to v. The fields of v must be
the max, strict second max, count of the max and sum of these values,
and the values seen at the root must be the array.
"""
def walk(v: int, lo: int, hi: int) -> list[float]:
if lo == hi:
seen = [self.mx1[v]]
else:
mid = (lo + hi) // 2
seen = walk(2 * v, lo, mid) + walk(2 * v + 1, mid + 1, hi)
seen = [min(s, self.mx1[v]) for s in seen]
top = max(seen)
rest = [s for s in seen if s != top]
assert self.mx1[v] == top, v
assert self.mx2[v] == (max(rest) if rest else NEG), v
assert self.cnt[v] == len(seen) - len(rest), v
assert self.sum[v] == sum(seen), v
return seen
assert walk(1, 0, self.n - 1) == self.values()
test_segment_tree_beats.py 83 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/segment-tree-beats/. The authors are listed in prov/4S5W6M.jsonl.
import math
import random
import pytest
from segment_tree_beats import NEG, SegmentTreeBeats
def test_example() -> None:
st = SegmentTreeBeats([5, 2, 8, 1, 9, 3, 7, 4])
assert (st.mx1[1], st.mx2[1], st.cnt[1], st.sum[1]) == (9, 8, 1, 39)
st.visits = 0
st.chmin(0, 7, 6)
assert st.visits == 5
assert (st.mx1[1], st.mx2[1], st.cnt[1], st.sum[1]) == (6, 5, 3, 33)
assert (st.mx1[2], st.sum[2]) == (6, 14)
assert st.mx1[5] == 8 # pending: node 5 still stores the old maximum
assert st.query_sum(2, 5) == 16
assert st.mx1[5] == 6 # the query pushed the cap down
assert st.values() == [5, 2, 6, 1, 6, 3, 6, 4]
st.chmin(1, 6, 3)
assert st.values() == [5, 2, 3, 1, 3, 3, 3, 4]
assert st.query_sum(0, 7) == 24
assert st.query_max(1, 6) == 3
st.check_invariant()
def test_single_and_equal() -> None:
st = SegmentTreeBeats([4])
assert st.mx2[1] == NEG
st.chmin(0, 0, 10)
assert st.query_sum(0, 0) == 4
st.chmin(0, 0, -2)
assert st.query_max(0, 0) == -2
eq = SegmentTreeBeats([7] * 5)
eq.chmin(0, 4, 3)
assert eq.query_sum(0, 4) == 15
eq.check_invariant()
with pytest.raises(IndexError):
eq.chmin(2, 5, 0)
with pytest.raises(ValueError):
SegmentTreeBeats([])
def test_random_against_brute_force() -> None:
rng = random.Random(4565)
for _ in range(40):
n = rng.randint(1, 40)
a = [rng.randint(-50, 50) for _ in range(n)]
st = SegmentTreeBeats(a)
for _ in range(40):
l = rng.randrange(n)
r = rng.randrange(l, n)
kind = rng.randrange(3)
if kind == 0:
x = rng.randint(-60, 60)
st.chmin(l, r, x)
for i in range(l, r + 1):
a[i] = min(a[i], x)
elif kind == 1:
assert st.query_sum(l, r) == sum(a[l : r + 1])
else:
assert st.query_max(l, r) == max(a[l : r + 1])
st.check_invariant()
assert st.values() == a
def test_amortized_visit_bound() -> None:
rng = random.Random(99)
for n in (64, 200, 513):
q = 2 * n
st = SegmentTreeBeats([rng.randint(0, 10**6) for _ in range(n)])
st.visits = 0
for _ in range(q):
l = rng.randrange(n)
r = rng.randrange(l, n)
st.chmin(l, r, rng.randint(0, 10**6))
lg = math.ceil(math.log2(n))
extra = n * (lg + 1) + q * (2 * lg + 1)
assert st.visits <= q * (4 * lg + 1) + 2 * extra
What the next level needs
- the tests did not run in this build