Kd树与空间查询(Python)
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-树的点查询:
实现代码:(类方法)
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中的最近邻查询还有其他方法,以及更加炫酷的操作,还需要大家自行探索哦。今天的分享就到这里啦。
更多推荐


所有评论(0)