Algorithm Search⌘K

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
kd_tree.pyRawGitHub
# 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

Read the full page about k-d tree