bioamla.cluster

bioamla.cluster

Clustering, dimensionality reduction, and novel-sound discovery.

Cluster audio embeddings (HDBSCAN, k-means, DBSCAN, agglomerative), reduce embedding dimensionality (UMAP, t-SNE, PCA), measure cluster quality, and detect novel/outlier sounds.

Heavy backends (umap-learn, hdbscan, scikit-learn, torch) ship in the base install but are imported lazily so importing this module stays fast.

Example

import numpy as np from bioamla.cluster import AudioClusterer, reduce_dimensions emb = np.random.rand(100, 32) reduced = reduce_dimensions(emb, method="pca", n_components=2) labels = AudioClusterer(method="kmeans").fit_predict(reduced)

AudioClusterer

Clustering for audio embeddings.

Supports multiple clustering algorithms with automatic parameter tuning.

__init__

__init__(
    config: ClusteringConfig | None = None,
    method: str | None = None,
    **kwargs: Any,
) -> None

Initialize clusterer.

Parameters:

Name	Type	Description	Default
config	ClusteringConfig | None	Clustering configuration	None
method	str | None	Clustering method (overrides config)	None
**kwargs	Any	Additional arguments for the clusterer	{}

fit

fit(embeddings: ndarray) -> AudioClusterer

Fit the clusterer on embeddings.

fit_predict

fit_predict(embeddings: ndarray) -> np.ndarray

Fit and return cluster labels.

predict

predict(embeddings: ndarray) -> np.ndarray

Predict cluster labels for new embeddings.

get_cluster_centers

get_cluster_centers(embeddings: ndarray) -> np.ndarray

Compute cluster centers from embeddings.

get_cluster_stats

get_cluster_stats(
    embeddings: ndarray,
) -> dict[int, dict[str, Any]]

Get statistics for each cluster.

ClusteringConfig dataclass

Configuration for clustering.

IncrementalReducer

Incremental dimensionality reducer for streaming data.

Fits on initial data and can transform new points without refitting.

__init__

__init__(
    method: str = "umap",
    n_components: int = 2,
    **kwargs: Any,
) -> None

Initialize incremental reducer.

Parameters:

Name	Type	Description	Default
method	str	Reduction method ("umap" or "pca")	'umap'
n_components	int	Number of output dimensions	2
**kwargs	Any	Additional arguments for the reducer	{}

fit

fit(embeddings: ndarray) -> IncrementalReducer

Fit the reducer on embeddings.

transform

transform(embeddings: ndarray) -> np.ndarray

Transform new embeddings.

fit_transform

fit_transform(embeddings: ndarray) -> np.ndarray

Fit and transform embeddings.

NoveltyDetector

Detect novel sound types in audio embeddings.

Uses distance-based and density-based methods to identify sounds that don't fit existing clusters.

__init__

__init__(
    method: str = "distance",
    threshold: float | None = None,
    contamination: float = 0.1,
) -> None

Initialize novelty detector.

Parameters:

Name	Type	Description	Default
method	str	Detection method ("distance", "isolation_forest", "lof")	'distance'
threshold	float | None	Novelty threshold (auto-computed if None)	None
contamination	float	Expected proportion of outliers	0.1

fit

fit(
    embeddings: ndarray, labels: ndarray | None = None
) -> NoveltyDetector

Fit the novelty detector.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Training embeddings (known sounds)	required
labels	ndarray | None	Optional cluster labels	None

Returns:

Type	Description
NoveltyDetector	self

Raises:

Type	Description
InvalidInputError	If the method is unknown.

predict

predict(embeddings: ndarray) -> list[NoveltyResult]

Detect novel sounds in new embeddings.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Embeddings to check for novelty	required

Returns:

Type	Description
list[NoveltyResult]	List of NoveltyResult for each embedding

get_novel_samples

get_novel_samples(
    embeddings: ndarray, n_samples: int | None = None
) -> list[int]

Get indices of most novel samples.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Embeddings to check	required
n_samples	int | None	Number of novel samples to return (all if None)	None

Returns:

Type	Description
list[int]	Indices of novel samples, sorted by novelty score

NoveltyResult dataclass

Result from novelty detection.

ReductionConfig dataclass

Configuration for dimensionality reduction.

