1 Kd树的概念

        KD树(k-dimensional tree)是一种用于组织k维空间中点的数据结构,适用于高效处理范围搜索和最近邻搜索等空间查询问题。它将空间递归地划分为若干区域,每个节点代表一个超矩形区域,并通过轴对齐分割(沿某一坐标轴划分)实现空间划分。

2 Kd树的构建

2.1 构建原理

   ① 选择分割轴:通常采用轮换策略(如轮流选择x、y、z轴)或最大方差轴作为分割方向。
   ②选择分割点:常用当前维度上数据的中位数作为分割点,确保左右子树平衡。
   ③递归划分:对分割后的左右子空间重复上述过程,直到子区域包含的点数少于阈值或达到最大深度。

2.2 构建方式一

直接构建法

        ①对输入点集S中的点,按顺序逐点插入到已构建的Kd-树中,若Kd-树为空, 则建立插入点对应的根节点;否则,从根节点开始,向下进行遍历,如果在当前遍历的 节点的分割维度上,插入点的坐标小于(等于)节点的坐标,则向下遍历当前节点的左分支,反之,则遍历其右分支; ③ 重复第②步,直到无法继续遍历当前节点的某个分支时,建立插入点对应的节点,作为该分支节点

示例点集:(2,3),(5,4),(9,6),(4,7),(8,1),(7,2


①输入(2,3),作为第一个分割维度X下的根节点;

②输入(5,4),由于 X:5>2,作为(2,3)节点的右子树;

③输入(9,6),由于第一层:X 9>2,进入右子树;第二层分割维度 Y:6>4,继续进行入右子树(为空),该点作为新的右子叶;

④...... 以此类推,得到最终的KD树;

        该方法的特点是,相同的点集,点的插入顺序不同,构造出的二叉树形状不一样;二叉树的平衡性(即左右分支树深度的差)影响搜索(查找)效率,可能会出现极度不平衡的二叉树(点的个数即为深度)。一般不建议使用。

2.2 构建方式二

平衡递归构建:

        ①规定每一层的分割维度,如奇数层为X,偶数层为Y;②对输入点集按照分割维度的坐标值进行由小到大排序,选择序列的中位点确定分割线,构建中位点对应点的节点Node,利用分割线将点集划分成两个子集,对应Node的两个子节点;③对于节点 两个子集,基于步骤2)分别进行递归构建,直到子集为空(或者满足某一终止条件)。

        依旧使用上面的示例点集,我们不是直接使用(2,3)作为根节点,而是在一维分割维度下对点集进行排序;

按照X排序后 :(2,3),(4,7),(5,4),(7,2),(8,1),(9,6)

选择中位点作为我们的当前根节点,偶数点集一般选择中间的任意一个,不太会影响结果。左右两边就被分为了左子树中的点集和右子树的。下一个分割维度只需对左右子树中的点集分别进行按维度排序即可。

        对比这个Kd树,明显比直接构建法要均衡一点,不仅深度减少一层,而且确保节点的左右子树所包含的节点数尽可能相等,这就是平衡递归构建的特点和优势。

代码实现:(类方法)

    def build_KDTree(self,points,depth=0):
        """
        递归构建KD树
        :param points:用于构建树的点集
        :param depth:当前深度,默认为0
        :return:构建好的KD树的根节点
        """
        # 空树
        if not points:
            return None
        #确定当前分割维度
        slipt_dim = depth%self.dim
        # 找到中位点
        mid_point, [left_points,right_points] = self.find_median(points,slipt_dim)
        # 创建当前的节点
        treeNode = PointNode(mid_point,slipt_dim,depth+1)
        # 递归构建左右子树
        treeNode.left_tree=self.build_KDTree(left_points,depth+1)
        treeNode.right_tree = self.build_KDTree(right_points, depth + 1)

        return treeNode

3 Kd树空间查询操作

3.1 Kd-树的点查询

