DBScan algorithm implement and detail commented

a.csv

b.csv

DBScan.py

from sklearn.datasets import make_blobs
from matplotlib import pyplot as plt
import numpy as np

def distance(ix, iy):
    return np.linalg.norm(x_map[ix] - x_map[iy])

def is_cluster(ix):
    return yy[ix] > 0

def is_noisy(ix):
    return yy[ix] == -1

def check_eps(ix, eps):
    '''
    查找编号为ix的对象在eps范围内的点,方式为线性遍历，可以考虑使用KD树方法优化搜索。
    @return 返回 (count,ix_list)
    '''
    ix_list = []

    for iy in x_map.keys():
        #         if ix == iy:
        #             continue
        if distance(ix, iy) <= eps:
            ix_list.append(iy)

    return len(ix_list), ix_list

def label_ix(ix, clust_id):
    yy[ix] = clust_id

def debug(*arg):
    if look:
        print(arg)

def run():
    for ix, x in x_map.items():
        debug(f"{ix},位置{x}")
        if is_noisy(ix):
            debug("\t是噪音")
            continue
        elif is_cluster(ix):
            debug(f"\t已经被聚类为{yy[ix]}")
            continue
        else:
            count, ix_list = check_eps(ix, eps)
            if count < min_pt:
                debug(f"\t直接可达只有{count}个，小于{min_pt}")
                label_ix(ix, -1)
            else:
                debug(f"\t直接可达：{ix_list}")
                C = len(cluset) + 1
                label_ix(ix, C)
                cluset.add(C)

                while len(ix_list) != 0:
                    iq = ix_list.pop(0)
                    
                    if yy[iq] == -1:
                        label_ix(iq,C)
                        
                    if yy[iq] != 0:
                        continue
                    
                    label_ix(iq, C)
                    
                    count, iq_list = check_eps(iq, eps)
                    
                    debug(f"\t{iq}，位置{x_map[iq]},间接可达：{iq_list}")
                    if count >= min_pt:
                        ix_list += iq_list

    plt.scatter(X[:, 0], X[:, 1], c=yy, s=11, marker='o')
    plt.show()

#数据一
# X = np.stack([np.array([1,0]),
#             np.array([4,0]),
#             np.array([0,1]),
#             np.array([1,1]),
#             np.array([2,1]),
#             np.array([3,1]),
#             np.array([4,1]),
#             np.array([5,1]),
#             np.array([0,2]),
#             np.array([1,2]),
#             np.array([4,2]),
#             np.array([1,3])])
# eps = 1
# min_pt =4

## 数据2
# X,_=make_blobs(n_samples=200,n_features=2,centers=[[2,-2],[-2,0],[2,2]],cluster_std=[0.4,0.5,0.5])
# eps = 1
# min_pt = 15

### 数据3
X = np.genfromtxt("a.csv",delimiter=",")
eps = 0.5
min_pt = 5

### 数据4
# X = np.genfromtxt("b.csv",delimiter=",")
# eps = 3
# min_pt = 2

cluset = set()  # 已有的簇的编号
x_map = {i: x for i, x in enumerate(X)}
yy = [0] * len(X)  # 标签，默认均为未处理，设定-1 为噪声，1以上为建立的族
look = False
run()