Multi-vantage-point tree in Python
A vantage-point tree that uses several vantage points per node to reduce distance computations.
- File
mvp_tree.py, 155 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/mvp-tree/. The authors are listed in prov/WR0RGE.jsonl.
import random
from dataclasses import dataclass, field
from typing import Callable, Generic, Optional, Sequence, TypeVar
T = TypeVar("T")
Metric = Callable[[T, T], float]
@dataclass
class Entry(Generic[T]):
obj: T
path: list[float] # distances to the first p vantage points above it
d1: float = 0.0
d2: float = 0.0
@dataclass
class Leaf(Generic[T]):
v1: Optional[T]
v2: Optional[T]
entries: list[Entry[T]]
@dataclass
class Internal(Generic[T]):
v1: T
v2: Optional[T]
# children[i][j] with the interval of d(., v1) of group i and of d(., v2) of subgroup j
ring1: list[tuple[float, float]]
ring2: list[list[tuple[float, float]]]
children: list[list[object]] = field(default_factory=list)
def _split(entries: list[Entry[T]], key: Callable[[Entry[T]], float], m: int) -> list[list[Entry[T]]]:
"""Split by rank of key into m groups of sizes as equal as possible (some may be empty)."""
s = sorted(entries, key=key)
n = len(s)
return [s[(i * n) // m : ((i + 1) * n) // m] for i in range(m)]
def _interval(group: list[Entry[T]], key: Callable[[Entry[T]], float]) -> tuple[float, float]:
if not group:
return (1.0, 0.0) # empty interval, never met
vals = [key(e) for e in group]
return (min(vals), max(vals))
class MVPTree(Generic[T]):
"""A multi-vantage-point tree with fan-out m*m, leaf capacity k and p path distances per object.
choose(n) gives the index of the first vantage point among n candidates; the default is random.
The second vantage point is the object farthest from the first. evaluations counts metric calls.
"""
def __init__(self, points: Sequence[T], d: Metric, m: int = 2, k: int = 4, p: int = 4, seed: int = 0,
choose: Optional[Callable[[int], int]] = None) -> None:
if m < 2 or k < 1 or p < 0:
raise ValueError("need m >= 2, k >= 1, p >= 0")
self._d, self.m, self.k, self.p = d, m, k, p
self.evaluations = 0
rng = random.Random(seed)
self._choose = choose or rng.randrange
self.root = self._build([Entry(x, []) for x in points])
def dist(self, a: T, b: T) -> float:
self.evaluations += 1
return self._d(a, b)
def _push(self, e: Entry[T], value: float) -> None:
if len(e.path) < self.p:
e.path.append(value)
def _vantage_points(self, s: list[Entry[T]]) -> tuple[Optional[Entry[T]], Optional[Entry[T]], list[Entry[T]]]:
"""v1 at random; d1 for the others; v2 the farthest from v1; d2 for the rest."""
if not s:
return None, None, []
v1 = s.pop(self._choose(len(s)))
for e in s:
e.d1 = self.dist(e.obj, v1.obj)
if not s:
return v1, None, []
j = max(range(len(s)), key=lambda i: s[i].d1)
v2 = s.pop(j)
for e in s:
e.d2 = self.dist(e.obj, v2.obj)
return v1, v2, s
def _build(self, s: list[Entry[T]]):
if len(s) <= self.k + 2: # @line L1
v1, v2, rest = self._vantage_points(s) # @line L2
return Leaf(v1.obj if v1 else None, v2.obj if v2 else None, rest)
v1 = s.pop(self._choose(len(s))) # @line L3
for e in s:
e.d1 = self.dist(e.obj, v1.obj)
self._push(e, e.d1)
groups = _split(s, lambda e: e.d1, self.m) # @line L4
ring1 = [_interval(g, lambda e: e.d1) for g in groups]
gi = max(range(self.m), key=lambda i: len(groups[i]))
j = max(range(len(groups[gi])), key=lambda i: groups[gi][i].d1)
v2 = groups[gi].pop(j) # @line L5
ring2: list[list[tuple[float, float]]] = []
children: list[list[object]] = []
for g in groups:
for e in g:
e.d2 = self.dist(e.obj, v2.obj)
self._push(e, e.d2)
sub = _split(g, lambda e: e.d2, self.m) # @line L6
ring2.append([_interval(h, lambda e: e.d2) for h in sub])
children.append([self._build(h) for h in sub]) # @line L7
return Internal(v1.obj, v2.obj, ring1, ring2, children)
def range(self, q: T, r: float) -> list[T]:
"""All stored objects x with d(q, x) <= r."""
out: list[T] = []
self._range(self.root, q, r, [], out)
return out
def _range(self, node, q: T, r: float, qpath: list[float], out: list[T]) -> None:
if isinstance(node, Leaf): # @line L8
if node.v1 is None:
return
t1 = self.dist(q, node.v1) # @line L9
if t1 <= r:
out.append(node.v1)
if node.v2 is None:
return
t2 = self.dist(q, node.v2)
if t2 <= r:
out.append(node.v2)
for e in node.entries: # @line L10
if abs(t1 - e.d1) > r or abs(t2 - e.d2) > r: # @line L11
continue
if any(abs(a - b) > r for a, b in zip(qpath, e.path)): # @line L12
continue
if self.dist(q, e.obj) <= r: # @line L13
out.append(e.obj)
return
t1 = self.dist(q, node.v1) # @line L14
t2 = self.dist(q, node.v2)
for t, v in ((t1, node.v1), (t2, node.v2)):
if t <= r:
out.append(v)
path = (qpath + [t1, t2])[: self.p] # @line L15
for i in range(self.m): # @line L16
lo, hi = node.ring1[i]
if lo > hi or t1 + r < lo or t1 - r > hi:
continue
for j in range(self.m): # @line L17
lo2, hi2 = node.ring2[i][j]
if lo2 > hi2 or t2 + r < lo2 or t2 - r > hi2:
continue
self._range(node.children[i][j], q, r, path, out) # @line L18
test_mvp_tree.py 74 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/mvp-tree/. The authors are listed in prov/WR0RGE.jsonl.
import math
import random
from mvp_tree import Internal, Leaf, MVPTree
P = [(1, 1), (2, 6), (4, 4), (5, 1), (6, 7), (8, 3), (9, 8), (3, 9), (7, 5), (0, 4), (10, 2), (6, 2)]
N = dict(zip("abcdefghijkl", P))
def depth(node) -> int:
if isinstance(node, Leaf):
return 1
return 1 + max(depth(c) for row in node.children for c in row)
def edit(a: str, b: str) -> int:
prev = list(range(len(b) + 1))
for i, ca in enumerate(a, 1):
cur = [i]
for j, cb in enumerate(b, 1):
cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb)))
prev = cur
return prev[-1]
def test_example() -> None:
t = MVPTree(P, math.dist, m=2, k=2, p=2, choose=lambda n: 0)
assert t.evaluations == 29
root = t.root
assert isinstance(root, Internal) and root.v1 == N["a"] and root.v2 == N["g"]
leaf = root.children[0][1]
assert {leaf.v1, leaf.v2} == {N["b"], N["d"]}
(j,) = leaf.entries
assert j.obj == N["j"] and [round(x, 4) for x in j.path] == [3.1623, 9.8489]
t.evaluations = 0
assert sorted(t.range((5, 5), 2)) == sorted([N["c"], N["i"]])
assert t.evaluations == 8
def test_edges() -> None:
assert MVPTree([], math.dist).range((0, 0), 1) == []
assert MVPTree([(1, 1)], math.dist).range((0, 0), 2) == [(1, 1)]
same = MVPTree([(2, 2)] * 20, math.dist, k=1, p=3)
assert len(same.range((2, 2), 0)) == 20
def test_height() -> None:
for n in [1, 5, 16, 64, 300]:
for m in [2, 3]:
t = MVPTree([(i, 0) for i in range(n)], math.dist, m=m, k=1, p=2, seed=n)
assert depth(t.root) <= math.ceil(math.log(max(n, 2), m * m)) + 1
assert t.evaluations <= 2 * n * depth(t.root)
def test_random_against_scan() -> None:
rng = random.Random(1997)
for trial in range(500):
m, k, p = rng.choice([2, 3]), rng.randint(1, 5), rng.randint(0, 6)
if trial % 2:
dim = rng.choice([1, 2, 3])
pts = [tuple(rng.randint(0, 9) for _ in range(dim)) for _ in range(rng.randint(0, 60))]
d = math.dist
q = tuple(rng.uniform(-1, 10) for _ in range(dim))
r = rng.uniform(0, 5)
else:
pts = ["".join(rng.choice("abc") for _ in range(rng.randint(0, 5))) for _ in range(rng.randint(0, 40))]
d = edit
q = "".join(rng.choice("abc") for _ in range(rng.randint(0, 5)))
r = rng.randint(0, 3)
t = MVPTree(pts, d, m=m, k=k, p=p, seed=trial)
assert sorted(t.range(q, r)) == sorted(x for x in pts if d(q, x) <= r)
What the next level needs
- the tests did not run in this build