k-d tree in Python
Binary tree that splits k-dimensional points by cycling through the coordinate axes at each level.
- File
kd_tree.py, 262 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/kd-tree/. The authors are listed in prov/39PCSM.jsonl.
import math
import random
from typing import Optional, Sequence
Point = tuple[float, ...]
def superkey(p: Sequence[float], a: int) -> tuple[float, ...]:
"""The key of p for axis a: (p[a], p[a+1], ..., p[a-1]), compared lexicographically."""
return tuple(p[a:]) + tuple(p[:a])
class Node:
__slots__ = ("point", "axis", "left", "right")
def __init__(self, point: Point, axis: int) -> None:
self.point = point
self.axis = axis
self.left: Optional[Node] = None
self.right: Optional[Node] = None
def _select(items: list[Point], r: int, a: int, rng: random.Random) -> Point:
"""Return the item of rank r (0-based) in the superkey order of axis a.
Quickselect with a random pivot and a three-way partition: expected O(len(items)).
"""
xs = items
while True:
pivot = superkey(xs[rng.randrange(len(xs))], a)
lo = [x for x in xs if superkey(x, a) < pivot]
eq = [x for x in xs if superkey(x, a) == pivot]
if r < len(lo):
xs = lo
elif r < len(lo) + len(eq):
return eq[0]
else:
r -= len(lo) + len(eq)
xs = [x for x in xs if superkey(x, a) > pivot]
class KdTree:
"""A k-d tree with one point in each node (Bentley's form).
Invariant: at a node t with axis a, each point of t.left is less than
t.point in the superkey order of axis a, and each point of t.right is
greater than or equal to it. The axis of a node is its depth mod k.
Duplicate points are allowed; they go to the right.
"""
def __init__(self, k: int, points: Sequence[Sequence[float]] = ()) -> None:
"""Build a tree on the points by median splits. Expected O(n log n) time."""
if k < 1:
raise ValueError("k must be at least 1")
self.k = k
self.root: Optional[Node] = None
self.size = 0
pts = [tuple(float(c) for c in p) for p in points]
for p in pts:
if len(p) != k:
raise ValueError("a point has the wrong dimension")
self.size = len(pts)
self.root = self._build(pts)
def _build(self, pts: list[Point]) -> Optional[Node]:
rng = random.Random(len(pts))
root: Optional[Node] = None
# each entry: (points, depth, parent, is_left); an explicit stack keeps
# the recursion of BUILD off the Python call stack
stack: list[tuple[list[Point], int, Optional[Node], bool]] = [(pts, 0, None, False)]
while stack:
p_set, depth, parent, is_left = stack.pop()
if not p_set: # @line L1
continue
a = depth % self.k # @line L2
m = _select(p_set, len(p_set) // 2, a, rng) # @line L3
t = Node(m, a) # @line L4
if parent is None:
root = t
elif is_left:
parent.left = t
else:
parent.right = t
km = superkey(m, a)
left = [p for p in p_set if superkey(p, a) < km] # @line L5
right = [p for p in p_set if superkey(p, a) >= km] # @line L6
right.remove(m) # one copy of m is the node itself
stack.append((left, depth + 1, t, True))
stack.append((right, depth + 1, t, False))
return root # @line L7
def insert(self, p: Sequence[float]) -> None:
"""Add the point p at a new leaf. O(h) time, h the height."""
q = tuple(float(c) for c in p)
if len(q) != self.k:
raise ValueError("the point has the wrong dimension")
self.size += 1
if self.root is None: # @line L13
self.root = Node(q, 0)
return
t = self.root
while True: # @line L14
a = t.axis
if superkey(q, a) < superkey(t.point, a): # @line L15
if t.left is None:
t.left = Node(q, (a + 1) % self.k)
return
t = t.left
else:
if t.right is None: # @line L16
t.right = Node(q, (a + 1) % self.k)
return
t = t.right
def contains(self, p: Sequence[float]) -> bool:
"""Return True when p is in the tree. O(h) time."""
q = tuple(float(c) for c in p)
t = self.root
while t is not None:
if t.point == q:
return True
t = t.left if superkey(q, t.axis) < superkey(t.point, t.axis) else t.right
return False
def range_search(self, lo: Sequence[float], hi: Sequence[float]) -> list[Point]:
"""Return the points p with lo[i] <= p[i] <= hi[i] for each axis i.
O(n^(1-1/k) + r) nodes for a tree built by the constructor on points
with distinct coordinates, r the output size; O(n) in any case.
"""
out: list[Point] = []
stack = [self.root]
while stack:
t = stack.pop()
if t is None: # @line L8
continue
if all(lo[i] <= t.point[i] <= hi[i] for i in range(self.k)): # @line L9
out.append(t.point)
a = t.axis # @line L10
if lo[a] <= t.point[a]: # @line L11
stack.append(t.left)
if hi[a] >= t.point[a]: # @line L12
stack.append(t.right)
return out
def nearest(self, q: Sequence[float]) -> Optional[Point]:
"""Return a point of the tree at the least Euclidean distance from q, or None.
Branch and bound; each subtree carries a lower bound on the squared
distance from q to its points.
"""
best: Optional[Point] = None
best_d = math.inf
stack: list[tuple[Optional[Node], float]] = [(self.root, 0.0)]
while stack:
t, bound = stack.pop()
if t is None or bound >= best_d: # @line L17
continue
d = sum((q[i] - t.point[i]) ** 2 for i in range(self.k))
if d < best_d: # @line L18
best, best_d = t.point, d
a = t.axis # @line L19
delta = q[a] - t.point[a]
near, far = (t.left, t.right) if delta < 0 else (t.right, t.left) # @line L20
stack.append((far, max(bound, delta * delta))) # @line L22: searched only if delta^2 < best
stack.append((near, bound)) # @line L21
return best
def _find_min(self, t: Node, a: int) -> tuple[Node, Optional[Node]]:
"""Return (node, parent) of a node of least superkey for axis a in the subtree t."""
best, best_parent = t, None
stack: list[tuple[Node, Optional[Node]]] = [(t, None)]
while stack:
u, par = stack.pop()
if superkey(u.point, a) < superkey(best.point, a):
best, best_parent = u, par
if u.left is not None:
stack.append((u.left, u))
if u.right is not None and u.axis != a: # @line L27: on axis a, the right side is not smaller
stack.append((u.right, u))
if best_parent is None and best is not t:
raise AssertionError("unreachable")
return best, best_parent
def delete(self, p: Sequence[float]) -> bool:
"""Remove one copy of p. Return False when p is not in the tree. O(h) to find p,
plus the cost of the replacement searches."""
q = tuple(float(c) for c in p)
parent: Optional[Node] = None
t = self.root
while t is not None and t.point != q: # @line L23
parent = t
t = t.left if superkey(q, t.axis) < superkey(t.point, t.axis) else t.right
if t is None:
return False
self.size -= 1
while True:
a = t.axis
if t.right is not None: # @line L24
m, mp = self._find_min(t.right, a)
t.point = m.point
parent, t = (mp if mp is not None else t), m
elif t.left is not None: # @line L25
m, mp = self._find_min(t.left, a)
t.point = m.point
t.right, t.left = t.left, None
parent, t = (mp if mp is not None else t), m
else: # @line L26: a leaf is cut off
if parent is None:
self.root = None
elif parent.left is t:
parent.left = None
else:
parent.right = None
return True
def height(self) -> int:
"""The number of nodes on a longest root-to-leaf path (0 for an empty tree)."""
h = 0
stack = [(self.root, 1)]
while stack:
t, d = stack.pop()
if t is not None:
h = max(h, d)
stack.append((t.left, d + 1))
stack.append((t.right, d + 1))
return h
def points(self) -> list[Point]:
out: list[Point] = []
stack = [self.root]
while stack:
t = stack.pop()
if t is not None:
out.append(t.point)
stack.append(t.left)
stack.append(t.right)
return out
def check_invariant(self) -> None:
"""Raise AssertionError if the axes or the order of some node are wrong."""
count = 0
# each entry: node, depth, list of (axis, key, side) constraints from ancestors
stack: list[tuple[Optional[Node], int, list[tuple[int, tuple[float, ...], bool]]]] = [(self.root, 0, [])]
while stack:
t, depth, cons = stack.pop()
if t is None:
continue
count += 1
assert t.axis == depth % self.k
for a, key, is_left in cons:
if is_left:
assert superkey(t.point, a) < key
else:
assert superkey(t.point, a) >= key
key = superkey(t.point, t.axis)
stack.append((t.left, depth + 1, cons + [(t.axis, key, True)]))
stack.append((t.right, depth + 1, cons + [(t.axis, key, False)]))
assert count == self.size
test_kd_tree.py 110 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/kd-tree/. The authors are listed in prov/39PCSM.jsonl.
import math
import random
from kd_tree import KdTree, Node
def shape(t: "Node | None") -> object:
if t is None:
return None
if t.left is None and t.right is None:
return t.point
return (shape(t.left), t.point, shape(t.right))
P = [(2, 3), (5, 4), (9, 6), (4, 7), (8, 1), (7, 2)]
def test_page_example() -> None:
t = KdTree(2, P)
t.check_invariant()
assert shape(t.root) == (((2, 3), (5, 4), (4, 7)), (7, 2), ((8, 1), (9, 6), None))
assert sorted(t.range_search((3, 1), (8, 5))) == [(5, 4), (7, 2), (8, 1)]
assert t.nearest((9, 2)) == (8, 1)
t.insert((6, 3))
t.check_invariant()
assert shape(t.root) == (((None, (2, 3), (6, 3)), (5, 4), (4, 7)), (7, 2), ((8, 1), (9, 6), None))
assert t.delete((7, 2))
t.check_invariant()
assert shape(t.root) == (((None, (2, 3), (6, 3)), (5, 4), (4, 7)), (8, 1), (9, 6))
assert t.delete((5, 4))
t.check_invariant()
assert shape(t.root) == (((None, (2, 3), (6, 3)), (4, 7), None), (8, 1), (9, 6))
def test_edges() -> None:
t = KdTree(3)
assert t.nearest((0, 0, 0)) is None
assert t.range_search((0, 0, 0), (1, 1, 1)) == []
assert not t.delete((1, 2, 3))
t.insert((1, 2, 3))
assert t.contains((1, 2, 3)) and t.nearest((5, 5, 5)) == (1, 2, 3)
assert t.delete((1, 2, 3)) and t.root is None
# duplicates and equal coordinates
d = KdTree(2, [(1, 1)] * 5 + [(1, 2), (2, 1)])
d.check_invariant()
assert len(d.range_search((1, 1), (1, 1))) == 5
for _ in range(5):
assert d.delete((1, 1))
d.check_invariant()
assert not d.contains((1, 1))
# one dimension
one = KdTree(1, [(x,) for x in [5, 3, 8, 1, 4]])
one.check_invariant()
assert sorted(one.range_search((2,), (5,))) == [(3,), (4,), (5,)]
def test_balanced_build_height() -> None:
rng = random.Random(3)
for n in [1, 2, 3, 7, 8, 100, 1000]:
pts = [(rng.random(), rng.random()) for _ in range(n)]
t = KdTree(2, pts)
t.check_invariant()
assert t.height() == math.floor(math.log2(n)) + 1
def test_sorted_insertions_make_a_path() -> None:
t = KdTree(2)
for i in range(200):
t.insert((i, i))
t.check_invariant()
assert t.height() == 200 # pitfall: no rebalancing on insertion
def test_random_operations_against_list() -> None:
rng = random.Random(1975)
for trial in range(60):
k = rng.randint(1, 3)
model = [tuple(float(rng.randint(0, 6)) for _ in range(k)) for _ in range(rng.randint(0, 20))]
t = KdTree(k, model)
t.check_invariant()
for _ in range(40):
op = rng.random()
if op < 0.3:
p = tuple(float(rng.randint(0, 6)) for _ in range(k))
t.insert(p)
model.append(p)
elif op < 0.5:
p = tuple(float(rng.randint(0, 6)) for _ in range(k))
ok = t.delete(p)
assert ok == (p in model)
if ok:
model.remove(p)
elif op < 0.8:
lo = [float(rng.randint(-1, 7)) for _ in range(k)]
hi = [lo[i] + rng.randint(0, 4) for i in range(k)]
want = sorted(p for p in model if all(lo[i] <= p[i] <= hi[i] for i in range(k)))
assert sorted(t.range_search(lo, hi)) == want
else:
q = [rng.uniform(-1, 7) for _ in range(k)]
got = t.nearest(q)
if not model:
assert got is None
else:
best = min(math.dist(q, p) for p in model)
assert got is not None and abs(math.dist(q, got) - best) < 1e-9
t.check_invariant()
assert sorted(t.points()) == sorted(model)
What the next level needs
- the tests did not run in this build