从根节点开始,根据输入点Q与节点所存储的点在分割维度上的大小关系,选择不同分支向下进行
递归遍历:

实现代码:(类方法)

    def find_point(self,node,point):
        """
        在KD数中查询指定点;
        :param node:当前子树的根节点
        :param point:待查找的点
        :return:找到的节点,若未找到则返回None
        """
        # 空树
        if node is None:
            print("未找到该点!")
            return None
        div_dim = node.split_dim
        # 当前节点是否是要查找的点
        if point==node.point:
            print("找到该点!")
            return node
        # 判断需要遍历当前节点的左子树还是右子树
        else:
            if point[div_dim]<node.point[div_dim]:
                return self.find_point(node.left_tree,point)
            else:
                return self.find_point(node.right_tree, point)

对此可以衍生出一个查找路径,也方便后续查找最邻近点的路径回溯:(创建了一个私有函数,方便公共成员访问)

    def find_path_method(self,node,point):
        """
        获取查询点在KD树中的搜索路径
        :param node: KD树的根节点
        :param point: 查询点
        :return: 搜索路径上的节点列表
        """
        self.path_list = []
        self._find_path(node,point)
        return self.path_list
    def _find_path(self,node,point):
        """辅助函数,递归记录搜索路径"""
        # 遍历到叶子节点结束
        if node is None:
            return
        # 记录当前节点
        self.path_list.append(node)
        # 决定遍历左或右子树
        div_dim = node.split_dim
        if point[div_dim] < node.point[div_dim]:
            self._find_path(node.left_tree,point)
        else:
            self._find_path(node.right_tree, point)

3.2 Kd-树 点的插入

对于后插入的点,和找到指定点的思路大致相同,不过最终是实现增加节点的操作。

实现代码:(类方法)

    def insert_point(self,node,point,depth=0):
        """
        向KD树中插入点
        :param node: (子)树的根节点
        :param point: 待插入点
        :param depth: 当前深度
        :return: 树根节点
        """
        point_dim = len(point)  # 点的维度
        # 如果为空树
        if node is None:
            # 创建以插入点为根节点的树
            node = PointNode(point,point_dim,0)
            return node
        # 如果不是空树
        else:
            div_dim = node.split_dim
            # 分割维度上小于子树节点值,遍历左子树
            if point[div_dim]<node.point[div_dim]:
                return self.insert_point(node.left_tree,point,depth+1)
            # 分割维度上不小于子树节点值,遍历右子树
            else:
                return self.insert_point(node.right_tree, point, depth + 1)
        return node

3.3 Kd-树的最近邻查询

最近邻查询:利用KD树的层次结构快速剪枝,减少距离计算次数。优先搜索更近的子树,并维护当前最小距离。

实现代码:(类方法)

 def neareat_point(self,node,point):
        """
        查找指定点的最近邻点
        :param node: KD树的根节点
        :param point: 查询点
        :return: 最近邻点及其距离
        """
        # 重置最近点和距离
        self.near_distance = np.inf
        self.near_point = None

        if node is None:
            return None,np.inf
        # 获取遍历路径
        path = self.find_path_method(node, point)
        if not path:
            return None, np.inf
        # 回溯查找路径
        for path_node in path[::-1]:
            self._nearestPoint(path_node,point,path)
        return self.near_point,np.around(self.near_distance,4)

    def _nearestPoint(self,node,point,path):
        """辅助函数,搜索子树查找更近的点"""
        if node is None:
            return
        # 获取分割维度
        div_dim = node.split_dim
        # 计算与当前点距离
        curr_distance = distance.pdist([node.point, point])[0]
        # 是否更新最近点
        if curr_distance < self.near_distance:
            self.near_distance = curr_distance
            self.near_point = node.point
        # 是否遍历该节点的其他子节点(该点离分割线的距离)
        distance_division = abs(point[div_dim]-node.point[div_dim])
        if self.near_distance>distance_division:
            if node.left_tree is not None and node.left_tree not in path:
                self._nearestPoint(node.left_tree,point,path)
            if node.right_tree is not None and node.right_tree not in path:
                self._nearestPoint(node.right_tree,point,path)
        return

