Skip to content

Base

Model base.

Contains a BaseEstimator which provides the base's class for the rest of the estimator.

There is no intention to use this class directly, but to be inherited by other classes. Implementation is based on scikit-learn's BaseEstimator in order to facilitate the integration with the library.

BaseEstimator

Bases: ABC, ClusterMixin

Base class for estimators in the clustlib package.

Attributes:

Name Type Description
labels_ ndarray

Labels of the dataset.
Notes

All estimators should specify all the parameters that can be set at the class level in their __init__ as explicit keyword arguments (no *args or **kwargs).

Source code in clustlib/model.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class BaseEstimator(ABC, SklearnBaseEstimator):
    """Base class for estimators in the clustlib package.

    Attributes:
        labels_ (numpy.ndarray): Labels of the dataset.

    Notes:
        All estimators should specify all the parameters that can be set at the class
        level in their `__init__` as explicit keyword
        arguments (no `*args` or `**kwargs`).

    """

    centroids: np.ndarray = None
    init: InitCentroid

    n_clusters: int
    tol: float
    max_iter: int = None

    constraints: SimpleConstraints
    X: np.ndarray
    _labels: np.array = None

    _delta: np.ndarray = None

    def fit(self, dataset: np.ndarray, labels: np.array = None):
        """Fit the model to the data.

        Args:
            dataset (numpy.ndarray): The data to cluster.
            labels (numpy.ndarray, optional): Ignored. This parameter exists only for
                compatibility with the sklearn API.

        Returns:
            BaseEstimator: The fitted estimator.

        """
        self.X = np.copy(dataset)

        if self.centroids is None:
            if isinstance(self.init, np.ndarray):
                self.centroids = self.init
            elif self.init == "random":
                self.centroids = random(dataset, self.n_clusters)
            elif self.init == "kmeans":
                self.centroids = kmeans(dataset, self.n_clusters)
            else:
                raise ValueError("Unknown initialization method")

        if self._labels is None:
            self._labels = np.random.randint(
                0, self.n_clusters, dataset.shape[0], dtype=int
            )
        else:
            self._labels = np.copy(labels)

        return self._fit()

    def _fit(self):
        """Fit the model to the data.

        Args:
            dataset (numpy.ndarray): The data to cluster.
            labels (numpy.ndarray, optional): Ignored. This parameter exists only for
                compatibility with the sklearn API.

        Returns:
            BaseEstimator: The fitted estimator.

        """
        raise NotImplementedError

    def predict(self, x: np.array) -> int:
        """Predict the cluster index for a given instance.

        Args:
            x (numpy.ndarray): The instance to be predicted.

        Returns:
            int: The index of the cluster to which the instance is assigned.

        """
        return np.argmin(
            np.linalg.norm(x[:, np.newaxis] - self.centroids, axis=2), axis=1
        )

    def calculte_delta(self, x: np.ndarray) -> np.ndarray:
        """Calculate the difference between the new and old centroids.

        This method is used to determine when the algorithm has converged.

        Args:
            x (numpy.ndarray): The old centroids.

        Returns:
            numpy.ndarray: The absolute difference between the new and old centroids.

        """
        return np.abs(self.centroids - x)

    def update(self):
        """Update the centroids of the clusters.

        This method calls the `_update` method to update the centroids of the clusters.
        It also updates the `_delta` attribute with the difference between the new and
        old centroids. The `_delta` attribute is a numpy array with the same shape as
        the centroids and is used to determine when the algorithm has converged.
        """
        aux = np.copy(self.centroids)
        self._update()
        self._delta = self.calculte_delta(aux)

    def _update(self):
        """Update the centroids of the clusters.

        This method should be implemented by any class that inherits from it.
        """
        raise NotImplementedError

    def _convergence(self):
        """Check convergence of the algorithm.

        Returns:
            bool: True if the algorithm has converged, False otherwise.

        """
        if self._delta is None:
            logger.debug("Delta is None, convergence cannot be checked.")
            return False

    def stop_criteria(self, iteration) -> bool:
        """Check if the algorithm has reached the stopping criteria.

        Args:
            iteration (int): The current iteration of the algorithm.

        Returns:
            bool: True if the algorithm has reached the stopping criteria,
                False otherwise.

        """
        if self._convergence():
            logger.debug("Convergence reached, stopping criteria met.")
            return True

        if self.max_iter is None:
            logger.debug("No maximum iterations set, stopping criteria not met.")
            return False

        return iteration > self.max_iter