ClusterAnalysis dataclass

Detailed cluster analysis results.

to_dict

to_dict() -> dict[str, Any]

Convert to a plain dictionary.

ClusteringSummary dataclass

Summary of clustering results.

to_dict

to_dict() -> dict[str, Any]

Convert to a plain dictionary.

NoveltyDetectionSummary dataclass

Summary of novelty detection results.

to_dict

to_dict() -> dict[str, Any]

Convert to a plain dictionary.

ClusteringError

Bases: BioamlaError

A clustering, dimensionality-reduction, or novelty-detection failure.

DependencyError

Bases: BioamlaError

A required dependency is unavailable at runtime.

All Python dependencies ship in the base install, so this is reserved for genuine environment problems (e.g. a missing system library or a broken install).

cluster_batch_files

cluster_batch_files(
    input_dir: str | Path,
    output_dir: str | Path,
    *,
    method: str = "hdbscan",
    n_clusters: int | None = None,
    min_cluster_size: int = 5,
    min_samples: int = 3,
    recursive: bool = True,
    continue_on_error: bool = True,
    **kwargs: Any,
) -> BatchResult

Cluster all embedding files under a directory (two-phase).

Discovers embedding files under input_dir and delegates to :func:cluster_embedding_files.

Parameters:

Name	Type	Description	Default
input_dir	str | Path	Directory containing embedding files.	required
output_dir	str | Path	Directory where cluster_assignments.json is written.	required
method	str	Clustering method ("hdbscan", "kmeans", "dbscan", "agglomerative").	'hdbscan'
n_clusters	int | None	Number of clusters (for k-means/agglomerative).	None
min_cluster_size	int	Minimum cluster size (for HDBSCAN).	5
min_samples	int	Minimum samples per cluster.	3
recursive	bool	Whether to recurse into subdirectories.	True
continue_on_error	bool	Keep going when an individual file fails to load.	True
**kwargs	Any	Additional arguments forwarded to the clusterer.	{}

Returns:

Name	Type	Description
A	BatchResult	class:bioamla.batch.BatchResult summarizing the run; on success its
	BatchResult	metadata holds clustering counts and the output file path.

cluster_embedding_files

cluster_embedding_files(
    files: list[str | Path],
    output_dir: str | Path,
    *,
    method: str = "hdbscan",
    n_clusters: int | None = None,
    min_cluster_size: int = 5,
    min_samples: int = 3,
    continue_on_error: bool = True,
    **kwargs: Any,
) -> BatchResult

Cluster an explicit list of embedding files (two-phase).

Phase 1 loads each embedding file (tracked via :func:bioamla.batch.run_batch so per-file failures are recorded). Phase 2 concatenates the successfully loaded embeddings, clusters them once, maps labels back to files and writes <output_dir>/cluster_assignments.json.

Parameters:

Name	Type	Description	Default
files	list[str | Path]	Explicit list of embedding file paths.	required
output_dir	str | Path	Directory where cluster_assignments.json is written.	required
method	str	Clustering method ("hdbscan", "kmeans", "dbscan", "agglomerative").	'hdbscan'
n_clusters	int | None	Number of clusters (for k-means/agglomerative).	None
min_cluster_size	int	Minimum cluster size (for HDBSCAN).	5
min_samples	int	Minimum samples per cluster.	3
continue_on_error	bool	Keep going when an individual file fails to load.	True
**kwargs	Any	Additional arguments forwarded to the clusterer.	{}

Returns:

Name	Type	Description
A	BatchResult	class:bioamla.batch.BatchResult summarizing the run.

load_embedding_file

load_embedding_file(file_path: str | Path) -> np.ndarray

Load a single embedding file as a 2D array.

Supports .npy, .pkl/.pickle and .json. 1D arrays are reshaped to (1, n_features) and arrays of rank > 2 are flattened to (n_samples, n_features).

Parameters:

Name	Type	Description	Default
file_path	str | Path	Path to the embedding file.	required

Returns:

Type	Description
ndarray	Embedding array of shape (n_samples, n_features).

Raises:

Type	Description
InvalidInputError	If the file extension is unsupported.

load_embeddings_batch

