impl of DBSCAN

dbscan.py

import numpy as np
from foc import fx

@fx
def dbscan(x, eps=1, min_samples=2):
    d = np.array(x).reshape(-1, 1)
    npoints = len(d)
    labels = np.full(npoints, -1)  # init all labels to -1
    visited = np.zeros(npoints, dtype=bool)

    def find_members(i):
        """find cluster-member candidates in the same cluster"""
        distances = np.sqrt(np.sum((d - d[i]) ** 2, axis=1))
        return np.where(distances <= eps)[0]

    def merge_members(i, cid, members):
        labels[i] = cid
        j = 0
        while j < len(members):
            k = members[j]
            if not visited[k]:
                visited[k] = True
                new_members = find_members(k)
                if len(new_members) >= min_samples:
                    members = np.append(members, new_members)
            if labels[k] == -1:
                labels[k] = cid
            j += 1

    cid = 0  # cluster_id
    for i in range(npoints):
        if not visited[i]:
            visited[i] = True
            members = find_members(i)
            if len(members) >= min_samples:
                merge_members(i, cid, members)
                cid += 1
    g = {}
    for k, v in zip(labels, d.tolist()):
        g[k] = g.get(k, [])
        g[k].extend(v) if len(v) == 1 else g[k].append(v)
    return g