calculte_delta(x)

Calculate the difference between the new and old centroids.

This method is used to determine when the algorithm has converged.

Parameters:

Name Type Description Default
x ndarray

The old centroids.

 required

Returns:

Type Description
ndarray

numpy.ndarray: The absolute difference between the new and old centroids.
Source code in clustlib/model.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
def calculte_delta(self, x: np.ndarray) -> np.ndarray:
    """Calculate the difference between the new and old centroids.

    This method is used to determine when the algorithm has converged.

    Args:
        x (numpy.ndarray): The old centroids.

    Returns:
        numpy.ndarray: The absolute difference between the new and old centroids.

    """
    return np.abs(self.centroids - x)

fit(dataset, labels=None)

Fit the model to the data.

Parameters:

Name Type Description Default
dataset ndarray

The data to cluster.

 required
labels ndarray

Ignored. This parameter exists only for compatibility with the sklearn API.

 None

Returns:

Name Type Description
BaseEstimator

The fitted estimator.
Source code in clustlib/model.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def fit(self, dataset: np.ndarray, labels: np.array = None):
    """Fit the model to the data.

    Args:
        dataset (numpy.ndarray): The data to cluster.
        labels (numpy.ndarray, optional): Ignored. This parameter exists only for
            compatibility with the sklearn API.

    Returns:
        BaseEstimator: The fitted estimator.

    """
    self.X = np.copy(dataset)

    if self.centroids is None:
        if isinstance(self.init, np.ndarray):
            self.centroids = self.init
        elif self.init == "random":
            self.centroids = random(dataset, self.n_clusters)
        elif self.init == "kmeans":
            self.centroids = kmeans(dataset, self.n_clusters)
        else:
            raise ValueError("Unknown initialization method")

    if self._labels is None:
        self._labels = np.random.randint(
            0, self.n_clusters, dataset.shape[0], dtype=int
        )
    else:
        self._labels = np.copy(labels)

    return self._fit()

predict(x)

Predict the cluster index for a given instance.

Parameters:

Name Type Description Default
x ndarray

The instance to be predicted.

 required

Returns:

Name Type Description
int int

The index of the cluster to which the instance is assigned.
Source code in clustlib/model.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def predict(self, x: np.array) -> int:
    """Predict the cluster index for a given instance.

    Args:
        x (numpy.ndarray): The instance to be predicted.

    Returns:
        int: The index of the cluster to which the instance is assigned.

    """
    return np.argmin(
        np.linalg.norm(x[:, np.newaxis] - self.centroids, axis=2), axis=1
    )

stop_criteria(iteration)

Check if the algorithm has reached the stopping criteria.

Parameters:

Name Type Description Default
iteration int

The current iteration of the algorithm.

 required

Returns:

Name Type Description
bool bool

True if the algorithm has reached the stopping criteria, False otherwise.
Source code in clustlib/model.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def stop_criteria(self, iteration) -> bool:
    """Check if the algorithm has reached the stopping criteria.

    Args:
        iteration (int): The current iteration of the algorithm.

    Returns:
        bool: True if the algorithm has reached the stopping criteria,
            False otherwise.

    """
    if self._convergence():
        logger.debug("Convergence reached, stopping criteria met.")
        return True

    if self.max_iter is None:
        logger.debug("No maximum iterations set, stopping criteria not met.")
        return False

    return iteration > self.max_iter

update()

Update the centroids of the clusters.

This method calls the _update method to update the centroids of the clusters. It also updates the _delta attribute with the difference between the new and old centroids. The _delta attribute is a numpy array with the same shape as the centroids and is used to determine when the algorithm has converged.

Source code in clustlib/model.py 
125
126
127
128
129
130
131
132
133
134
135
def update(self):
    """Update the centroids of the clusters.

    This method calls the `_update` method to update the centroids of the clusters.
    It also updates the `_delta` attribute with the difference between the new and
    old centroids. The `_delta` attribute is a numpy array with the same shape as
    the centroids and is used to determine when the algorithm has converged.
    """
    aux = np.copy(self.centroids)
    self._update()
    self._delta = self.calculte_delta(aux)