load_embeddings_batch(
    input_dir: str | Path, *, recursive: bool = True
) -> tuple[list[np.ndarray], list[str]]

Phase 1: discover and load every embedding file under input_dir.

Parameters:

Name	Type	Description	Default
input_dir	str | Path	Directory to search for embedding files.	required
recursive	bool	Whether to recurse into subdirectories.	True

Returns:

Type	Description
list[ndarray]	Tuple (embeddings, filepaths) of per-file arrays and their string
list[str]	paths, in matching order.

analyze_clusters

analyze_clusters(
    embeddings: ndarray,
    labels: ndarray,
    metadata: list[dict[str, Any]] | None = None,
) -> dict[str, Any]

Comprehensive cluster analysis.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Input embeddings	required
labels	ndarray	Cluster labels	required
metadata	list[dict[str, Any]] | None	Optional metadata for each sample	None

Returns:

Type	Description
dict[str, Any]	Analysis results as a dictionary

analyze_clusters_summary

analyze_clusters_summary(
    embeddings: ndarray,
    labels: ndarray,
    filepaths: list[str] | None = None,
) -> ClusterAnalysis

Perform detailed analysis of clustering results and return a dataclass.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Input embeddings	required
labels	ndarray	Cluster labels	required
filepaths	list[str] | None	Optional list of file paths for per-cluster metadata	None

Returns:

Name	Type	Description
A	ClusterAnalysis	class:ClusterAnalysis with quality metrics and per-cluster stats.

cluster_embeddings

cluster_embeddings(
    embeddings: ndarray,
    method: str = "hdbscan",
    n_clusters: int | None = None,
    min_cluster_size: int = 5,
    min_samples: int = 3,
    **kwargs: Any,
) -> ClusteringSummary

Cluster embeddings and return a summary.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Input embeddings of shape (n_samples, n_features)	required
method	str	Clustering method ("hdbscan", "kmeans", "dbscan", "agglomerative")	'hdbscan'
n_clusters	int | None	Number of clusters (for k-means/agglomerative)	None
min_cluster_size	int	Minimum cluster size (for HDBSCAN)	5
min_samples	int	Minimum samples per cluster	3
**kwargs	Any	Additional arguments forwarded to the clusterer constructor	{}

Returns:

Name	Type	Description
A	ClusteringSummary	class:ClusteringSummary whose labels hold the per-sample
	ClusteringSummary	cluster assignments.

Raises:

Type	Description
ClusteringError	If clustering fails for any other reason.

compute_cluster_similarity

compute_cluster_similarity(
    embeddings: ndarray,
    labels: ndarray,
    metric: str = "cosine",
) -> np.ndarray

Compute pairwise similarity between clusters.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Input embeddings	required
labels	ndarray	Cluster labels	required
metric	str	Distance metric ("cosine", "euclidean")	'cosine'

Returns:

Type	Description
ndarray	Similarity matrix of shape (n_clusters, n_clusters)

Raises:

Type	Description
InvalidInputError	If the metric is unknown.

detect_novelty

detect_novelty(
    embeddings: ndarray,
    known_embeddings: ndarray | None = None,
    known_labels: ndarray | None = None,
    method: str = "distance",
    threshold: float | None = None,
    contamination: float = 0.1,
) -> tuple[NoveltyDetectionSummary, np.ndarray, np.ndarray]

Detect novel/outlier samples in embeddings.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Embeddings to check for novelty	required
known_embeddings	ndarray | None	Optional known embeddings to fit the detector on	None
known_labels	ndarray | None	Optional labels for known embeddings	None
method	str	Detection method ("distance", "isolation_forest", "lof")	'distance'
threshold	float | None	Novelty threshold (auto if None)	None
contamination	float	Expected proportion of outliers	0.1

Returns:

Type	Description
NoveltyDetectionSummary	Tuple of (summary, is_novel, novelty_scores) where is_novel is a
ndarray	boolean array and novelty_scores a float array, both per sample.

discover_novel_sounds

discover_novel_sounds(
    embeddings: ndarray,
    known_labels: ndarray | None = None,
    method: str = "distance",
    threshold: float | None = None,
    return_scores: bool = False,
) -> np.ndarray | tuple[np.ndarray, np.ndarray]

