import logging
import numpy as np
import pandas as pd
from sklearn.cluster import DBSCAN
import warnings
try:
from ._labels import renumber_by_population
except ImportError: # pragma: no cover - fallback for direct source usage
from _labels import renumber_by_population
warnings.filterwarnings("ignore", category=RuntimeWarning)
logging.basicConfig(level=logging.INFO, format="%(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
[docs]
def compute_cluster_dbscan(
df,
feature_cols=None,
eps=0.5,
min_samples=5,
):
"""
Cluster the data using the DBSCAN algorithm.
Parameters
----------
df : pandas.DataFrame
Data to cluster. If `feature_cols` is given, only those columns
are used to fit DBSCAN (the returned dataframe keeps ALL of
`df`'s original columns, plus `cluster`) -- use this whenever
`df` also carries metadata columns (sim_name, replica, frame,
...) that must not be fed to the clustering itself. If
`feature_cols` is None (default, backward-compatible), `df` is
used as-is -- every column must be numeric.
feature_cols : list of str, optional
Column names in `df` to actually cluster on. Default None (use
all of `df`).
eps : float, optional
The maximum distance between two samples for one to be
considered in the neighborhood of the other. Default 0.5.
min_samples : int, optional
The number of samples in a neighborhood for a point to be
considered as a core point. Default 5.
Returns
-------
new_df : pandas.DataFrame
`df` (all original columns preserved) with a 'cluster' column
added: real clusters are renumbered 1..k by decreasing
population, and noise/unclustered points are labeled `-1` --
the standard scikit-learn/DBSCAN convention, and what
`plot_cluster_timeline`/`assign_cluster_representative` expect
by default (`noise_labels=(-1,)` / `noise_label=-1`).
figures : dict
Always `{}` -- DBSCAN has no natural model-selection curve to
plot (no `k` to search over). Kept in the return signature for
API consistency with the other `compute_cluster_*` functions
(`new_df, figures = compute_cluster_dbscan(...)`). To visualize
the resulting clusters, merge the `cluster` column back into
your full dataframe and use `plot_scatter` (or
`Dataset.plot_scatter`).
"""
X = df[feature_cols] if feature_cols is not None else df
clusterer = DBSCAN(eps=eps, min_samples=min_samples).fit(X)
labels = clusterer.labels_
n_clusters_ = len(set(labels)) - (1 if -1 in labels else 0)
n_noise_ = int(np.sum(labels == -1))
logger.info(
"Number of clusters: %d, unclustered (noise) points: %.1f%%",
n_clusters_,
100 * n_noise_ / len(labels),
)
new_labels = renumber_by_population(labels)
new_df = df.copy()
new_df["cluster"] = new_labels
for clust, size in new_df["cluster"].value_counts().sort_index().items():
percentage = 100 * size / len(new_df)
name = "Noise (-1)" if clust == -1 else f"Cluster {clust}"
logger.info(f"{name:14}{size:6} | {percentage:6.2f}%")
logger.info("DBSCAN clustering completed. Number of clusters: %d", n_clusters_)
return new_df, {}
def _renumber_by_population(labels):
"""
Renumber cluster labels 1..k by decreasing population, keeping noise
as `-1` (not remapped, not turned into NaN). Shared logic between
DBSCAN and HDBSCAN, which both use scikit-learn's `-1`-for-noise
convention on `.labels_`.
"""
labels = np.asarray(labels)
counts = pd.Series(labels).value_counts() # excludes nothing; -1 included
# Sort real clusters by population (descending); noise handled separately.
real_clusters = [c for c in counts.index if c != -1]
real_clusters_sorted = sorted(real_clusters, key=lambda c: -counts[c])
remap = {old: new for new, old in enumerate(real_clusters_sorted, start=1)}
remap[-1] = -1
new_labels = np.vectorize(remap.get)(labels)
return pd.Categorical(new_labels)