3.4 Kd-树的正交区域查询

给定一个矩形区域,然后对Kd-树进行访问,查询落到该矩形区域的所有点集;

 # 正交区域查询
    def orthogonal_region(self,node,R):
        point_set = []
        self._orth_region(node,R,point_set)
        return point_set
    # 正交区域查询内部函数
    def _orth_region(self,node,R,point_set):
        """
        辅助函数,递归搜索正交区域
        :param node: KD树根节点
        :param R: 矩形区域[(xmin,xmax),[ymin,ymax]...]
        :param point_set: 接受包含在区域内的点集
        :return: 矩形区域内所包含的点集,包括边界上的点
        """
        if node is None:
            return
        flag = 1 #判断当前节点是否在正交区域内
        for i in range(self.dim):
            if not (R[i][0] <= node.point[i] <= R[i][1]):
                flag = 0
                break
        # 当前节点在正交区域内,添加到点集中
        if flag:
            point_set.append(node.point)
            self._orth_region(node.left_tree,R,point_set)
            self._orth_region(node.right_tree,R,point_set)
        else:
            div_dim = node.split_dim
            if R[div_dim][1]<node.point[div_dim]:
                self._orth_region(node.left_tree,R,point_set)
            elif R[div_dim][1]>node.point[div_dim]:
                self._orth_region(node.right_tree,R,point_set)
            else:
                # 区域跨分割线,需搜索左右子树
                self._orth_region(node.left_tree, R, point_set)
                self._orth_region(node.right_tree, R, point_set)

4 代码测试

完整代码:

import numpy as np
import pandas as pd
from fontTools.misc.cython import returns
from scipy.interpolate import splint
from scipy.spatial import distance

# KD树节点类
class PointNode:
    def __init__(self,point,dim,depth):
        """
        初始化KD树节点
        :param point: 节点存储的点坐标,列表或数组
        :param split_dim: 分割维度,整数
        :param depth: 节点所在深度,整数
        """
        self.point = point  # 点坐标
        self.split_dim = dim  # 分割维度
        self.depth = depth  # 节点深度
        self.left_tree = None  # 左子树
        self.right_tree = None  # 右子树