Discover novel sound types in embeddings.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	All embeddings to analyze	required
known_labels	ndarray | None	Labels for known sounds (None = all unknown)	None
method	str	Detection method	'distance'
threshold	float | None	Novelty threshold	None
return_scores	bool	Whether to return novelty scores	False

Returns:

Type	Description
ndarray | tuple[ndarray, ndarray]	Binary array indicating novel sounds (and optionally scores)

export_clusters

export_clusters(
    labels: ndarray,
    filepaths: list[str],
    output_dir: str,
    copy_files: bool = False,
) -> str

Export clustering results to directory structure.

Parameters:

Name	Type	Description	Default
labels	ndarray	Cluster labels	required
filepaths	list[str]	List of file paths	required
output_dir	str	Output directory	required
copy_files	bool	Whether to copy files to cluster directories	False

Returns:

Type	Description
str	Path to output directory

export_clusters_to_csv

export_clusters_to_csv(
    labels: ndarray,
    filepaths: list[str],
    output_path: str,
    reduced_embeddings: ndarray | None = None,
) -> str

Export clustering results to a CSV file.

Parameters:

Name	Type	Description	Default
labels	ndarray	Cluster labels	required
filepaths	list[str]	List of file paths	required
output_path	str	Output CSV path	required
reduced_embeddings	ndarray | None	Optional 2D coordinates to include as x/y columns	None

Returns:

Type	Description
str	Path to the written CSV file.

extract_embeddings_batch

extract_embeddings_batch(
    model: Any,
    dataloader: Any,
    device: Any = None,
    layer_name: str | None = None,
) -> np.ndarray

Extract embeddings from a model for a batch of data.

Parameters:

Name	Type	Description	Default
model	Any	PyTorch model	required
dataloader	Any	DataLoader with audio data	required
device	Any	Device to use	None
layer_name	str | None	Name of layer to extract (uses last hidden if None)	None

Returns:

Type	Description
ndarray	Embeddings array of shape (n_samples, embedding_dim)

find_optimal_clusters

find_optimal_clusters(
    embeddings: ndarray,
    method: str = "silhouette",
    k_range: tuple[int, int] = (2, 20),
) -> int

Find optimal number of clusters.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Input embeddings	required
method	str	Method for determining optimal k ("silhouette", "elbow", "gap")	'silhouette'
k_range	tuple[int, int]	Range of k values to try	(2, 20)

Returns:

Type	Description
int	Optimal number of clusters

reduce_dimensions

reduce_dimensions(
    embeddings: ndarray,
    config: ReductionConfig | None = None,
    method: str | None = None,
    n_components: int = 2,
    random_state: int = 42,
    **kwargs: Any,
) -> np.ndarray

Reduce dimensionality of embeddings.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Input embeddings of shape (n_samples, n_features)	required
config	ReductionConfig | None	Reduction configuration	None
method	str | None	Reduction method (overrides config)	None
n_components	int	Number of output dimensions	2
random_state	int	Random seed	42
**kwargs	Any	Additional arguments for the reducer	{}

Returns:

Type	Description
ndarray	Reduced embeddings of shape (n_samples, n_components)

Raises:

Type	Description
InvalidInputError	If the reduction method is unknown.

sort_by_similarity

sort_by_similarity(
    embeddings: ndarray,
    reference: ndarray | None = None,
    method: str = "nearest_neighbor",
) -> np.ndarray

Sort embeddings by similarity.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Input embeddings	required
reference	ndarray | None	Reference embedding (uses first embedding if None)	None
method	str	Sorting method ("nearest_neighbor", "spectral")	'nearest_neighbor'

Returns:

Type	Description
ndarray	Sorted indices

Raises:

Type	Description
InvalidInputError	If the method is unknown.

sort_clusters_by_similarity

sort_clusters_by_similarity(
    embeddings: ndarray,
    labels: ndarray,
    reference_label: int | None = None,
) -> list[int]

Sort cluster labels by similarity to each other.

Parameters:

Name	Type	Description	Default
embeddings	ndarray	Input embeddings	required
labels	ndarray	Cluster labels	required
reference_label	int | None	Starting cluster (uses largest if None)	None

Returns:

Type	Description
list[int]	Sorted list of cluster labels