diff --git a/.github/workflows/Documentation.yml b/.github/workflows/Documentation.yml index 39ee3fa..560b101 100644 --- a/.github/workflows/Documentation.yml +++ b/.github/workflows/Documentation.yml @@ -8,51 +8,44 @@ on: tags: '*' pull_request: -# sphinx-build doc _build permissions: - contents: write + contents: write + jobs: docs: runs-on: ubuntu-latest + steps: - uses: actions/checkout@v4 with: - lfs: 'true' + lfs: true fetch-depth: 0 - - uses: actions/setup-python@v3 - - name: Install dependencies + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: "pip" + + - name: Fetch all branches and tags run: | - # pip install -r docs/requirements.txt - pip install -e .[docs] + git fetch origin '+refs/heads/*:refs/remotes/origin/*' --tags --force - - name: Sphinx Build Develop + - name: Install dependencies run: | - git checkout develop - git fetch --all - git pull --all - cd ${{ github.workspace }}/docs - # make html - sphinx-multiversion source build/html + python -m pip install --upgrade pip + python -m pip install -e ".[docs]" + python --version + python -m pip show sphinx sphinx-multiversion - - name: Sphinx Build Main + - name: Build docs run: | - git checkout main - cd ${{ github.workspace }}/docs + cd docs sphinx-multiversion source build/html - # - name: Make symlinks - # run: | - # ln -sf ${{ github.workspace }}/build/html/main ${{ github.workspace }}/build/html/stable - # ln -sf ${{ github.workspace }}/build/html/develop ${{ github.workspace }}/build/html/dev - - name: Deploy uses: peaceiris/actions-gh-pages@v3 - if: ${{ github.event_name == 'push' && ((github.ref == 'refs/heads/main') || (github.ref == 'refs/heads/develop'))}} + if: ${{ github.event_name == 'push' && (github.ref == 'refs/heads/main' || github.ref == 'refs/heads/develop') }} with: publish_branch: gh-pages github_token: ${{ secrets.GITHUB_TOKEN }} - # publish_dir: docs/build/ - publish_dir: docs/build/html/ - # publish_dir: _build/ - # force_orphan: true + publish_dir: docs/build/html/ \ No newline at end of file diff --git a/.github/workflows/Publish.yml b/.github/workflows/Publish.yml index 87667ba..d1bd87d 100644 --- a/.github/workflows/Publish.yml +++ b/.github/workflows/Publish.yml @@ -34,7 +34,7 @@ jobs: run: python -m build - name: Publish package # uses: pypa/gh-action-pypi-publish@27b31702a0e7fc50959f5ad993c78deac1bdfc29 - uses: pypa/gh-action-pypi-publish@v1.5.1 + uses: pypa/gh-action-pypi-publish@v1.13.0 with: user: __token__ password: ${{ secrets.PYPI_API_TOKEN }} diff --git a/.github/workflows/Test.yml b/.github/workflows/Test.yml index 181e629..f5ddbfc 100644 --- a/.github/workflows/Test.yml +++ b/.github/workflows/Test.yml @@ -14,14 +14,13 @@ jobs: strategy: matrix: # Test on current Python LTS versions - python-version: ["3.7", "3.8", "3.9", "3.10", "3.11", "3.12"] + python-version: ["3.9", "3.10", "3.11", "3.12", "3.13", "3.14"] os: - ubuntu-latest # - macOS-latest # - windows-latest arch: - x64 - - x86 steps: - uses: actions/checkout@v3 diff --git a/.gitignore b/.gitignore index 8d506e0..89a44af 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,7 @@ # IDE ignores .vscode/ +.idea/ # Project ignores _dev/ @@ -142,3 +143,4 @@ dmypy.json # Pyre type checker .pyre/ +.idea/ diff --git a/README.md b/README.md index af99720..d9bafbf 100644 --- a/README.md +++ b/README.md @@ -40,6 +40,7 @@ A Python package implementing both batch and incremental cluster validity indice - [Usage](#usage) - [Quickstart](#quickstart) - [Detailed Usage](#detailed-usage) + - [Remove and Merge](#remove-and-merge) - [Implemented CVIs](#implemented-cvis) - [History](#history) - [Acknowledgements](#acknowledgements) @@ -79,7 +80,7 @@ pip install cvi You can also specify a version to install in the usual way with ```python -pip install cvi==v0.6.0 +pip install cvi==v0.7.0 ``` Alternatively, you can manually install a release from the [releases page](https://github.com/AP6YC/cvi/releases) on GitHub. @@ -114,6 +115,14 @@ for ix in range(n_samples): criterion_values = my_cvi.get_cvi(samples[ix, :], labels[ix]) ``` +Users can also query the `.info` property of the CVI objects to obtain relevant +scaling and naming information. + +``` +>>> print(my_cvi.info) +CVIInfo(name='Calinski-Harabasz', name_short='CH', index_min=0.0, index_max=inf, optimality='max') +``` + ### Detailed Usage The `cvi` package contains a set of implemented CVIs with batch and incremental update methods. @@ -169,21 +178,38 @@ for ix in range(n_samples): > **NOTE**: > -> Currently only using _either_ batch _or_ incremental methods is supported; switching from batch to incremental updates with the same is not yet implemented. +> After batch initialization, additional samples may be added incrementally by passing a single sample and label to `get_cvi`. + +### Remove and Merge + +An initialized CVI can remove a previously added sample or merge two existing clusters without retaining and replaying the full dataset: + +```python +# Remove a sample from its current cluster. +criterion_value = my_cvi.remove(sample, label) + +# Merge every member of source_label into target_label. +criterion_value = my_cvi.merge(target_label, source_label) +``` + +Both methods update the object in place and return its new criterion value. Removing the final sample of a cluster deletes that cluster, while `merge` retains `target_label` and deletes `source_label`. The caller is responsible for ensuring that a removed sample belongs to the supplied label. + +Add, remove, and merge are supported after either incremental or batch initialization. ## Implemented CVIs The following CVIs have been implemented as of the latest version of `cvi`: -- **CH**: Calinski-Harabasz -- **cSIL**: Centroid-based Silhouette -- **DB**: Davies-Bouldin +- **CH**: Calinski-Harabasz. +- **CONN**: Prototype-based intra- and inter-cluster connectivity index. +- **cSIL**: Centroid-based Silhouette index. +- **DB**: Davies-Bouldin index. - **GD43**: Generalized Dunn's Index 43. - **GD53**: Generalized Dunn's Index 53. - **PS**: Partition Separation. - **rCIP**: (Renyi's) representative Cross Information Potential. - **WB**: WB-index. -- **XB**: Xie-Beni. +- **XB**: Xie-Beni index. ## History diff --git a/docs/source/conf.py b/docs/source/conf.py index 2f7d26a..0e38ffd 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -16,8 +16,8 @@ project = 'cvi' copyright = '2024, Sasha Petrenko' author = 'Sasha Petrenko' -release = '0.6.0' -version = '0.6.0' +release = '0.7.0' +version = '0.7.0' # -- General configuration --------------------------------------------------- # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration @@ -31,8 +31,6 @@ 'sphinx.ext.intersphinx', 'sphinx.ext.napoleon', "sphinx_multiversion", - # 'sphinx.ext.autosectionlabel', - # 'sphinx_autopackagesummary', ] autosummary_generate_overwrite = True @@ -54,8 +52,6 @@ # -- Options for HTML output ------------------------------------------------- # https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output -# html_theme = 'alabaster' -# html_theme = 'sphinx_rtd_theme' html_theme = 'furo' html_static_path = ['_static'] @@ -72,47 +68,27 @@ ], } -# html_css_files = [ -# 'css/rtd.css', -# ] - # -- Options for EPUB output epub_show_urls = 'footnote' -# Whitelist pattern for tags (set to None to ignore all tags) -# smv_tag_whitelist = r'^.*$' -# smv_tag_whitelist = r'v*' +# Ignore tags for now smv_tag_whitelist = None -# smv_tag_pattern = r'^v\d*\.\d*\.\d*$' -# smv_tag_pattern = r'^.*(?!alpha)$' - -# Whitelist pattern for branches (set to None to ignore all branches) -# smv_branch_whitelist = r'^.*$ -# smv_branch_whitelist = None -# smv_branch_whitelist = r'^(main|develop)$' -# smv_branch_whitelist = r'^self-host-docs$' -smv_branch_whitelist = r'^(main|develop)$' - -# Whitelist pattern for remotes (set to None to use local branches only) -smv_remote_whitelist = None - -# Pattern for released versions -# smv_released_pattern = r'^v.*$' -# smv_released_pattern = r'v\d*\.\d*\.\d*' + +# Build docs for main and develop, whether sphinx-multiversion sees them +# as local branches or as remote branches. +smv_branch_whitelist = r'^(origin/)?(main|develop)$' + +# Allow origin/main and origin/develop. +smv_remote_whitelist = r'^origin$' + +# No tags are being built, so this does not matter much right now. smv_released_pattern = r'^tags/.*$' -# Format for versioned output directories inside the build directory +# Use simple output directories: main, develop. smv_outputdir_format = '{ref.name}' -# Determines whether remote or local git branches/tags are preferred if their output dirs conflict -smv_prefer_remote_refs = False - -# # Skip param objects because of their weird rendering in docs -# def maybe_skip_member(app, what, name, obj, skip, options): -# # print app, what, name, obj, skip, options -# # if name == "" -# return True +# Prefer remote refs in CI, since GitHub Actions reliably has origin/main +# and origin/develop after fetching. +smv_prefer_remote_refs = True -# def setup(app): -# app.connect('autodoc-skip-member', maybe_skip_member) diff --git a/docs/source/guide.rst b/docs/source/guide.rst index 3f6a36c..aaa8875 100644 --- a/docs/source/guide.rst +++ b/docs/source/guide.rst @@ -107,7 +107,24 @@ The incremental methods are used automatically based upon the dimensions of the criterion_values[ix] = my_cvi.get_cvi(sample, label) .. note:: - Currently only using *either* batch *or* incremental methods is supported; switching from batch to incremental updates with the same is not yet implemented. + After batch initialization, additional samples may be added incrementally by passing a single sample and label to ``get_cvi``. + +Remove and Merge +---------------- + +An initialized CVI can remove a previously added sample or merge two existing clusters without retaining and replaying the full dataset: + +.. code-block:: python + + # Remove a sample from its current cluster. + criterion_value = my_cvi.remove(sample, label) + + # Merge every member of source_label into target_label. + criterion_value = my_cvi.merge(target_label, source_label) + +Both methods update the object in place and return its new criterion value. Removing the final sample of a cluster deletes that cluster, while ``merge`` retains ``target_label`` and deletes ``source_label``. The caller is responsible for ensuring that a removed sample belongs to the supplied label. + +Add, remove, and merge are supported after either incremental or batch initialization. Implemented CVIs ---------------- diff --git a/pyproject.toml b/pyproject.toml index 99789cc..eb4235c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,7 +13,7 @@ authors = [ # # For a discussion on single-sourcing the version, see # https://packaging.python.org/guides/single-sourcing-package-version/ -version = "0.6.0" +version = "0.7.0" # This is a one-line description or tagline of what your project does. This # corresponds to the "Summary" metadata field: @@ -34,7 +34,7 @@ readme = "README.md" # Optional # 'Programming Language' classifiers above, 'pip install' will check this # and refuse to install the project if the version does not match. See # https://packaging.python.org/guides/distributing-packages-using-setuptools/#python-requires -requires-python = ">=3.6" +requires-python = ">=3.9" # This is either text indicating the license for the distribution, or a file # that contains the license @@ -86,13 +86,12 @@ classifiers = [ # Optional # that you indicate you support Python 3. These classifiers are *not* # checked by "pip install". See instead "python_requires" below. "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.6", - "Programming Language :: Python :: 3.7", - "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3 :: Only", ] @@ -104,6 +103,8 @@ classifiers = [ # Optional # https://packaging.python.org/discussions/install-requires-vs-requirements/ dependencies = [ # Optional "numpy", + "artlib>=0.1.7", + "scikit-learn", ] # List additional groups of dependencies here (e.g. development @@ -127,11 +128,10 @@ test = [ "pytest-cov", "coverage", "pandas", - "scikit-learn", "flake8", ] docs = [ - "sphinx", + "sphinx<9", "furo", "sphinx-multiversion", ] diff --git a/src/cvi/__init__.py b/src/cvi/__init__.py index b78bc4f..34d6551 100644 --- a/src/cvi/__init__.py +++ b/src/cvi/__init__.py @@ -3,12 +3,13 @@ """ # Set the version variable of the package -__version__ = "0.6.0" +__version__ = "0.7.0" # Import CVI modules to the top level from .modules import ( CVI, CH, + CONN, cSIL, DB, GD43, @@ -26,6 +27,7 @@ __all__ = [ "CVI", "CH", + "CONN", "cSIL", "DB", "GD43", @@ -40,6 +42,7 @@ # Convenience variable containing all implemented modules MODULES = [ CH, + CONN, cSIL, DB, GD43, diff --git a/src/cvi/modules/CH.py b/src/cvi/modules/CH.py index 8a26113..0f30a1d 100644 --- a/src/cvi/modules/CH.py +++ b/src/cvi/modules/CH.py @@ -28,6 +28,15 @@ class CH(_base.CVI): 3. M. Moshtaghi, J. C. Bezdek, S. M. Erfani, C. Leckie, and J. Bailey, "Online Cluster Validity Indices for Streaming Data," ArXiv e-prints, 2018, arXiv:1801.02937v1 [stat.ML]. [Online]. 4. M. Moshtaghi, J. C. Bezdek, S. M. Erfani, C. Leckie, J. Bailey, "Online cluster validity indices for performance monitoring of streaming data clustering," Int. J. Intell. Syst., pp. 1-23, 2018. """ + info = _base.CVIInfo( + name="Calinski-Harabasz", + name_short="CH", + index_min=0.0, + index_max=np.inf, + optimality="max" + ) + _supports_remove_merge = True + _uses_compactness_stats = True def __init__(self): """ @@ -139,16 +148,18 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): # Take the average across all samples, but cast to 1-D vector self._mu = np.mean(data, axis=0) - u = np.unique(labels) - self._n_clusters = u.size - self._n = np.zeros(self._n_clusters, dtype=int) + u = self._setup_batch_labels(labels) + self._n_clusters = len(u) + self._n = [0 for _ in range(self._n_clusters)] self._v = np.zeros((self._n_clusters, self._dim)) - self._CP = np.zeros(self._n_clusters) + self._CP = [0.0 for _ in range(self._n_clusters)] + self._G = np.zeros((self._n_clusters, self._dim)) self._SEP = np.zeros(self._n_clusters) - for ix in range(self._n_clusters): + for ix, external_label in enumerate(u): subset_indices = ( - [x for x in range(len(labels)) if labels[x] == ix] + [x for x in range(len(labels)) + if labels[x] == external_label] ) subset = data[subset_indices, :] self._n[ix] = subset.shape[0] @@ -157,6 +168,23 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): self._CP[ix] = np.sum(diff_x_v ** 2) self._SEP[ix] = self._n[ix] * np.sum((self._v[ix, :] - self._mu) ** 2) + def _rebuild_after_operation(self): + """Rebuild separation statistics after a remove or merge.""" + + if self._n_clusters == 0: + self._mu = np.zeros(0) + self._SEP = np.zeros(0) + self._BGSS = 0.0 + self._WGSS = 0.0 + return + + self._SEP = np.asarray([ + self._n[ix] * np.sum((self._v[ix, :] - self._mu) ** 2) + for ix in range(self._n_clusters) + ]) + self._WGSS = sum(self._CP) + self._BGSS = sum(self._SEP) + @_base._add_docs(_base._evaluate_doc) def _evaluate(self): """ diff --git a/src/cvi/modules/CONN.py b/src/cvi/modules/CONN.py new file mode 100644 index 0000000..8cb3d26 --- /dev/null +++ b/src/cvi/modules/CONN.py @@ -0,0 +1,853 @@ +""" +Connectivity-based CONN Cluster Validity Index. + +This implementation follows the CONN-style validity index for prototype-based +partitions. Unlike distance-only CVIs, CONN depends on the first and second +best matching prototypes associated with each sample. + +Notes +----- +Incremental mode uses FuzzyART and assumes samples are already normalized to +the ART input domain, typically [0, 1]. Batch mode supports FuzzyART, KMeans, +and MiniBatchKMeans and can optionally normalize the full dataset before +processing. + +The iCONN initialization rule is handled explicitly: + 1. The first sample creates the first ART category. + 2. The second sample forces creation of the second ART category by + temporarily setting ART vigilance to 1.0. + 3. Subsequent samples use ordinary ART dynamics. + +References +---------- +1. E. Merényi, "A new cluster validity index for prototype based clustering algorithms based on inter-and intra-cluster density," 2007 International Joint Conference on Neural Networks, 2007. +2. K. Tasdemir and E. Merényi, "A validity index for prototype-based clustering of data sets with complex cluster structures," IEEE Transactions on Systems, Man, and Cybernetics, Part B (Cybernetics), vol. 41, no. 4, pp. 1039-1053, 2011. +3. L. E. Brito da Silva, N. M. Melton, and D. C. Wunsch II, "Incremental cluster validity indices for online learning of hard partitions: Extensions and comparative study," IEEE Access, vol. 8, pp. 22025-22047, 2020. +""" + +# Standard library imports +from collections import defaultdict +from typing import Dict, Literal, Optional, Union +import numbers + +# Third-party imports +import numpy as np +from sklearn.cluster import KMeans, MiniBatchKMeans + +# ART imports +from artlib import FuzzyART, SimpleARTMAP +from artlib.common.utils import complement_code + +# Local imports +from . import _base + + +class _GrowingSquareArray: + """ + Utility for square arrays whose size is determined online. + """ + + def __init__(self, dtype=float): + self.array = np.zeros((0, 0), dtype=dtype) + + def _ensure_size(self, i: int, j: int): + size = max(i + 1, j + 1) + + if size > self.array.shape[0]: + new_array = np.zeros((size, size), dtype=self.array.dtype) + + if self.array.size > 0: + old_size = self.array.shape[0] + new_array[:old_size, :old_size] = self.array + + self.array = new_array + + def __getitem__(self, idx): + i, j = idx + + # Allow NumPy advanced indexing without resizing. + if ( + not isinstance(i, numbers.Integral) + or not isinstance(j, numbers.Integral) + ): + return self.array[idx] + + self._ensure_size(i, j) + return self.array[i, j] + + def __setitem__(self, idx, value): + i, j = idx + self._ensure_size(i, j) + self.array[i, j] = value + + def increment(self, i: int, j: int, value=1): + self._ensure_size(i, j) + self.array[i, j] += value + + def asarray(self): + return self.array.copy() + + def __repr__(self): + return repr(self.array) + + +class _GrowingArray1D: + """ + Utility for one-dimensional arrays whose size is determined online. + """ + + def __init__(self, dtype=float): + self.array = np.zeros(0, dtype=dtype) + + def _ensure_size(self, i: int): + if i >= self.array.size: + new_array = np.zeros(i + 1, dtype=self.array.dtype) + new_array[:self.array.size] = self.array + self.array = new_array + + def __getitem__(self, i: int): + self._ensure_size(i) + return self.array[i] + + def __setitem__(self, i: int, value): + self._ensure_size(i) + self.array[i] = value + + def increment(self, i: int, value=1): + self._ensure_size(i) + self.array[i] += value + + def asarray(self): + return self.array.copy() + + def __iter__(self): + return iter(self.array) + + def __len__(self): + return len(self.array) + + def __repr__(self): + return repr(self.array) + + +class _CONNFuzzyART(FuzzyART): + """ + FuzzyART extension that exposes the first and second best matching + categories for CONN updates. + """ + + def step_pred_first_and_second(self, sample: np.ndarray): + """ + Return the first and second best matching ART categories. + + Parameters + ---------- + sample : np.ndarray + Complement-coded sample. + + Returns + ------- + tuple[int, int] + First and second category indices. + + Raises + ------ + RuntimeError + If fewer than two ART categories exist. + """ + + if len(self.W) < 2: + raise RuntimeError( + "CONN requires at least two ART categories. " + "The second ART category should be forced during the " + "second-sample initialization step." + ) + + choices = [ + self.category_choice(sample, w, params=self.params)[0] + for w in self.W + ] + + choices = np.asarray(choices, dtype=float) + + first = int(np.argmax(choices)) + choices[first] = -np.inf + second = int(np.argmax(choices)) + + return first, second + + +class _CONNSimpleARTMAP(SimpleARTMAP): + """ + SimpleARTMAP extension with CONN-specific match reset behavior. + """ + + def match_reset_func( + self, + i: np.ndarray, + w: np.ndarray, + cluster_a, + params: dict, + extra: dict, + cache: Optional[dict] = None, + ) -> bool: + """ + CONN-specific match reset. + """ + + cluster_b = extra["cluster_b"] + + b_samples = sum( + self.module_a.weight_sample_counter_[a] + for a, b in self.map.items() + if b == cluster_b + ) + + if b_samples == 1: + return False + + if cluster_a in self.map and self.map[cluster_a] != cluster_b: + return False + + return True + + +class CONN(_base.CVI): + """ + CONN Cluster Validity Index. + + Incremental mode uses a FuzzyART/SimpleARTMAP model. Batch mode can use + FuzzyART or fit class-owned KMeans/MiniBatchKMeans prototypes. + + References + ---------- + 1. E. Merényi, "A new cluster validity index for prototype based clustering algorithms based on inter-and intra-cluster density," 2007 International Joint Conference on Neural Networks, 2007. + 2. K. Tasdemir and E. Merényi, "A validity index for prototype-based clustering of data sets with complex cluster structures," IEEE Transactions on Systems, Man, and Cybernetics, Part B (Cybernetics), vol. 41, no. 4, pp. 1039-1053, 2011. + 3. L. E. Brito da Silva, N. M. Melton, and D. C. Wunsch II, "Incremental cluster validity indices for online learning of hard partitions: Extensions and comparative study," IEEE Access, vol. 8, pp. 22025-22047, 2020. + """ + + info = _base.CVIInfo( + name="Connectivity", + name_short="CONN", + index_min=0.0, + index_max=1.0, + optimality="max", + ) + + def __init__( + self, + rho: float = 0.9, + alpha: float = 1e-10, + beta: float = 1.0, + match_tracking: str = "MT+", + normalize_batch: bool = True, + check_incremental_normalized: bool = True, + model_type: Literal["Fuzzy", "KMeans", "MiniBatchKMeans"] = ( + "MiniBatchKMeans" + ), + kmeans_k: Union[int, Dict[int, int]] = 8, + kmeans_kwargs: Optional[dict] = None, + ): + """ + CONN initialization routine. + + Parameters + ---------- + rho : float, default=0.9 + FuzzyART vigilance parameter. + alpha : float, default=1e-10 + FuzzyART choice parameter. + beta : float, default=1.0 + FuzzyART learning rate. + match_tracking : str, default="MT+" + Match-tracking mode passed to SimpleARTMAP. + normalize_batch : bool, default=True + If True, batch data are min-max normalized before prototype + fitting. Incremental data are not normalized online. + check_incremental_normalized : bool, default=True + If True, incremental samples are checked to ensure values lie in + [0, 1]. + model_type : {"Fuzzy", "KMeans", "MiniBatchKMeans"}, default="MiniBatchKMeans" + Prototype backend. KMeans backends support batch mode only. + kmeans_k : int or dict[int, int], default=8 + Number of KMeans prototypes per input label. Dictionary values are + keyed by the original input labels. Counts are capped at the number + of samples carrying each label. + kmeans_kwargs : dict, optional + Keyword arguments forwarded to the selected scikit-learn KMeans + estimator. ``n_clusters`` must be configured through ``kmeans_k``. + """ + + super().__init__() + + self.rho = rho + self.alpha = alpha + self.beta = beta + self.match_tracking = match_tracking + self.normalize_batch = normalize_batch + self.check_incremental_normalized = check_incremental_normalized + self.model_type = model_type + self.kmeans_k = kmeans_k + self.kmeans_kwargs = kmeans_kwargs + + self._validate_backend_params() + + if self.kmeans_kwargs is not None: + self.kmeans_kwargs = dict(self.kmeans_kwargs) + + self._data_min = None + self._data_max = None + + self._init_conn_state() + + def _init_conn_state(self): + """ + Initialize or reset all CONN-specific state. + """ + + module_a = _CONNFuzzyART( + rho=self.rho, + alpha=self.alpha, + beta=self.beta, + ) + self._artmap = _CONNSimpleARTMAP(module_a) + + # ART-category-level matrices. + self._CADJ = _GrowingSquareArray(dtype=float) + self._CONN = _GrowingSquareArray(dtype=float) + + # Label-level arrays/matrices. + self._INTRA = _GrowingArray1D(dtype=float) + self._INTER = _GrowingSquareArray(dtype=float) + + self._intra_conn = 0.0 + self._inter_conn = 0.0 + + # Internal label -> set of ART categories assigned to that label. + self._rev_map = defaultdict(set) + + # Number of samples per internal label. + self._cluster_cardinality = _GrowingArray1D(dtype=float) + + # Batch centroid-backend state. + self._kmeans_models = {} + self._cluster_centers = np.zeros((0, self._dim), dtype=float) + self._prototype_label_map = {} + + @staticmethod + def _validate_positive_int(value, name: str) -> int: + """Validate and return a strictly positive integer parameter.""" + + if isinstance(value, bool) or not isinstance(value, numbers.Integral): + raise ValueError(f"{name} must be a positive integer.") + + value = int(value) + + if value <= 0: + raise ValueError(f"{name} must be a positive integer.") + + return value + + def _validate_backend_params(self): + """Validate backend selection and KMeans configuration.""" + + valid_model_types = {"Fuzzy", "KMeans", "MiniBatchKMeans"} + + if self.model_type not in valid_model_types: + raise ValueError( + "model_type must be one of " + "{'Fuzzy', 'KMeans', 'MiniBatchKMeans'}." + ) + + if isinstance(self.kmeans_k, dict): + for label, value in self.kmeans_k.items(): + if ( + isinstance(label, bool) + or not isinstance(label, numbers.Integral) + ): + raise ValueError("kmeans_k dictionary keys must be integers.") + + self._validate_positive_int( + value, + f"kmeans_k for label {int(label)}", + ) + else: + self._validate_positive_int(self.kmeans_k, "kmeans_k") + + if self.kmeans_kwargs is not None: + if not isinstance(self.kmeans_kwargs, dict): + raise ValueError("kmeans_kwargs must be a dictionary or None.") + + if "n_clusters" in self.kmeans_kwargs: + raise ValueError( + "Configure n_clusters through kmeans_k, not kmeans_kwargs." + ) + + @_base._add_docs(_base._setup_doc) + def _setup(self, sample: np.ndarray): + """ + CONN setup routine. + """ + + super()._setup(sample) + + def _normalize_batch_data(self, data: np.ndarray) -> np.ndarray: + """ + Min-max normalize batch data featurewise. + + Constant-valued features are mapped to zero. + """ + + data = np.asarray(data, dtype=float) + + self._data_min = np.min(data, axis=0) + self._data_max = np.max(data, axis=0) + + denom = self._data_max - self._data_min + denom[denom == 0.0] = 1.0 + + return (data - self._data_min) / denom + + def _get_kmeans_k(self, label: int, n_samples: int) -> int: + """Resolve and cap the prototype count for one external label.""" + + if isinstance(self.kmeans_k, dict): + if label not in self.kmeans_k: + raise ValueError( + f"kmeans_k is missing a value for label {label}." + ) + + requested = self.kmeans_k[label] + else: + requested = self.kmeans_k + + requested = self._validate_positive_int( + requested, + f"kmeans_k for label {label}", + ) + + return min(requested, n_samples) + + def _fit_centroid_backend( + self, + data: np.ndarray, + labels: np.ndarray, + ): + """Fit class-owned KMeans prototypes and populate their label maps.""" + + estimator_type = ( + KMeans if self.model_type == "KMeans" else MiniBatchKMeans + ) + model_kwargs = self.kmeans_kwargs or {} + center_blocks = [] + next_prototype = 0 + ordered_labels = list(dict.fromkeys(int(label) for label in labels)) + + if isinstance(self.kmeans_k, dict): + missing_labels = [ + label for label in ordered_labels if label not in self.kmeans_k + ] + + if missing_labels: + raise ValueError( + "kmeans_k is missing values for labels " + f"{missing_labels}." + ) + + for label in ordered_labels: + rows = np.flatnonzero(labels == label) + i_label = self._label_map.get_internal_label(label) + n_clusters = self._get_kmeans_k(label, len(rows)) + + model = estimator_type( + n_clusters=n_clusters, + **model_kwargs, + ).fit(data[rows]) + centers = np.asarray(model.cluster_centers_, dtype=float) + prototype_ids = range( + next_prototype, + next_prototype + len(centers), + ) + + self._kmeans_models[label] = model + self._rev_map[i_label].update(prototype_ids) + + for prototype_id in prototype_ids: + self._prototype_label_map[prototype_id] = i_label + + center_blocks.append(centers) + next_prototype += len(centers) + + self._cluster_centers = np.vstack(center_blocks) + + last_prototype = len(self._cluster_centers) - 1 + self._CADJ._ensure_size(last_prototype, last_prototype) + self._CONN._ensure_size(last_prototype, last_prototype) + + def _update_conn_from_centroids( + self, + data: np.ndarray, + labels: np.ndarray, + ): + """Compute batch CONN statistics from fixed centroid prototypes.""" + + self._fit_centroid_backend(data, labels) + + for sample, label in zip(data, labels): + i_label = self._label_map.get_internal_label(int(label)) + own_prototypes = np.asarray( + sorted(self._rev_map[i_label]), + dtype=int, + ) + distances = np.sum( + (self._cluster_centers - sample) ** 2, + axis=1, + ) + + own_distances = distances[own_prototypes] + bmu1 = int(own_prototypes[np.argmin(own_distances)]) + + distances[bmu1] = np.inf + bmu2 = int(np.argmin(distances)) + + self._finish_conn_update( + i_label, + bmu1, + bmu2, + prototype_label_map=self._prototype_label_map, + update_metric=False, + ) + self._n_samples += 1 + + self._sync_base_cluster_count() + + # Recompute every row from the completed adjacency matrices so the + # fixed-prototype batch result does not depend on update order. + for i_label in sorted(self._rev_map): + self._update_metric(i_label, i_label) + + def _check_sample_normalized(self, sample: np.ndarray): + """ + Validate that an incremental sample is in the ART input domain. + """ + + if not self.check_incremental_normalized: + return + + if np.any(sample < 0.0) or np.any(sample > 1.0): + raise ValueError( + "Incremental CONN assumes samples are already normalized " + "to the ART input domain [0, 1]. For offline evaluation, " + "use batch mode with normalize_batch=True." + ) + + def _set_module_rho(self, rho: float): + """ + Set FuzzyART vigilance in a way that is robust to artlib storing rho + both as an attribute and inside the params dictionary. + """ + + self._artmap.module_a.rho = rho + + if hasattr(self._artmap.module_a, "params"): + self._artmap.module_a.params["rho"] = rho + + def _get_module_rho(self) -> float: + """ + Get the current FuzzyART vigilance. + """ + + if hasattr(self._artmap.module_a, "rho"): + return self._artmap.module_a.rho + + return self._artmap.module_a.params["rho"] + + def _force_second_category(self, sample_cc: np.ndarray, i_label: int): + """ + Force creation of the second ART category for iCONN initialization. + + This temporarily sets ART vigilance to 1.0, performs one ARTMAP update, + and then restores the original vigilance. + """ + + old_rho = self._get_module_rho() + + try: + self._set_module_rho(1.0) + + self._artmap = self._artmap.partial_fit( + np.asarray([sample_cc]), + np.asarray([i_label]), + match_tracking=self.match_tracking, + ) + + finally: + self._set_module_rho(old_rho) + + if len(self._artmap.module_a.W) < 2: + raise RuntimeError( + "Failed to force creation of the second ART category. " + "This can occur if the second sample perfectly resonates " + "with the first sample at rho=1.0." + ) + + def _sync_base_cluster_count(self): + """ + Synchronize the base CVI cluster counter with the internal label map. + """ + + self._n_clusters = len(self._label_map.map) + + def _calc_inter(self, i_label: int, j_label: int) -> float: + """ + Compute directed INTER connectivity from one internal label to another. + """ + + if i_label not in self._rev_map or j_label not in self._rev_map: + return 0.0 + + s1 = np.asarray(sorted(self._rev_map[i_label]), dtype=int) + s2 = np.asarray(sorted(self._rev_map[j_label]), dtype=int) + + if s1.size == 0 or s2.size == 0: + return 0.0 + + cadj_sub = self._CADJ[np.ix_(s1, s2)] + conn_sub = self._CONN[np.ix_(s1, s2)] + + inter_numer = conn_sub.sum() + + valid_rows = np.any(cadj_sub > 0, axis=1) + inter_denom = conn_sub[valid_rows, :].sum() + + if inter_denom == 0.0: + return 0.0 + + return float(inter_numer / inter_denom) + + def _update_metric(self, y: int, y2: int): + """ + Update INTRA, INTER, and the final CONN criterion value. + """ + + categories_y = np.asarray(sorted(self._rev_map[y]), dtype=int) + + if categories_y.size == 0 or self._cluster_cardinality[y] == 0: + self._INTRA[y] = 0.0 + else: + intra_numer = self._CADJ[np.ix_(categories_y, categories_y)].sum() + self._INTRA[y] = intra_numer / self._cluster_cardinality[y] + + active_labels = sorted(self._rev_map.keys()) + + if len(active_labels) == 0: + self._intra_conn = 0.0 + else: + self._intra_conn = float( + sum(self._INTRA[label] for label in active_labels) + / len(active_labels) + ) + + if y != y2: + self._INTER[y, y2] = self._calc_inter(y, y2) + self._INTER[y2, y] = self._calc_inter(y2, y) + else: + for label in active_labels: + if label != y: + self._INTER[y, label] = self._calc_inter(y, label) + + if len(active_labels) < 2: + self._inter_conn = 0.0 + else: + row_maxes = [] + + for label in active_labels: + off_diag_values = [ + self._INTER[label, other] + for other in active_labels + if other != label + ] + + if off_diag_values: + row_maxes.append(max(off_diag_values)) + else: + row_maxes.append(0.0) + + self._inter_conn = float(np.mean(row_maxes)) + + self.criterion_value = self._intra_conn * (1.0 - self._inter_conn) + + def _finish_conn_update( + self, + i_label: int, + bmu1: int, + bmu2: int, + prototype_label_map: Optional[dict] = None, + update_metric: bool = True, + ): + """ + Finish the CONN bookkeeping once BMU1 and BMU2 are known. + """ + + if prototype_label_map is None: + prototype_label_map = self._artmap.map + + self._rev_map[i_label].add(bmu1) + self._cluster_cardinality.increment(i_label, 1) + + self._CADJ.increment(bmu1, bmu2, 1) + + # CONN is the symmetrized co-adjacency. + conn_value = self._CADJ[bmu1, bmu2] + self._CADJ[bmu2, bmu1] + self._CONN[bmu1, bmu2] = conn_value + self._CONN[bmu2, bmu1] = conn_value + + if bmu1 not in prototype_label_map: + raise RuntimeError("BMU1 is missing from the prototype label map.") + + if int(prototype_label_map[bmu1]) != i_label: + raise RuntimeError( + "Internal prototype mapping disagrees with the provided label." + ) + + if not update_metric: + return + + if bmu2 in prototype_label_map: + y2 = int(prototype_label_map[bmu2]) + else: + y2 = i_label + + self._update_metric(i_label, y2) + + def _update_conn_from_sample(self, sample: np.ndarray, label: int): + """ + Update ART state and CONN sufficient statistics using one sample. + """ + + sample = np.asarray(sample, dtype=float) + i_label = self._label_map.get_internal_label(int(label)) + + if not self._is_setup: + self._setup(sample) + + self._check_sample_normalized(sample) + + # ART operates on complement-coded samples. + sample_cc = complement_code(np.asarray([sample]))[0] + + # First sample: + # Learn normally. CONN/iCONN is not yet defined because there is no + # second ART category. + if self._n_samples == 0: + self._artmap = self._artmap.partial_fit( + np.asarray([sample_cc]), + np.asarray([i_label]), + match_tracking=self.match_tracking, + ) + + bmu1 = int(self._artmap.module_a.labels_[-1]) + self._rev_map[i_label].add(bmu1) + self._cluster_cardinality.increment(i_label, 1) + + self._n_samples += 1 + self._sync_base_cluster_count() + + self.criterion_value = 0.0 + return + + # Second sample: + # Force creation of the second ART category according to the iCONN + # initialization rule. + if self._n_samples == 1 and len(self._artmap.module_a.W) == 1: + self._force_second_category(sample_cc, i_label) + + bmu1 = int(self._artmap.module_a.labels_[-1]) + + # With exactly two categories, the other category is BMU2. + bmu2 = 1 - bmu1 + + self._finish_conn_update(i_label, bmu1, bmu2) + + self._n_samples += 1 + self._sync_base_cluster_count() + return + + # Normal update after iCONN initialization. + self._artmap = self._artmap.partial_fit( + np.asarray([sample_cc]), + np.asarray([i_label]), + match_tracking=self.match_tracking, + ) + + bmu1 = int(self._artmap.module_a.labels_[-1]) + + c1, c2 = self._artmap.module_a.step_pred_first_and_second(sample_cc) + bmu2 = c2 if bmu1 == c1 else c1 + + self._finish_conn_update(i_label, bmu1, bmu2) + + self._n_samples += 1 + self._sync_base_cluster_count() + + @_base._add_docs(_base._param_inc_doc) + def _param_inc(self, sample: np.ndarray, label: int): + """ + Incremental parameter update for the CONN CVI. + """ + + if self.model_type != "Fuzzy": + raise ValueError( + f"model_type={self.model_type!r} supports batch mode only. " + "Use model_type='Fuzzy' for incremental CONN updates." + ) + + self._update_conn_from_sample(sample, label) + + @_base._add_docs(_base._param_batch_doc) + def _param_batch(self, data: np.ndarray, labels: np.ndarray): + """ + Batch parameter update for the CONN CVI. + + Fuzzy batch mode processes samples sequentially because CONN depends + on online ART category dynamics. KMeans backends first fit fixed + class-owned prototypes and then accumulate connectivity. + """ + + data = np.asarray(data, dtype=float) + labels = np.asarray(labels, dtype=int) + + if self.normalize_batch: + data = self._normalize_batch_data(data) + + super()._setup_batch(data) + + # Reset all base and CONN-specific state after setup_batch sets dim. + self._label_map = _base.LabelMap() + self._n_samples = 0 + self._n = [] + self._v = np.zeros([0, self._dim]) + self._CP = [] + self._G = np.zeros([0, self._dim]) + self._n_clusters = 0 + self.criterion_value = 0.0 + + self._init_conn_state() + + if self.model_type == "Fuzzy": + for sample, label in zip(data, labels): + self._update_conn_from_sample(sample, int(label)) + else: + self._update_conn_from_centroids(data, labels) + + @_base._add_docs(_base._evaluate_doc) + def _evaluate(self): + """ + Criterion value evaluation method for CONN. + + The CONN value is updated during parameter updates because the update + requires the label pair touched by the most recent ART transition. + """ + + if self._n_clusters <= 0: + self.criterion_value = 0.0 diff --git a/src/cvi/modules/DB.py b/src/cvi/modules/DB.py index 56275cd..3769bac 100644 --- a/src/cvi/modules/DB.py +++ b/src/cvi/modules/DB.py @@ -27,6 +27,16 @@ class DB(_base.CVI): 3. M. Moshtaghi, J. C. Bezdek, S. M. Erfani, C. Leckie, J. Bailey, "Online cluster validity indices for performance monitoring of streaming data clustering," Int. J. Intell. Syst., pp. 1-23, 2018. """ + info = _base.CVIInfo( + name="Davies-Bouldin", + name_short="DB", + index_min=0.0, + index_max=np.inf, + optimality="min" + ) + _supports_remove_merge = True + _uses_compactness_stats = True + def __init__(self): """ Davies-Bouldin (DB) initialization routine. @@ -160,8 +170,8 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): # Take the average across all samples, but cast to 1-D vector self._mu = np.mean(data, axis=0) - u = np.unique(labels) - self._n_clusters = u.size + u = self._setup_batch_labels(labels) + self._n_clusters = len(u) # self._n = np.zeros(self._n_clusters, dtype=int) self._n = [0 for _ in range(self._n_clusters)] self._v = np.zeros((self._n_clusters, self._dim)) @@ -172,10 +182,11 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): # self._S = np.zeros(self._n_clusters) self._S = [0 for _ in range(self._n_clusters)] - for ix in range(self._n_clusters): + for ix, external_label in enumerate(u): # subset_indices = lambda x: labels[x] == ix subset_indices = ( - [x for x in range(len(labels)) if labels[x] == ix] + [x for x in range(len(labels)) + if labels[x] == external_label] ) subset = data[subset_indices, :] self._n[ix] = subset.shape[0] @@ -192,6 +203,28 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): self._D = self._D + np.transpose(self._D) + def _rebuild_after_operation(self): + """Rebuild dispersion and centroid-distance state.""" + + if self._n_clusters == 0: + self._mu = np.zeros(0) + self._S = [] + self._D = np.zeros((0, 0)) + self._R = np.zeros((0, 0)) + return + + self._S = [ + self._CP[ix] / self._n[ix] + for ix in range(self._n_clusters) + ] + self._D = self._pairwise_matrix( + self._n_clusters, + lambda ix, jx: np.sum( + (self._v[ix, :] - self._v[jx, :]) ** 2 + ), + ) + self._R = np.zeros((self._n_clusters, self._n_clusters)) + @_base._add_docs(_base._evaluate_doc) def _evaluate(self): """ diff --git a/src/cvi/modules/GD43.py b/src/cvi/modules/GD43.py index 388780a..7d57150 100644 --- a/src/cvi/modules/GD43.py +++ b/src/cvi/modules/GD43.py @@ -31,6 +31,16 @@ class GD43(_base.CVI): 5. J. C. Bezdek and N. R. Pal, "Some new indexes of cluster validity," IEEE Trans. Syst., Man, and Cybern., vol. 28, no. 3, pp. 301-315, Jun. 1998. """ + info = _base.CVIInfo( + name="Generalized Dunn's 43", + name_short="GD43", + index_min=0.0, + index_max=np.inf, + optimality="max" + ) + _supports_remove_merge = True + _uses_compactness_stats = True + def __init__(self): """ Generalized Dunn's Index 43 (GD43) initialization routine. @@ -159,17 +169,18 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): # Take the average across all samples, but cast to 1-D vector self._mu = np.mean(data, axis=0) - u = np.unique(labels) - self._n_clusters = u.size + u = self._setup_batch_labels(labels) + self._n_clusters = len(u) self._n = [0 for _ in range(self._n_clusters)] self._v = np.zeros((self._n_clusters, self._dim)) self._CP = [0.0 for _ in range(self._n_clusters)] self._G = np.zeros((self._n_clusters, self._dim)) self._D = np.zeros((self._n_clusters, self._n_clusters)) - for ix in range(self._n_clusters): + for ix, external_label in enumerate(u): subset_indices = ( - [x for x in range(len(labels)) if labels[x] == ix] + [x for x in range(len(labels)) + if labels[x] == external_label] ) subset = data[subset_indices, :] self._n[ix] = subset.shape[0] @@ -185,6 +196,26 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): self._D = self._D + np.transpose(self._D) + def _rebuild_after_operation(self): + """Rebuild centroid distances after a remove or merge.""" + + if self._n_clusters == 0: + self._mu = np.zeros(0) + self._D = np.zeros((0, 0)) + self._inter = 0.0 + self._intra = 0.0 + return + + self._D = self._pairwise_matrix( + self._n_clusters, + lambda ix, jx: np.linalg.norm( + self._v[ix, :] - self._v[jx, :] + ), + ) + if self._n_clusters < 2: + self._inter = 0.0 + self._intra = 0.0 + @_base._add_docs(_base._evaluate_doc) def _evaluate(self): """ diff --git a/src/cvi/modules/GD53.py b/src/cvi/modules/GD53.py index 4134a40..d08796c 100644 --- a/src/cvi/modules/GD53.py +++ b/src/cvi/modules/GD53.py @@ -31,6 +31,16 @@ class GD53(_base.CVI): 5. J. C. Bezdek and N. R. Pal, "Some new indexes of cluster validity," IEEE Trans. Syst., Man, and Cybern., vol. 28, no. 3, pp. 301-315, Jun. 1998. """ + info = _base.CVIInfo( + name="Generalized Dunn's 53", + name_short="GD53", + index_min=0.0, + index_max=np.inf, + optimality="max" + ) + _supports_remove_merge = True + _uses_compactness_stats = True + def __init__(self): """ Generalized Dunn's Index 53 (GD53) initialization routine. @@ -159,18 +169,19 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): # Take the average across all samples, but cast to 1-D vector self._mu = np.mean(data, axis=0) - u = np.unique(labels) - self._n_clusters = u.size + u = self._setup_batch_labels(labels) + self._n_clusters = len(u) self._n = [0 for _ in range(self._n_clusters)] self._v = np.zeros((self._n_clusters, self._dim)) self._CP = [0.0 for _ in range(self._n_clusters)] self._G = np.zeros((self._n_clusters, self._dim)) self._D = np.zeros((self._n_clusters, self._n_clusters)) - for ix in range(self._n_clusters): + for ix, external_label in enumerate(u): # subset_indices = lambda x: labels[x] == ix subset_indices = ( - [x for x in range(len(labels)) if labels[x] == ix] + [x for x in range(len(labels)) + if labels[x] == external_label] ) subset = data[subset_indices, :] self._n[ix] = subset.shape[0] @@ -186,6 +197,27 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): self._D = self._D + np.transpose(self._D) + def _rebuild_after_operation(self): + """Rebuild pairwise dispersion after a remove or merge.""" + + if self._n_clusters == 0: + self._mu = np.zeros(0) + self._D = np.zeros((0, 0)) + self._inter = 0.0 + self._intra = 0.0 + return + + self._D = self._pairwise_matrix( + self._n_clusters, + lambda ix, jx: ( + (self._CP[ix] + self._CP[jx]) + / (self._n[ix] + self._n[jx]) + ), + ) + if self._n_clusters < 2: + self._inter = 0.0 + self._intra = 0.0 + @_base._add_docs(_base._evaluate_doc) def _evaluate(self): """ diff --git a/src/cvi/modules/PS.py b/src/cvi/modules/PS.py index 83b9101..2484b43 100644 --- a/src/cvi/modules/PS.py +++ b/src/cvi/modules/PS.py @@ -25,6 +25,15 @@ class PS(_base.CVI): 2. E. Lughofer, "Extensions of vector quantization for incremental clustering," Pattern Recognit., vol. 41, no. 3, pp. 995-1011, 2008. """ + info = _base.CVIInfo( + name="Partition Separation", + name_short="PS", + index_min=0.0, + index_max=1.0, + optimality="max" + ) + _supports_remove_merge = True + def __init__(self): """ Partition Separation (PS) initialization routine. @@ -131,15 +140,16 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): # Take the average across all samples, but cast to 1-D vector self._mu = np.mean(data, axis=0) - u = np.unique(labels) - self._n_clusters = u.size + u = self._setup_batch_labels(labels) + self._n_clusters = len(u) self._n = [0 for _ in range(self._n_clusters)] self._v = np.zeros((self._n_clusters, self._dim)) self._D = np.zeros((self._n_clusters, self._n_clusters)) - for ix in range(self._n_clusters): + for ix, external_label in enumerate(u): subset_indices = ( - [x for x in range(len(labels)) if labels[x] == ix] + [x for x in range(len(labels)) + if labels[x] == external_label] ) subset = data[subset_indices, :] self._n[ix] = subset.shape[0] @@ -154,6 +164,75 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): self._D = self._D + np.transpose(self._D) + def _remove(self, sample: np.ndarray, label: int, i_label: int): + """Remove one sample from the PS centroid statistics.""" + + n_old = self._n[i_label] + v_old = self._v[i_label, :].copy() + n_samples_new = self._n_samples - 1 + + if n_old == 1: + self._validate_singleton_removal(sample, v_old) + + self._delete_cluster(label, i_label) + self._n_samples = n_samples_new + + if n_samples_new == 0: + self._clear_common_state() + + self._rebuild_after_operation() + return + + n_new = n_old - 1 + v_new = (n_old * v_old - sample) / n_new + + self._n[i_label] = n_new + self._v[i_label, :] = v_new + self._n_samples = n_samples_new + self._rebuild_after_operation() + + def _merge( + self, + target_label: int, + source_label: int, + target_i: int, + source_i: int, + ): + """Merge two PS centroid summaries.""" + + n_target = self._n[target_i] + n_source = self._n[source_i] + n_new = n_target + n_source + v_new = ( + n_target * self._v[target_i, :] + + n_source * self._v[source_i, :] + ) / n_new + + self._n[target_i] = n_new + self._v[target_i, :] = v_new + self._delete_cluster(source_label, source_i) + self._rebuild_after_operation() + + def _rebuild_after_operation(self): + """Rebuild pairwise centroid distances.""" + + if self._n_clusters == 0: + self._D = np.zeros((0, 0)) + self._v_bar = [] + self._beta_t = 0.0 + self._PS_i = np.zeros(0) + return + + self._D = self._pairwise_matrix( + self._n_clusters, + lambda ix, jx: np.sum( + (self._v[ix, :] - self._v[jx, :]) ** 2 + ), + ) + self._v_bar = [] + self._beta_t = 0.0 + self._PS_i = np.zeros(self._n_clusters) + @_base._add_docs(_base._evaluate_doc) def _evaluate(self): """ diff --git a/src/cvi/modules/WB.py b/src/cvi/modules/WB.py index d1694eb..0f5cb24 100644 --- a/src/cvi/modules/WB.py +++ b/src/cvi/modules/WB.py @@ -31,6 +31,16 @@ class WB(_base.CVI): 5. M. Moshtaghi, J. C. Bezdek, S. M. Erfani, C. Leckie, J. Bailey, "Online cluster validity indices for performance monitoring of streaming data clustering," Int. J. Intell. Syst., pp. 1-23, 2018. """ + info = _base.CVIInfo( + name="Within/Between", + name_short="WB", + index_min=0.0, + index_max=np.inf, + optimality="min" + ) + _supports_remove_merge = True + _uses_compactness_stats = True + def __init__(self): """ WB initialization routine. @@ -141,16 +151,18 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): # Take the average across all samples, but cast to 1-D vector self._mu = np.mean(data, axis=0) - u = np.unique(labels) - self._n_clusters = u.size - self._n = np.zeros(self._n_clusters, dtype=int) + u = self._setup_batch_labels(labels) + self._n_clusters = len(u) + self._n = [0 for _ in range(self._n_clusters)] self._v = np.zeros((self._n_clusters, self._dim)) - self._CP = np.zeros(self._n_clusters) + self._CP = [0.0 for _ in range(self._n_clusters)] + self._G = np.zeros((self._n_clusters, self._dim)) self._SEP = np.zeros(self._n_clusters) - for ix in range(self._n_clusters): + for ix, external_label in enumerate(u): subset_indices = ( - [x for x in range(len(labels)) if labels[x] == ix] + [x for x in range(len(labels)) + if labels[x] == external_label] ) subset = data[subset_indices, :] self._n[ix] = subset.shape[0] @@ -159,6 +171,23 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): self._CP[ix] = np.sum(diff_x_v ** 2) self._SEP[ix] = self._n[ix] * np.sum((self._v[ix, :] - self._mu) ** 2) + def _rebuild_after_operation(self): + """Rebuild separation statistics after a remove or merge.""" + + if self._n_clusters == 0: + self._mu = np.zeros(0) + self._SEP = np.zeros(0) + self._BGSS = 0.0 + self._WGSS = 0.0 + return + + self._SEP = np.asarray([ + self._n[ix] * np.sum((self._v[ix, :] - self._mu) ** 2) + for ix in range(self._n_clusters) + ]) + self._WGSS = sum(self._CP) + self._BGSS = sum(self._SEP) + @_base._add_docs(_base._evaluate_doc) def _evaluate(self): """ diff --git a/src/cvi/modules/XB.py b/src/cvi/modules/XB.py index 53dea02..40a5a45 100644 --- a/src/cvi/modules/XB.py +++ b/src/cvi/modules/XB.py @@ -27,6 +27,16 @@ class XB(_base.CVI): 3. M. Moshtaghi, J. C. Bezdek, S. M. Erfani, C. Leckie, J. Bailey, "Online cluster validity indices for performance monitoring of streaming data clustering," Int. J. Intell. Syst., pp. 1-23, 2018. """ + info = _base.CVIInfo( + name="Xie-Beni", + name_short="XB", + index_min=0.0, + index_max=np.inf, + optimality="min" + ) + _supports_remove_merge = True + _uses_compactness_stats = True + def __init__(self): """ XB initialization routine. @@ -157,16 +167,18 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): # Take the average across all samples, but cast to 1-D vector self._mu = np.mean(data, axis=0) - u = np.unique(labels) - self._n_clusters = u.size - self._n = np.zeros(self._n_clusters, dtype=int) + u = self._setup_batch_labels(labels) + self._n_clusters = len(u) + self._n = [0 for _ in range(self._n_clusters)] self._v = np.zeros((self._n_clusters, self._dim)) - self._CP = np.zeros(self._n_clusters) + self._CP = [0.0 for _ in range(self._n_clusters)] + self._G = np.zeros((self._n_clusters, self._dim)) self._D = np.zeros((self._n_clusters, self._n_clusters)) - for ix in range(self._n_clusters): + for ix, external_label in enumerate(u): subset_indices = ( - [x for x in range(len(labels)) if labels[x] == ix] + [x for x in range(len(labels)) + if labels[x] == external_label] ) subset = data[subset_indices, :] self._n[ix] = subset.shape[0] @@ -180,6 +192,28 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): np.sum((self._v[ix, :] - self._v[jx, :]) ** 2) ) + self._D = self._D + np.transpose(self._D) + + def _rebuild_after_operation(self): + """Rebuild centroid distances after a remove or merge.""" + + if self._n_clusters == 0: + self._mu = np.zeros(0) + self._SEP = np.zeros(0) + self._D = np.zeros((0, 0)) + self._WGSS = 0.0 + return + + self._D = self._pairwise_matrix( + self._n_clusters, + lambda ix, jx: np.sum( + (self._v[ix, :] - self._v[jx, :]) ** 2 + ), + ) + self._WGSS = sum(self._CP) + if self._n_clusters < 2: + self._SEP = 0.0 + @_base._add_docs(_base._evaluate_doc) def _evaluate(self): """ diff --git a/src/cvi/modules/__init__.py b/src/cvi/modules/__init__.py index 1293e6f..4efb380 100644 --- a/src/cvi/modules/__init__.py +++ b/src/cvi/modules/__init__.py @@ -14,6 +14,7 @@ ) from .CH import CH +from .CONN import CONN from .cSIL import cSIL from .DB import DB from .GD43 import GD43 @@ -30,6 +31,7 @@ "_param_inc_doc", "_param_batch_doc", "CH", + "CONN", "cSIL", "DB", "GD43", diff --git a/src/cvi/modules/_base.py b/src/cvi/modules/_base.py index e931411..86f7824 100644 --- a/src/cvi/modules/_base.py +++ b/src/cvi/modules/_base.py @@ -12,6 +12,9 @@ ) from abc import abstractmethod +from dataclasses import dataclass +from typing import ClassVar + # Custom imports import numpy as np @@ -19,6 +22,13 @@ # CLASSES # --------------------------------------------------------------------------- # +@dataclass +class CVIInfo: + name: str + name_short: str + index_min: float + index_max: float + optimality: str class LabelMap(): """ @@ -47,12 +57,50 @@ def get_internal_label(self, label: int) -> int: return internal_label + def get_existing_label(self, label: int) -> int: + """ + Gets an existing internal label without modifying the label map. + + Raises + ------ + ValueError + If the external label is not present in the map. + """ + + if label not in self.map: + raise ValueError(f"Unknown cluster label: {label}") + + return self.map[label] + + def remove_label(self, label: int) -> int: + """ + Removes a label and compacts all following internal labels. + + Returns + ------- + int + The removed internal label. + """ + + internal_label = self.get_existing_label(label) + del self.map[label] + + for external_label, mapped_label in self.map.items(): + if mapped_label > internal_label: + self.map[external_label] = mapped_label - 1 + + return internal_label + class CVI(): """ Superclass containing elements shared between all CVIs. """ + info: ClassVar[CVIInfo] + _supports_remove_merge: ClassVar[bool] = False + _uses_compactness_stats: ClassVar[bool] = False + def __init__(self): """ CVI base class initialization method. @@ -103,6 +151,19 @@ def _setup_batch(self, data: np.ndarray): self._n_samples, self._dim = data.shape self._is_setup = True + def _setup_batch_labels(self, labels: np.ndarray): + """Populate the label map and return labels in first-seen order.""" + + self._label_map = LabelMap() + unique_labels = [] + for label in np.asarray(labels): + external_label = label.item() if hasattr(label, "item") else label + if external_label not in self._label_map.map: + self._label_map.get_internal_label(external_label) + unique_labels.append(external_label) + + return unique_labels + @abstractmethod def _param_inc(self, sample: np.ndarray, label: int): raise NotImplementedError @@ -115,6 +176,265 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): def _evaluate(self): raise NotImplementedError + def _require_operations(self): + """Validate that structural operations are supported and available.""" + + if not self._supports_remove_merge: + raise NotImplementedError( + f"{type(self).__name__} does not support remove or merge" + ) + + if not self._is_setup: + raise ValueError( + "Remove and merge require an initialized CVI" + ) + + def _validate_sample(self, sample: np.ndarray) -> np.ndarray: + """Validate and normalize a sample used by a structural operation.""" + + sample = np.asarray(sample, dtype=float) + + if sample.ndim != 1: + raise ValueError("Remove requires a one-dimensional sample") + + if sample.shape[0] != self._dim: + raise ValueError( + f"Expected a sample with {self._dim} features, " + f"received {sample.shape[0]}" + ) + + if not np.all(np.isfinite(sample)): + raise ValueError("Remove requires a sample containing finite values") + + return sample + + @staticmethod + def _nonnegative_or_error(value: float, scale: float, name: str) -> float: + """Clip floating-point noise or reject a materially negative statistic.""" + + tolerance = 1e-10 * max(1.0, abs(scale)) + if value < -tolerance: + raise ValueError( + f"The requested operation produces invalid {name}; " + "check the supplied sample and cluster label" + ) + + return max(0.0, float(value)) + + @staticmethod + def _validate_singleton_removal( + sample: np.ndarray, + centroid: np.ndarray, + ): + """Validate that a removed sample matches a singleton centroid.""" + + if not np.allclose(sample, centroid, rtol=1e-10, atol=1e-12): + raise ValueError( + "The supplied sample does not match the singleton cluster" + ) + + @staticmethod + def _delete_vector_entry(values, index: int): + """Delete one entry from either a list or a NumPy vector.""" + + if isinstance(values, list): + del values[index] + return values + + return np.delete(values, index) + + @staticmethod + def _pairwise_matrix(n_clusters: int, measure: Callable) -> np.ndarray: + """Build a symmetric pairwise cluster matrix.""" + + matrix = np.zeros((n_clusters, n_clusters)) + for ix in range(n_clusters - 1): + for jx in range(ix + 1, n_clusters): + value = measure(ix, jx) + matrix[ix, jx] = value + matrix[jx, ix] = value + + return matrix + + def _delete_cluster(self, label: int, i_label: int): + """Delete universally shared state for one cluster.""" + + self._n = self._delete_vector_entry(self._n, i_label) + self._v = np.delete(self._v, i_label, axis=0) + self._label_map.remove_label(label) + self._n_clusters -= 1 + + def _delete_common_cluster(self, label: int, i_label: int): + """Delete a compactness-based cluster and compact its internal label.""" + + self._CP = self._delete_vector_entry(self._CP, i_label) + self._G = np.delete(self._G, i_label, axis=0) + self._delete_cluster(label, i_label) + + def _clear_common_state(self): + """Return the common CVI state to its pre-initialization values.""" + + self._label_map = LabelMap() + self._dim = 0 + self._n_samples = 0 + self._n = [] + self._v = np.zeros([0, 0]) + self._CP = [] + self._G = np.zeros([0, 0]) + self._n_clusters = 0 + self.criterion_value = 0.0 + self._is_setup = False + + def _rebuild_after_operation(self): + """Rebuild CVI-specific derived state after remove or merge.""" + + raise NotImplementedError + + def _remove(self, sample: np.ndarray, label: int, i_label: int): + """Remove a sample from a compactness-based CVI.""" + + if not self._uses_compactness_stats: + raise NotImplementedError + + n_old = self._n[i_label] + v_old = self._v[i_label, :].copy() + n_samples_new = self._n_samples - 1 + + if n_old == 1: + self._validate_singleton_removal(sample, v_old) + + mu_new = None + if n_samples_new > 0: + mu_new = ( + self._n_samples * self._mu - sample + ) / n_samples_new + + self._delete_common_cluster(label, i_label) + self._n_samples = n_samples_new + + if n_samples_new == 0: + self._clear_common_state() + else: + self._mu = mu_new + + self._rebuild_after_operation() + return + + n_new = n_old - 1 + v_new = (n_old * v_old - sample) / n_new + distance = float(np.inner(sample - v_old, sample - v_old)) + correction = (n_old / n_new) * distance + CP_new = self._nonnegative_or_error( + self._CP[i_label] - correction, + max(abs(self._CP[i_label]), correction), + "cluster compactness", + ) + mu_new = ( + self._n_samples * self._mu - sample + ) / n_samples_new + + self._n[i_label] = n_new + self._v[i_label, :] = v_new + self._CP[i_label] = CP_new + self._G[i_label, :] = np.zeros(self._dim) + self._n_samples = n_samples_new + self._mu = mu_new + self._rebuild_after_operation() + + def _merge( + self, + target_label: int, + source_label: int, + target_i: int, + source_i: int, + ): + """Merge two clusters in a compactness-based CVI.""" + + if not self._uses_compactness_stats: + raise NotImplementedError + + n_target = self._n[target_i] + n_source = self._n[source_i] + n_new = n_target + n_source + v_target = self._v[target_i, :].copy() + v_source = self._v[source_i, :].copy() + v_new = (n_target * v_target + n_source * v_source) / n_new + difference = v_source - v_target + CP_new = ( + self._CP[target_i] + + self._CP[source_i] + + (n_target * n_source / n_new) + * np.inner(difference, difference) + ) + + if not np.isfinite(CP_new): + raise ValueError("The requested merge produces invalid compactness") + + self._n[target_i] = n_new + self._v[target_i, :] = v_new + self._CP[target_i] = max(0.0, float(CP_new)) + self._G[target_i, :] = np.zeros(self._dim) + self._delete_common_cluster(source_label, source_i) + self._rebuild_after_operation() + + def remove(self, sample: np.ndarray, label: int) -> float: + """ + Remove a sample from an initialized CVI. + + The caller is responsible for ensuring that the sample belongs to the + supplied cluster label. If the sample is the cluster's final member, + the empty cluster and its label are removed. + + Parameters + ---------- + sample : numpy.ndarray + One sample vector of features. + label : int + External label of the cluster containing the sample. + + Returns + ------- + float + The updated CVI criterion value. + """ + + self._require_operations() + sample = self._validate_sample(sample) + i_label = self._label_map.get_existing_label(label) + self._remove(sample, label, i_label) + self._evaluate() + return self.criterion_value + + def merge(self, target_label: int, source_label: int) -> float: + """ + Merge a source cluster into a target cluster. + + The target external label is retained and the source label is removed. + + Parameters + ---------- + target_label : int + External label of the cluster that remains after the merge. + source_label : int + External label of the cluster merged into the target. + + Returns + ------- + float + The updated CVI criterion value. + """ + + self._require_operations() + + if target_label == source_label: + raise ValueError("Merge requires two different cluster labels") + + target_i = self._label_map.get_existing_label(target_label) + source_i = self._label_map.get_existing_label(source_label) + self._merge(target_label, source_label, target_i, source_i) + self._evaluate() + return self.criterion_value + def get_cvi(self, data: np.ndarray, label: Union[int, np.ndarray]) -> float: """ Updates the CVI parameters and then evaluates and returns the criterion value. @@ -136,6 +456,11 @@ def get_cvi(self, data: np.ndarray, label: Union[int, np.ndarray]) -> float: # If we got 1D data, do a quick update if (data.ndim == 1): + if self._is_setup and data.shape[0] != self._dim: + raise ValueError( + f"Expected a sample with {self._dim} features, " + f"received {data.shape[0]}" + ) self._param_inc(data, label) # Otherwise, we got 2D data and do the correct update @@ -153,18 +478,12 @@ def get_cvi(self, data: np.ndarray, label: Union[int, np.ndarray]) -> float: # Do a batch update self._param_batch(data, label) - # Otherwise, we are already setup + # Otherwise, a second batch update was requested else: - - # Error until batch to incremental is supported raise ValueError( - "Switching from batch to incremental not supported" + "Repeated batch updates are not supported" ) - # Do many incremental updates - # for ix in range(len(label)): - # self._param_inc(data[ix, :], label[ix]) - # Otherwise, we got incorrectly dimensioned data else: diff --git a/src/cvi/modules/cSIL.py b/src/cvi/modules/cSIL.py index 93c0911..813ef99 100644 --- a/src/cvi/modules/cSIL.py +++ b/src/cvi/modules/cSIL.py @@ -27,6 +27,15 @@ class cSIL(_base.CVI): 3. M. Rawashdeh and A. Ralescu, "Center-wise intra-inter silhouettes," in Scalable Uncertainty Management, E. Hüllermeier, S. Link, T. Fober et al., Eds. Berlin, Heidelberg: Springer, 2012, pp. 406-419. """ + info = _base.CVIInfo( + name="Centroid-based Silhouette", + name_short="cSIL", + index_min=-1.0, + index_max=1.0, + optimality="max" + ) + _supports_remove_merge = True + def __init__(self): """ Centroid-based Silhouette (cSIL) initialization routine. @@ -81,20 +90,18 @@ def _param_inc(self, sample: np.ndarray, label: int): S_row_new = np.zeros(self._n_clusters + 1) S_col_new = np.zeros(self._n_clusters + 1) for cl in range(self._n_clusters): - # Column "bmu_temp - D_new" - C = ( + # Dissimilarity of the new cluster to an old centroid. + S_col_new[cl] = ( CP_new + np.inner(self._v[cl, :], self._v[cl, :]) - - np.inner(G_new, self._v[cl, :]) + - 2 * np.inner(G_new, self._v[cl, :]) ) - S_col_new[cl] = C - C = ( + # Dissimilarity of an old cluster to the new centroid. + S_row_new[cl] = ( self._CP[cl] + self._n[cl] * np.inner(v_new, v_new) - 2 * np.inner(self._G[cl, :], v_new) - ) - S_row_new[cl] = C / self._n[cl] - # Column "ind_minus" - F + ) / self._n[cl] S_col_new[i_label] = 0 S_row_new[i_label] = S_col_new[i_label] S_new[i_label, :] = S_col_new @@ -131,36 +138,25 @@ def _param_inc(self, sample: np.ndarray, label: int): S_row_new = np.zeros(self._n_clusters) S_col_new = np.zeros(self._n_clusters) for cl in range(self._n_clusters): - # Skip the i_label iteration - if cl == i_label: - continue - # Column "bmu_temp" - D_new - diff_x_v = sample - self._v[cl, :] - C = ( - self._CP[i_label] - + np.inner(diff_x_v, diff_x_v) - + self._n[i_label] * np.inner(self._v[cl, :], self._v[cl, :]) - - 2 * np.inner(G_new, self._v[cl, :]) - ) - S_col_new[cl] = C / n_new - # Row "bmu_temp" - E - C = ( + centroid = v_new if cl == i_label else self._v[cl, :] + S_col_new[cl] = ( + CP_new + + n_new * np.inner(centroid, centroid) + - 2 * np.inner(G_new, centroid) + ) / n_new + S_row_new[cl] = ( self._CP[cl] + self._n[cl] * np.inner(v_new, v_new) - 2 * np.inner(self._G[cl, :], v_new) - ) - S_row_new[cl] = C / self._n[cl] + ) / self._n[cl] - # Column "ind_minus" - F - diff_x_v = sample - v_new - C = ( - self._CP[i_label] - + np.inner(diff_x_v, diff_x_v) - + self._n[i_label] * np.inner(v_new, v_new) - - 2 * np.inner(self._G[i_label, :], v_new) - ) - S_col_new[i_label] = C / n_new - S_row_new[i_label] = S_col_new[i_label] + diagonal = ( + CP_new + + n_new * np.inner(v_new, v_new) + - 2 * np.inner(G_new, v_new) + ) / n_new + S_col_new[i_label] = diagonal + S_row_new[i_label] = diagonal # Update parameters self._n[i_label] = n_new @@ -186,34 +182,125 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): super()._setup_batch(data) # Take the average across all samples, but cast to 1-D vector - u = np.unique(labels) - self._n_clusters = u.size - self._n = np.zeros(self._n_clusters, dtype=int) + u = self._setup_batch_labels(labels) + self._n_clusters = len(u) + self._n = [0 for _ in range(self._n_clusters)] self._v = np.zeros((self._n_clusters, self._dim)) - self._CP = np.zeros(self._n_clusters) + self._CP = [0.0 for _ in range(self._n_clusters)] + self._G = np.zeros((self._n_clusters, self._dim)) self._S = np.zeros((self._n_clusters, self._n_clusters)) - D = np.zeros((self._n_samples, self._n_samples)) - for ix in range(self._n_clusters): + D = np.zeros((self._n_clusters, self._n_samples)) + for ix, external_label in enumerate(u): subset_indices = ( - [x for x in range(len(labels)) if labels[x] == ix] + [x for x in range(len(labels)) + if labels[x] == external_label] ) subset = data[subset_indices, :] self._n[ix] = subset.shape[0] self._v[ix, :] = np.mean(subset, axis=0) - # Compute CP in case of switching back to incremental mode - diff_x_v = subset - self._v[ix, :] * np.ones((self._n[ix], 1)) - self._CP[ix] = np.sum(diff_x_v ** 2) + # Retain zero-centered raw moments for subsequent updates. + self._CP[ix] = np.sum(subset ** 2) + self._G[ix, :] = np.sum(subset, axis=0) d_temp = (data - self._v[ix, :] * np.ones((self._n_samples, 1))) ** 2 D[ix, :] = np.transpose(np.sum(d_temp, axis=1)) # D[ix, :] = np.sum(d_temp, axis=1) for ix in range(self._n_clusters): - for jx in range(self._n_clusters): - subset_ind = [x for x in range(len(labels)) if labels[x] == jx] + for jx, external_label in enumerate(u): + subset_ind = [ + x for x in range(len(labels)) + if labels[x] == external_label + ] self._S[jx, ix] = sum(D[ix, subset_ind]) / self._n[jx] + def _delete_cluster(self, label: int, i_label: int): + """Delete one cSIL cluster and compact its internal label.""" + + self._CP = self._delete_vector_entry(self._CP, i_label) + self._G = np.delete(self._G, i_label, axis=0) + super()._delete_cluster(label, i_label) + + def _remove(self, sample: np.ndarray, label: int, i_label: int): + """Remove a sample from cSIL's zero-centered raw moments.""" + + n_old = self._n[i_label] + v_old = self._v[i_label, :].copy() + n_samples_new = self._n_samples - 1 + + if n_old == 1: + self._validate_singleton_removal(sample, v_old) + + self._delete_cluster(label, i_label) + self._n_samples = n_samples_new + + if n_samples_new == 0: + self._clear_common_state() + + self._rebuild_after_operation() + return + + n_new = n_old - 1 + G_new = self._G[i_label, :] - sample + v_new = G_new / n_new + raw_CP_new = self._CP[i_label] - np.inner(sample, sample) + centered_CP_new = raw_CP_new - n_new * np.inner(v_new, v_new) + centered_CP_new = self._nonnegative_or_error( + centered_CP_new, + max(abs(self._CP[i_label]), abs(raw_CP_new)), + "cluster compactness", + ) + raw_CP_new = centered_CP_new + n_new * np.inner(v_new, v_new) + + self._n[i_label] = n_new + self._v[i_label, :] = v_new + self._CP[i_label] = raw_CP_new + self._G[i_label, :] = G_new + self._n_samples = n_samples_new + self._rebuild_after_operation() + + def _merge( + self, + target_label: int, + source_label: int, + target_i: int, + source_i: int, + ): + """Merge two cSIL raw-moment summaries.""" + + n_new = self._n[target_i] + self._n[source_i] + G_new = self._G[target_i, :] + self._G[source_i, :] + CP_new = self._CP[target_i] + self._CP[source_i] + + self._n[target_i] = n_new + self._v[target_i, :] = G_new / n_new + self._CP[target_i] = CP_new + self._G[target_i, :] = G_new + self._delete_cluster(source_label, source_i) + self._rebuild_after_operation() + + def _rebuild_after_operation(self): + """Rebuild the centroid-to-cluster dissimilarity matrix.""" + + if self._n_clusters == 0: + self._S = np.empty((0, 0)) + self._sil_coefs = [] + return + + self._S = np.zeros((self._n_clusters, self._n_clusters)) + for cluster_i in range(self._n_clusters): + for centroid_i in range(self._n_clusters): + value = ( + self._CP[cluster_i] + + self._n[cluster_i] + * np.inner(self._v[centroid_i, :], self._v[centroid_i, :]) + - 2 + * np.inner(self._G[cluster_i, :], self._v[centroid_i, :]) + ) / self._n[cluster_i] + self._S[cluster_i, centroid_i] = max(0.0, float(value)) + self._sil_coefs = np.zeros(self._n_clusters) + @_base._add_docs(_base._evaluate_doc) def _evaluate(self): """ diff --git a/src/cvi/modules/rCIP.py b/src/cvi/modules/rCIP.py index e9bf5a4..333f8f9 100644 --- a/src/cvi/modules/rCIP.py +++ b/src/cvi/modules/rCIP.py @@ -27,6 +27,15 @@ class rCIP(_base.CVI): 3. M. Moshtaghi, J. C. Bezdek, S. M. Erfani, C. Leckie, J. Bailey, "Online cluster validity indices for performance monitoring of streaming data clustering," Int. J. Intell. Syst., pp. 1-23, 2018. """ + info = _base.CVIInfo( + name="Representative Cross Information Potential", + name_short="rCIP", + index_min=0.0, + index_max=np.inf, + optimality="min" + ) + _supports_remove_merge = True + def __init__(self): """ (Renyi's) representative Cross Information Potential (rCIP) initialization routine. @@ -166,16 +175,18 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): self._constant = 1 / np.sqrt((2 * np.pi) ** self._dim) # Take the average across all samples, but cast to 1-D vector - u = np.unique(labels) - self._n_clusters = u.size + u = self._setup_batch_labels(labels) + self._n_clusters = len(u) self._n = [0 for _ in range(self._n_clusters)] self._v = np.zeros((self._n_clusters, self._dim)) + self._G = np.zeros((0, self._dim)) self._sigma = np.zeros((self._dim, self._dim, self._n_clusters)) self._D = np.zeros((self._n_clusters, self._n_clusters)) - for ix in range(self._n_clusters): + for ix, external_label in enumerate(u): subset_indices = ( - [x for x in range(len(labels)) if labels[x] == ix] + [x for x in range(len(labels)) + if labels[x] == external_label] ) subset = data[subset_indices, :] self._n[ix] = subset.shape[0] @@ -204,6 +215,136 @@ def _param_batch(self, data: np.ndarray, labels: np.ndarray): self._D = self._D + np.transpose(self._D) + @staticmethod + def _stabilize_covariance(covariance: np.ndarray) -> np.ndarray: + """Symmetrize covariance and clip insignificant negative eigenvalues.""" + + covariance = (covariance + covariance.T) / 2 + eigenvalues, eigenvectors = np.linalg.eigh(covariance) + tolerance = 1e-10 * max(1.0, np.linalg.norm(covariance, ord=2)) + + if np.min(eigenvalues) < -tolerance: + raise ValueError( + "The requested operation produces invalid covariance; " + "check the supplied sample and cluster label" + ) + + eigenvalues = np.maximum(eigenvalues, 0.0) + return (eigenvectors * eigenvalues) @ eigenvectors.T + + def _delete_cluster(self, label: int, i_label: int): + """Delete one rCIP cluster and compact its internal label.""" + + self._sigma = np.delete(self._sigma, i_label, axis=2) + super()._delete_cluster(label, i_label) + + def _remove(self, sample: np.ndarray, label: int, i_label: int): + """Remove a sample from rCIP's mean and covariance statistics.""" + + n_old = self._n[i_label] + v_old = self._v[i_label, :].copy() + n_samples_new = self._n_samples - 1 + + if n_old == 1: + self._validate_singleton_removal(sample, v_old) + + self._delete_cluster(label, i_label) + self._n_samples = n_samples_new + + if n_samples_new == 0: + self._clear_common_state() + self._CP = None + self._D = np.zeros((0, 0)) + self._sigma = np.zeros((0, 0, 0)) + self._delta_term = np.zeros((0, 0)) + self._constant = 0.0 + + self._rebuild_after_operation() + return + + n_new = n_old - 1 + v_new = (n_old * v_old - sample) / n_new + + if n_new == 1: + sigma_new = self._delta_term.copy() + else: + covariance_old = ( + self._sigma[:, :, i_label] - self._delta_term + ) + difference = sample - v_old + covariance_new = ( + ((n_old - 1) / (n_old - 2)) * covariance_old + - (n_old / ((n_old - 1) * (n_old - 2))) + * np.outer(difference, difference) + ) + covariance_new = self._stabilize_covariance(covariance_new) + sigma_new = covariance_new + self._delta_term + + self._n[i_label] = n_new + self._v[i_label, :] = v_new + self._sigma[:, :, i_label] = sigma_new + self._n_samples = n_samples_new + self._rebuild_after_operation() + + def _merge( + self, + target_label: int, + source_label: int, + target_i: int, + source_i: int, + ): + """Merge two rCIP mean and covariance summaries.""" + + n_target = self._n[target_i] + n_source = self._n[source_i] + n_new = n_target + n_source + v_target = self._v[target_i, :].copy() + v_source = self._v[source_i, :].copy() + v_new = (n_target * v_target + n_source * v_source) / n_new + covariance_target = self._sigma[:, :, target_i] - self._delta_term + covariance_source = self._sigma[:, :, source_i] - self._delta_term + difference = v_source - v_target + covariance_new = ( + ((n_target - 1) / (n_new - 1)) * covariance_target + + ((n_source - 1) / (n_new - 1)) * covariance_source + + (n_target * n_source / (n_new * (n_new - 1))) + * np.outer(difference, difference) + ) + covariance_new = self._stabilize_covariance(covariance_new) + sigma_new = covariance_new + self._delta_term + + self._n[target_i] = n_new + self._v[target_i, :] = v_new + self._sigma[:, :, target_i] = sigma_new + self._delete_cluster(source_label, source_i) + self._rebuild_after_operation() + + def _rebuild_after_operation(self): + """Rebuild pairwise representative information potentials.""" + + if self._n_clusters == 0: + self._D = np.zeros((0, 0)) + return + + def information_potential(ix, jx): + difference = self._v[ix, :] - self._v[jx, :] + sigma_q = self._sigma[:, :, ix] + self._sigma[:, :, jx] + return ( + self._constant + * (1 / np.sqrt(np.linalg.det(sigma_q))) + * np.exp( + -0.5 + * difference + @ np.linalg.inv(sigma_q) + @ difference + ) + ) + + self._D = self._pairwise_matrix( + self._n_clusters, + information_potential, + ) + @_base._add_docs(_base._evaluate_doc) def _evaluate(self): """ diff --git a/tests/test_conn.py b/tests/test_conn.py new file mode 100644 index 0000000..daf2a12 --- /dev/null +++ b/tests/test_conn.py @@ -0,0 +1,237 @@ +"""Focused tests for the CONN validity index.""" + +import numpy as np +import pytest + +import src.cvi as cvi + + +KMEANS_KWARGS = {"random_state": 0, "n_init": 1} + + +def _separated_data(): + data = np.asarray( + [ + [0.0, 0.0], + [0.1, 0.0], + [10.0, 10.0], + [10.1, 10.0], + ] + ) + labels = np.asarray([10, 10, 20, 20]) + return data, labels + + +@pytest.mark.parametrize("model_type", ["KMeans", "MiniBatchKMeans"]) +def test_centroid_backends_compute_expected_connectivity(model_type): + data, labels = _separated_data() + conn = cvi.CONN( + model_type=model_type, + kmeans_k=2, + kmeans_kwargs=KMEANS_KWARGS, + ) + + value = conn.get_cvi(data, labels) + + expected_cadj = np.asarray( + [ + [0.0, 1.0, 0.0, 0.0], + [1.0, 0.0, 0.0, 0.0], + [0.0, 0.0, 0.0, 1.0], + [0.0, 0.0, 1.0, 0.0], + ] + ) + assert value == pytest.approx(1.0) + np.testing.assert_array_equal(conn._CADJ.asarray(), expected_cadj) + np.testing.assert_array_equal( + conn._CONN.asarray(), + expected_cadj + expected_cadj.T, + ) + np.testing.assert_array_equal(conn._INTRA.asarray(), [1.0, 1.0]) + np.testing.assert_array_equal(conn._INTER.asarray(), np.zeros((2, 2))) + + +@pytest.mark.parametrize("model_type", ["KMeans", "MiniBatchKMeans"]) +def test_centroid_backends_detect_cross_label_connectivity(model_type): + data = np.asarray( + [ + [0.0, 0.0], + [10.0, 10.0], + [0.1, 0.0], + [10.1, 10.0], + ] + ) + labels = np.asarray([10, 10, 20, 20]) + conn = cvi.CONN( + model_type=model_type, + kmeans_k=2, + kmeans_kwargs=KMEANS_KWARGS, + ) + + assert conn.get_cvi(data, labels) == pytest.approx(0.0) + np.testing.assert_array_equal(conn._INTRA.asarray(), [0.0, 0.0]) + np.testing.assert_array_equal( + conn._INTER.asarray(), + [[0.0, 1.0], [1.0, 0.0]], + ) + + +def test_kmeans_counts_are_per_external_label_and_capped_by_support(): + data = np.asarray( + [ + [0.0, 0.0], + [0.1, 0.0], + [10.0, 10.0], + [10.1, 10.0], + [10.2, 10.0], + ] + ) + labels = np.asarray([10, 10, 20, 20, 20]) + conn = cvi.CONN( + model_type="KMeans", + kmeans_k={10: 8, 20: 1}, + kmeans_kwargs=KMEANS_KWARGS, + ) + + conn.get_cvi(data, labels) + + assert conn._kmeans_models[10].n_clusters == 2 + assert conn._kmeans_models[20].n_clusters == 1 + assert len(conn._rev_map[conn._label_map.map[10]]) == 2 + assert len(conn._rev_map[conn._label_map.map[20]]) == 1 + assert conn._cluster_centers.shape == (3, 2) + + +@pytest.mark.parametrize("model_type", ["KMeans", "MiniBatchKMeans"]) +def test_kmeans_kwargs_are_forwarded(model_type): + data, labels = _separated_data() + conn = cvi.CONN( + model_type=model_type, + kmeans_k=2, + kmeans_kwargs={"random_state": 17, "n_init": 1}, + ) + + conn.get_cvi(data, labels) + + for model in conn._kmeans_models.values(): + assert model.random_state == 17 + assert model.n_init == 1 + + +@pytest.mark.parametrize("model_type", ["KMeans", "MiniBatchKMeans"]) +def test_batch_normalization_is_affine_scale_invariant(model_type): + data, labels = _separated_data() + scaled_data = data * np.asarray([3.0, 7.0]) + np.asarray([-4.0, 20.0]) + kwargs = { + "model_type": model_type, + "kmeans_k": 2, + "kmeans_kwargs": KMEANS_KWARGS, + } + + conn = cvi.CONN(**kwargs) + scaled_conn = cvi.CONN(**kwargs) + value = conn.get_cvi(data, labels) + scaled_value = scaled_conn.get_cvi(scaled_data, labels) + + assert scaled_value == pytest.approx(value) + np.testing.assert_array_equal( + scaled_conn._CADJ.asarray(), + conn._CADJ.asarray(), + ) + + +@pytest.mark.parametrize("model_type", ["KMeans", "MiniBatchKMeans"]) +def test_centroid_backends_reject_incremental_updates_without_mutation( + model_type, +): + conn = cvi.CONN(model_type=model_type) + + with pytest.raises(ValueError, match="supports batch mode only"): + conn.get_cvi(np.asarray([0.0, 0.0]), 0) + + assert conn._is_setup is False + assert conn._n_samples == 0 + assert conn._label_map.map == {} + + +def test_fuzzy_backend_retains_batch_and_incremental_behavior(): + data = np.asarray( + [[0.0, 0.0], [0.0, 1.0], [1.0, 0.0], [1.0, 1.0]] + ) + labels = np.asarray([0, 0, 1, 1]) + incremental = cvi.CONN(model_type="Fuzzy", normalize_batch=False) + batch = cvi.CONN(model_type="Fuzzy", normalize_batch=False) + + for sample, label in zip(data, labels): + incremental.get_cvi(sample, label) + + batch_value = batch.get_cvi(data, labels) + + assert incremental.criterion_value == pytest.approx(0.125) + assert batch_value == pytest.approx(incremental.criterion_value) + np.testing.assert_array_equal( + batch._CADJ.asarray(), + incremental._CADJ.asarray(), + ) + + +@pytest.mark.parametrize( + "kwargs, message", + [ + ({"model_type": "unknown"}, "model_type must be one of"), + ({"kmeans_k": 0}, "must be a positive integer"), + ({"kmeans_k": False}, "must be a positive integer"), + ({"kmeans_k": {0: 0}}, "must be a positive integer"), + ({"kmeans_k": {"0": 1}}, "dictionary keys must be integers"), + ({"kmeans_kwargs": []}, "must be a dictionary or None"), + ( + {"kmeans_kwargs": {"n_clusters": 2}}, + "Configure n_clusters through kmeans_k", + ), + ], +) +def test_invalid_backend_parameters_are_rejected(kwargs, message): + with pytest.raises(ValueError, match=message): + cvi.CONN(**kwargs) + + +def test_kmeans_dictionary_must_cover_observed_labels(): + data, labels = _separated_data() + conn = cvi.CONN( + model_type="KMeans", + kmeans_k={10: 2}, + kmeans_kwargs=KMEANS_KWARGS, + ) + + with pytest.raises(ValueError, match="missing values for labels.*20"): + conn.get_cvi(data, labels) + + +@pytest.mark.filterwarnings("ignore:Number of distinct clusters") +@pytest.mark.parametrize("model_type", ["KMeans", "MiniBatchKMeans"]) +def test_duplicate_centers_retain_full_prototype_matrices(model_type): + data = np.zeros((6, 2)) + labels = np.asarray([10, 10, 10, 20, 20, 20]) + conn = cvi.CONN( + model_type=model_type, + kmeans_k=3, + kmeans_kwargs=KMEANS_KWARGS, + ) + + value = conn.get_cvi(data, labels) + + assert np.isfinite(value) + assert conn._cluster_centers.shape == (6, 2) + assert conn._CADJ.asarray().shape == (6, 6) + assert conn._CONN.asarray().shape == (6, 6) + assert len(conn._prototype_label_map) == 6 + + +def test_conn_is_public_and_defaults_to_minibatch_kmeans(): + assert cvi.modules.CONN is cvi.CONN + assert cvi.CONN in cvi.MODULES + assert cvi.CONN().model_type == "MiniBatchKMeans" + assert cvi.CONN.info.name_short == "CONN" + assert cvi.CONN.info.index_min == 0.0 + assert cvi.CONN.info.index_max == 1.0 + assert cvi.CONN.info.optimality == "max" diff --git a/tests/test_cvi.py b/tests/test_cvi.py index 26ebf64..b6e6e1d 100644 --- a/tests/test_cvi.py +++ b/tests/test_cvi.py @@ -56,8 +56,10 @@ def get_cvis() -> List[cvi.CVI]: """ # Construct a list of CVI objects + # CONN has backend-specific batch/incremental behavior and is covered by + # focused tests instead of the generic mode-equivalence test below. cvis = [ - local_cvi() for local_cvi in cvi.MODULES + local_cvi() for local_cvi in cvi.MODULES if local_cvi is not cvi.CONN ] return cvis @@ -418,9 +420,9 @@ def test_error_3d_invalid(self): # Try passing a 3D array local_cvi.get_cvi(local_data, local_label) - def test_error_batch_to_inc(self): + def test_error_repeated_batch(self): """ - Tests that batch to incremental mode is not supported yet. + Tests that repeated batch updates are not supported. """ # Create some dummy 2D data @@ -432,7 +434,7 @@ def test_error_batch_to_inc(self): local_cvi = get_one_cvi() local_cvi._is_setup = True - # Test that switching from batch to incremental is not supported + # Test that another batch update is not supported with pytest.raises(ValueError): local_cvi.get_cvi(local_data, local_label) diff --git a/tests/test_operations.py b/tests/test_operations.py new file mode 100644 index 0000000..8ce97c3 --- /dev/null +++ b/tests/test_operations.py @@ -0,0 +1,393 @@ +"""Tests for CVI add, remove, and merge operations.""" + +import copy + +import numpy as np +import pytest + +import src.cvi as cvi +from src.cvi.modules.CONN import CONN + + +SAMPLES = np.asarray([ + [0.0, 0.0], + [0.2, 0.1], + [-0.1, 0.3], + [1.5, 1.4], + [1.7, 1.5], + [1.4, 1.8], + [3.0, 0.1], + [3.2, 0.0], + [2.8, 0.3], +]) +LABELS = np.asarray([10, 10, 10, 20, 20, 20, 30, 30, 30]) + +# This is a patch to test every module except for CONN index that currently has +# its own API for batch and incremental usage (due to the selected internal clustering method) +CVIS_NOT_CONN = [m for m in cvi.MODULES if m is not cvi.CONN] + + +def build_incrementally(cvi_type, samples=SAMPLES, labels=LABELS): + """Build one CVI by replaying samples incrementally.""" + + local_cvi = cvi_type() + for sample, label in zip(samples, labels): + local_cvi.get_cvi(sample, int(label)) + return local_cvi + + +def build_in_batch(cvi_type, samples=SAMPLES, labels=LABELS): + """Build one CVI from a batch while retaining external labels.""" + + local_cvi = cvi_type() + local_cvi.get_cvi(samples, labels) + return local_cvi + + +def assert_equivalent(actual, expected): + """Compare the common and index-specific sufficient statistics.""" + + assert actual._label_map.map == expected._label_map.map + assert actual._n_samples == expected._n_samples + assert actual._n_clusters == expected._n_clusters + np.testing.assert_array_equal(np.asarray(actual._n), np.asarray(expected._n)) + np.testing.assert_allclose(actual._v, expected._v, rtol=1e-8, atol=1e-10) + np.testing.assert_allclose( + actual.criterion_value, + expected.criterion_value, + rtol=1e-7, + atol=1e-10, + ) + + for attribute in ("_CP", "_G", "_D", "_S", "_SEP", "_sigma"): + if not hasattr(actual, attribute) or not hasattr(expected, attribute): + continue + + actual_value = getattr(actual, attribute) + expected_value = getattr(expected, attribute) + if actual_value is None or expected_value is None: + assert actual_value is expected_value + continue + + np.testing.assert_allclose( + np.asarray(actual_value), + np.asarray(expected_value), + rtol=1e-7, + atol=1e-9, + ) + + +def core_snapshot(local_cvi): + """Copy state used to prove failed operations are atomic.""" + + return { + "label_map": copy.deepcopy(local_cvi._label_map.map), + "n_samples": local_cvi._n_samples, + "n_clusters": local_cvi._n_clusters, + "n": copy.deepcopy(local_cvi._n), + "v": local_cvi._v.copy(), + "criterion_value": local_cvi.criterion_value, + } + + +def assert_snapshot(local_cvi, snapshot): + """Assert that core CVI state agrees with a prior snapshot.""" + + assert local_cvi._label_map.map == snapshot["label_map"] + assert local_cvi._n_samples == snapshot["n_samples"] + assert local_cvi._n_clusters == snapshot["n_clusters"] + np.testing.assert_array_equal(np.asarray(local_cvi._n), snapshot["n"]) + np.testing.assert_array_equal(local_cvi._v, snapshot["v"]) + assert local_cvi.criterion_value == snapshot["criterion_value"] + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_remove_matches_incremental_replay(cvi_type): + """Removing a member must equal replaying all other samples.""" + + remove_index = 1 + actual = build_incrementally(cvi_type) + returned = actual.remove(SAMPLES[remove_index], int(LABELS[remove_index])) + + keep = np.arange(len(SAMPLES)) != remove_index + expected = build_incrementally(cvi_type, SAMPLES[keep], LABELS[keep]) + + assert returned == actual.criterion_value + assert_equivalent(actual, expected) + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_merge_matches_relabelled_incremental_replay(cvi_type): + """Merging labels must equal replaying with the source relabelled.""" + + actual = build_incrementally(cvi_type) + returned = actual.merge(target_label=20, source_label=10) + + merged_labels = LABELS.copy() + merged_labels[merged_labels == 10] = 20 + expected = build_incrementally(cvi_type, SAMPLES, merged_labels) + + assert returned == actual.criterion_value + assert_equivalent(actual, expected) + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_add_then_remove_restores_state(cvi_type): + """An add/remove round trip must restore the previous summary.""" + + expected = build_incrementally(cvi_type) + actual = build_incrementally(cvi_type) + sample = np.asarray([0.05, 0.15]) + + actual.get_cvi(sample, 10) + actual.remove(sample, 10) + + assert_equivalent(actual, expected) + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_merge_to_single_cluster(cvi_type): + """Merging the final two clusters leaves the CVI undefined at zero.""" + + samples = SAMPLES[:6] + labels = LABELS[:6] + actual = build_incrementally(cvi_type, samples, labels) + actual.merge(target_label=20, source_label=10) + + expected = build_incrementally( + cvi_type, + samples, + np.full(labels.shape, 20), + ) + + assert_equivalent(actual, expected) + assert actual.criterion_value == 0.0 + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_singleton_removal_compacts_and_reuses_label(cvi_type): + """Deleting a singleton removes its mapping and permits label reuse.""" + + singleton = np.asarray([[4.0, 4.0]]) + samples = np.vstack((SAMPLES, singleton)) + labels = np.append(LABELS, 99) + actual = build_incrementally(cvi_type, samples, labels) + + actual.remove(singleton[0], 99) + assert 99 not in actual._label_map.map + assert actual._n_clusters == 3 + assert_equivalent(actual, build_incrementally(cvi_type)) + + replacement = np.asarray([4.2, 3.9]) + actual.get_cvi(replacement, 99) + assert actual._label_map.map[99] == 3 + assert actual._n[3] == 1 + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_removing_final_sample_resets_object(cvi_type): + """Removing the final sample returns the object to fresh state.""" + + local_cvi = build_incrementally( + cvi_type, + samples=np.asarray([[0.25, 0.75]]), + labels=np.asarray([42]), + ) + local_cvi.remove(np.asarray([0.25, 0.75]), 42) + + assert local_cvi._n_samples == 0 + assert local_cvi._n_clusters == 0 + assert local_cvi._label_map.map == {} + assert local_cvi._is_setup is False + assert local_cvi.criterion_value == 0.0 + + local_cvi.get_cvi(np.asarray([0.1, 0.2]), 42) + assert local_cvi._n_samples == 1 + assert local_cvi._label_map.map == {42: 0} + + +def test_rcip_remove_from_two_sample_cluster(): + """rCIP handles the special covariance transition to a singleton.""" + + samples = SAMPLES[:5] + labels = np.asarray([10, 10, 20, 20, 20]) + actual = build_incrementally(cvi.rCIP, samples, labels) + actual.remove(samples[0], 10) + + expected = build_incrementally(cvi.rCIP, samples[1:], labels[1:]) + assert_equivalent(actual, expected) + + +def test_rcip_merge_singletons(): + """Merging rCIP singletons creates the correct sample covariance.""" + + samples = np.asarray([[0.0, 0.0], [1.0, 1.0], [2.0, 0.0]]) + labels = np.asarray([10, 20, 30]) + actual = build_incrementally(cvi.rCIP, samples, labels) + actual.merge(target_label=20, source_label=10) + + merged_labels = np.asarray([20, 20, 30]) + expected = build_incrementally(cvi.rCIP, samples, merged_labels) + assert_equivalent(actual, expected) + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_invalid_operation_arguments_are_atomic(cvi_type): + """Label and dimension errors must not mutate the object.""" + + local_cvi = build_incrementally(cvi_type) + snapshot = core_snapshot(local_cvi) + + with pytest.raises(ValueError, match="Unknown cluster label"): + local_cvi.remove(SAMPLES[0], 999) + assert_snapshot(local_cvi, snapshot) + + with pytest.raises(ValueError, match="Unknown cluster label"): + local_cvi.merge(10, 999) + assert_snapshot(local_cvi, snapshot) + + with pytest.raises(ValueError, match="Expected a sample"): + local_cvi.remove(np.asarray([1.0, 2.0, 3.0]), 10) + assert_snapshot(local_cvi, snapshot) + + with pytest.raises(ValueError, match="two different"): + local_cvi.merge(10, 10) + assert_snapshot(local_cvi, snapshot) + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_operations_require_initialized_state(cvi_type): + """Fresh CVIs reject structural operations.""" + + fresh = cvi_type() + with pytest.raises(ValueError, match="initialized CVI"): + fresh.remove(SAMPLES[0], 10) + with pytest.raises(ValueError, match="initialized CVI"): + fresh.merge(10, 20) + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +@pytest.mark.parametrize( + ("sample", "label"), + [ + (np.asarray([0.05, 0.15]), 10), + (np.asarray([4.0, 4.0]), 99), + ], + ids=["existing-label", "new-label"], +) +def test_batch_then_add_matches_incremental_replay(cvi_type, sample, label): + """A batch-initialized CVI can accept another scalar sample.""" + + actual = build_in_batch(cvi_type) + returned = actual.get_cvi(sample, label) + + expected = build_incrementally( + cvi_type, + np.vstack((SAMPLES, sample)), + np.append(LABELS, label), + ) + + assert returned == actual.criterion_value + assert_equivalent(actual, expected) + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_batch_then_add_rejects_wrong_dimension_atomically(cvi_type): + """An invalid scalar update must not create a new batch-state label.""" + + actual = build_in_batch(cvi_type) + snapshot = core_snapshot(actual) + + with pytest.raises(ValueError, match="Expected a sample"): + actual.get_cvi(np.asarray([1.0, 2.0, 3.0]), 99) + + assert_snapshot(actual, snapshot) + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_batch_then_remove_matches_incremental_replay(cvi_type): + """A batch-initialized CVI can remove a sample by external label.""" + + remove_index = 1 + actual = build_in_batch(cvi_type) + returned = actual.remove( + SAMPLES[remove_index], + int(LABELS[remove_index]), + ) + + keep = np.arange(len(SAMPLES)) != remove_index + expected = build_incrementally(cvi_type, SAMPLES[keep], LABELS[keep]) + + assert returned == actual.criterion_value + assert_equivalent(actual, expected) + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_batch_then_merge_matches_incremental_replay(cvi_type): + """A batch-initialized CVI can merge clusters by external label.""" + + actual = build_in_batch(cvi_type) + returned = actual.merge(target_label=20, source_label=10) + + merged_labels = LABELS.copy() + merged_labels[merged_labels == 10] = 20 + expected = build_incrementally(cvi_type, SAMPLES, merged_labels) + + assert returned == actual.criterion_value + assert_equivalent(actual, expected) + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_batch_add_then_remove_restores_state(cvi_type): + """A scalar add/remove round trip restores batch-initialized state.""" + + expected = build_in_batch(cvi_type) + actual = build_in_batch(cvi_type) + sample = np.asarray([0.05, 0.15]) + + actual.get_cvi(sample, 10) + actual.remove(sample, 10) + + assert_equivalent(actual, expected) + + +@pytest.mark.parametrize("cvi_type", CVIS_NOT_CONN) +def test_batch_singleton_removal_deletes_and_reuses_label(cvi_type): + """Batch labels are compacted and reusable after singleton deletion.""" + + singleton = np.asarray([[4.0, 4.0]]) + samples = np.vstack((SAMPLES, singleton)) + labels = np.append(LABELS, 99) + actual = build_in_batch(cvi_type, samples, labels) + + actual.remove(singleton[0], 99) + assert_equivalent(actual, build_incrementally(cvi_type)) + + actual.get_cvi(np.asarray([4.2, 3.9]), 99) + assert actual._label_map.map[99] == 3 + assert actual._n[3] == 1 + + +@pytest.mark.parametrize("cvi_type", [cvi.CH, cvi.cSIL, cvi.rCIP]) +def test_inconsistent_remove_is_atomic(cvi_type): + """Statistics-bearing CVIs reject a sample inconsistent with a cluster.""" + + local_cvi = build_incrementally(cvi_type) + snapshot = core_snapshot(local_cvi) + + with pytest.raises(ValueError, match="invalid (cluster compactness|covariance)"): + local_cvi.remove(np.asarray([100.0, 100.0]), 10) + + assert_snapshot(local_cvi, snapshot) + + +def test_conn_operations_are_explicitly_unsupported(): + """CONN exposes the common interface but defers reversible state.""" + + conn = CONN() + + with pytest.raises(NotImplementedError, match="does not support"): + conn.remove(np.asarray([0.1, 0.2]), 0) + + with pytest.raises(NotImplementedError, match="does not support"): + conn.merge(0, 1)