# KD树实现
class KDTree:
    def __init__(self,dim,node=None):
        """
        初始化KD树
        :param dim:数据点的维度
        :param node:
        """
        self.dim = dim
        self.root = node
        self.path_list = [] #用于存储查询路径
        # 初始化候选节点(方便寻找最近点)
        self.near_distance = np.inf  # 初始化一个极大最近距离
        self.near_point = None # 最近点

    def find_median(self,points,dim):
        """
        寻找指定维度下的中位点,并分割点集
        :param points: 点集
        :param dim: 要分割的维度
        :return: 中位点,分割的左右点集
        """
        #按分割纬度排序
        sorted_points = sorted(points,key=lambda t:t[dim])
        len_points = len(sorted_points)
        mid_index = len_points // 2
        # 点数量为偶数时,选择更合适的中位点
        if len_points%2==0:
            # 比较中间两个点哪个更适合作为中位点
            left_mid = sorted_points[mid_index - 1]
            right_mid = sorted_points[mid_index]
            # 创建中间参考点
            mid_value = (left_mid[dim]+right_mid[dim])/2.0
            pt = np.zeros(self.dim)
            pt[dim] = mid_value
            # 选择在该维度上离上一节点最近的
            if distance.pdist([left_mid,pt])[0]<distance.pdist([right_mid,pt])[0]:
                mid_index = mid_index-1

        left_points = sorted_points[:mid_index] #左子树点集
        right_points = sorted_points[mid_index+1:]  # 右子树点集
        return sorted_points[mid_index],[left_points,right_points]

    def build_KDTree(self,points,depth=0):
        """
        递归构建KD树
        :param points:用于构建树的点集
        :param depth:当前深度,默认为0
        :return:构建好的KD树的根节点
        """
        # 空树
        if not points:
            return None
        #确定当前分割维度
        slipt_dim = depth%self.dim
        # 找到中位点
        mid_point, [left_points,right_points] = self.find_median(points,slipt_dim)
        # 创建当前的节点
        treeNode = PointNode(mid_point,slipt_dim,depth+1)
        # 递归构建左右子树
        treeNode.left_tree=self.build_KDTree(left_points,depth+1)
        treeNode.right_tree = self.build_KDTree(right_points, depth + 1)

        return treeNode

    def insert_point(self,node,point,depth=0):
        """
        向KD树中插入点
        :param node: (子)树的根节点
        :param point: 待插入点
        :param depth: 当前深度
        :return: 树根节点
        """
        point_dim = len(point)  # 点的维度
        # 如果为空树
        if node is None:
            # 创建以插入点为根节点的树
            node = PointNode(point,point_dim,0)
            return node
        # 如果不是空树
        else:
            div_dim = node.split_dim
            # 分割维度上小于子树节点值,遍历左子树
            if point[div_dim]<node.point[div_dim]:
                return self.insert_point(node.left_tree,point,depth+1)
            # 分割维度上不小于子树节点值,遍历右子树
            else:
                return self.insert_point(node.right_tree, point, depth + 1)
        return node

    def find_point(self,node,point):
        """
        在KD数中查询指定点;
        :param node:当前子树的根节点
        :param point:待查找的点
        :return:找到的节点,若未找到则返回None
        """
        # 空树
        if node is None:
            print("未找到该点!")
            return None
        div_dim = node.split_dim
        # 当前节点是否是要查找的点
        if point==node.point:
            print("找到该点!")
            return node
        # 判断需要遍历当前节点的左子树还是右子树
        else:
            if point[div_dim]<node.point[div_dim]:
                return self.find_point(node.left_tree,point)
            else:
                return self.find_point(node.right_tree, point)

    def find_path_method(self,node,point):
        """
        获取查询点在KD树中的搜索路径
        :param node: KD树的根节点
        :param point: 查询点
        :return: 搜索路径上的节点列表
        """
        self.path_list = []
        self._find_path(node,point)
        return self.path_list
    def _find_path(self,node,point):
        """辅助函数,递归记录搜索路径"""
        # 遍历到叶子节点结束
        if node is None:
            return
        # 记录当前节点
        self.path_list.append(node)
        # 决定遍历左或右子树
        div_dim = node.split_dim
        if point[div_dim] < node.point[div_dim]:
            self._find_path(node.left_tree,point)
        else:
            self._find_path(node.right_tree, point)

    def neareat_point(self,node,point):
        """
        查找指定点的最近邻点
        :param node: KD树的根节点
        :param point: 查询点
        :return: 最近邻点及其距离
        """
        # 重置最近点和距离
        self.near_distance = np.inf
        self.near_point = None

        if node is None:
            return None,np.inf
        # 获取遍历路径
        path = self.find_path_method(node, point)
        if not path:
            return None, np.inf
        # 回溯查找路径
        for path_node in path[::-1]:
            self._nearestPoint(path_node,point,path)
        return self.near_point,np.around(self.near_distance,4)

    def _nearestPoint(self,node,point,path):
        """辅助函数,搜索子树查找更近的点"""
        if node is None:
            return
        # 获取分割维度
        div_dim = node.split_dim
        # 计算与当前点距离
        curr_distance = distance.pdist([node.point, point])[0]
        # 是否更新最近点
        if curr_distance < self.near_distance:
            self.near_distance = curr_distance
            self.near_point = node.point
        # 是否遍历该节点的其他子节点(该点离分割线的距离)
        distance_division = abs(point[div_dim]-node.point[div_dim])
        if self.near_distance>distance_division:
            if node.left_tree is not None and node.left_tree not in path:
                self._nearestPoint(node.left_tree,point,path)
            if node.right_tree is not None and node.right_tree not in path:
                self._nearestPoint(node.right_tree,point,path)
        return

    # 正交区域查询
    def orthogonal_region(self,node,R):
        point_set = []
        self._orth_region(node,R,point_set)
        return point_set
    # 正交区域查询内部函数
    def _orth_region(self,node,R,point_set):
        """
        辅助函数,递归搜索正交区域
        :param node: KD树根节点
        :param R: 矩形区域[(xmin,xmax),[ymin,ymax]...]
        :param point_set: 接受包含在区域内的点集
        :return: 矩形区域内所包含的点集,包括边界上的点
        """
        if node is None:
            return
        flag = 1 #判断当前节点是否在正交区域内
        for i in range(self.dim):
            if not (R[i][0] <= node.point[i] <= R[i][1]):
                flag = 0
                break
        # 当前节点在正交区域内,添加到点集中
        if flag:
            point_set.append(node.point)
            self._orth_region(node.left_tree,R,point_set)
            self._orth_region(node.right_tree,R,point_set)
        else:
            div_dim = node.split_dim
            if R[div_dim][1]<node.point[div_dim]:
                self._orth_region(node.left_tree,R,point_set)
            elif R[div_dim][1]>node.point[div_dim]:
                self._orth_region(node.right_tree,R,point_set)
            else:
                # 区域跨分割线,需搜索左右子树
                self._orth_region(node.left_tree, R, point_set)
                self._orth_region(node.right_tree, R, point_set)

测试示例:

if __name__=="__main__":
    # 示例用法
    points = [[2,3],[5,4],[9,6],[4,7],[8,1],[7,2]]
    kdtree = KDTree(2)
    # 创建KD树
    tree_node = kdtree.build_KDTree(points,0)
    # 正交区域查询
    region =[[1, 6], [1, 7]]
    region_points = kdtree.orthogonal_region(tree_node, region)
    print(f"正交区域{region}内的点集:")
    print(region_points, "\n")
    # 最近邻查询
    query_point = [6.5, 0]
    nearest_point, nearest_dist = kdtree.neareat_point(tree_node,query_point)
    print(f"查询点{query_point}的最近邻点:{nearest_point}")
    print(f"最近邻距离:{nearest_dist}")
    # 插入新点测试
    new_point = [6, 3]
    root = kdtree.insert_point(tree_node, new_point)
    print(f"\n插入点 {new_point} 后查询结果:")
    print(kdtree.find_point(root, new_point).point)

运行结果如下:(测试用例不太全面,可能有些错误无法检测出来,所以代码仅供参考)

5 scipy 的KDTree

在scipy中包含许多空间处理的方法,其中也包含最近邻查询,可实现多种类型的最近邻查询,总结知识点如下:

编程实现以下最基本的查找指定点最近邻的一个点:

from scipy.spatial import KDTree
points = [[2, 3], [5, 4], [9, 6], [4, 7], [8, 1], [7, 2]]
# 构建Kd-树
root = KDTree(points)
# 查询最邻近
query_point = [6.5,0]
distance,index = root.query(query_point)
print(f"查询点{query_point}的最近邻点:{points[index]}")
print(f"最近邻距离:{distance}")
# 查询以(5,4)为圆心,半径为sqrt(10)范围内的点集
circle_point = [5,4]
radius = np.sqrt(10)
point_index = root.query_ball_point(circle_point,radius)
print(f"在查询点{circle_point}距离{np.round(radius,3)}范围内的点:")
print([points[i] for i in point_index])

结果如下:

正好可以验证我们之前的结果,该方法可以直接创建好kd-树,但是就是不知道他是怎么构建的,尝试了好几种方法,都没能解决节点的内容输出,求大家指点迷津!scipy中的最近邻查询还有其他方法,以及更加炫酷的操作,还需要大家自行探索哦。今天的分享就到这里啦。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