Merge pull request #87 from omadson/feature/upgrade-pydantic-to-v2

feature/upgrade pydantic to v2

.pre-commit-config.yaml

@@ -16,15 +16,6 @@ repos:
       - id: trailing-whitespace
       - id: mixed-line-ending
       - id: check-added-large-files
-
--   repo: https://github.com/psf/black
-    rev: 22.1.0
-    hooks:
-      - id: black
-        args: ['--line-length=79']
-        files: '(\.pyi?|wscript|ipynb)$'
-        language_version: python3
-        additional_dependencies: [black-nb]
 -   repo: https://github.com/asottile/blacken-docs
     rev: v1.8.0
     hooks:

fcmeans/main.py

@@ -1,18 +1,20 @@
-from typing import Optional, Dict, Union, Callable
 from enum import Enum
+from typing import Callable, Dict, Optional, Union
 
-from joblib import Parallel, delayed
 import numpy as np
-from numpy.typing import NDArray
-from pydantic import BaseModel, Extra, Field, validate_arguments
 import tqdm
+from joblib import Parallel, delayed
+from numpy.typing import NDArray
+from pydantic import BaseModel, ConfigDict, Field, validate_call
 
 
 class DistanceOptions(str, Enum):
+    """Implemented distances"""
     euclidean = 'euclidean'
     minkowski = 'minkowski'
     cosine = 'cosine'
 
+
 class FCM(BaseModel):
     r"""Fuzzy C-means Model
 
@@ -39,22 +41,22 @@ class FCM(BaseModel):
         ReferenceError: If called without the model being trained
     """
 
-    class Config:
-        extra = Extra.allow
-        arbitrary_types_allowed = True
+    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
 
     n_clusters: int = Field(5, ge=1)
     max_iter: int = Field(150, ge=1, le=1000)
     m: float = Field(2.0, ge=1.0)
     error: float = Field(1e-5, ge=1e-9)
     random_state: Optional[int] = None
-    trained: bool = Field(False, const=True)
+    trained: bool = False
     n_jobs: int = Field(1, ge=1)
     verbose: Optional[bool] = False
-    distance: Optional[Union[DistanceOptions, Callable]] = DistanceOptions.euclidean
+    distance: Optional[Union[DistanceOptions, Callable]] = (
+        DistanceOptions.euclidean
+    )
     distance_params: Optional[Dict] = {}
 
-    @validate_arguments(config=dict(arbitrary_types_allowed=True))
+    @validate_call(config=dict(arbitrary_types_allowed=True))
     def fit(self, X: NDArray) -> None:
         """Train the fuzzy-c-means model
 
@@ -64,7 +66,9 @@ def fit(self, X: NDArray) -> None:
         self.rng = np.random.default_rng(self.random_state)
         n_samples = X.shape[0]
         self.u = self.rng.uniform(size=(n_samples, self.n_clusters))
-        self.u = self.u / np.tile(self.u.sum(axis=1)[np.newaxis].T, self.n_clusters)
+        self.u = self.u / np.tile(
+            self.u.sum(axis=1)[np.newaxis].T, self.n_clusters
+        )
         for _ in tqdm.tqdm(
             range(self.max_iter), desc="Training", disable=not self.verbose
         ):
@@ -76,7 +80,7 @@ def fit(self, X: NDArray) -> None:
                 break
         self.trained = True
 
-    @validate_arguments(config=dict(arbitrary_types_allowed=True))
+    @validate_call(config=dict(arbitrary_types_allowed=True))
     def soft_predict(self, X: NDArray) -> NDArray:
         """Soft predict of FCM
 
@@ -87,15 +91,22 @@ def soft_predict(self, X: NDArray) -> NDArray:
             NDArray: Fuzzy partition array, returned as an array with
             n_samples rows and n_clusters columns.
         """
-        temp = FCM._dist(X, self._centers, self.distance, self.distance_params) ** (2 / (self.m - 1))
+        temp = FCM._dist(
+            X,
+            self._centers,
+            self.distance,
+            self.distance_params
+        ) ** (2 / (self.m - 1))
         u_dist = Parallel(n_jobs=self.n_jobs)(
-            delayed(lambda data, col: (data[:, col] / data.T).sum(0))(temp, col)
+            delayed(
+                lambda data, col: (data[:, col] / data.T).sum(0)
+            )(temp, col)
             for col in range(temp.shape[1])
         )
         u_dist = np.vstack(u_dist).T
         return 1 / u_dist
 
-    @validate_arguments(config=dict(arbitrary_types_allowed=True))
+    @validate_call(config=dict(arbitrary_types_allowed=True))
     def predict(self, X: NDArray) -> NDArray:
         """Predict the closest cluster each sample in X belongs to.
 
@@ -121,17 +132,28 @@ def _is_trained(self) -> bool:
         return False
 
     @staticmethod
-    def _dist(A: NDArray, B: NDArray, distance: str, distance_params: str) -> NDArray:
+    def _dist(
+        A: NDArray,
+        B: NDArray,
+        distance: Optional[Union[DistanceOptions, Callable]] = (
+            DistanceOptions.euclidean
+        ),
+        distance_params: Optional[Dict] = {}
+    ) -> NDArray:
         """Compute the distance between two matrices"""
-        if isinstance(distance, Callable):
+        if callable(distance):
             return distance(A, B, distance_params)
         elif distance == 'minkowski':
-            return FCM._minkowski(A, B, distance_params.get("p", 1.0))
+            if isinstance(distance_params, dict):
+                p = distance_params.get("p", 1.0)
+            else:
+                p = 1.0
+            return FCM._minkowski(A, B, p)
         elif distance == 'cosine':
-            return FCM._cosine_similarity(A, B)
+            return FCM._cosine(A, B)
         else:
             return FCM._euclidean(A, B)
-    
+
     @staticmethod
     def _euclidean(A: NDArray, B: NDArray) -> NDArray:
         """Compute the euclidean distance between two matrices"""
@@ -141,13 +163,18 @@ def _euclidean(A: NDArray, B: NDArray) -> NDArray:
     def _minkowski(A: NDArray, B: NDArray, p: float) -> NDArray:
         """Compute the minkowski distance between two matrices"""
         return (np.einsum("ijk->ij", (A[:, None, :] - B) ** p)) ** (1/p)
-    
+
     @staticmethod
     def _cosine_similarity(A: NDArray, B: NDArray) -> NDArray:
         """Compute the cosine similarity between two matrices"""
-        p1 = np.sqrt(np.sum(A**2,axis=1))[:,np.newaxis]
-        p2 = np.sqrt(np.sum(B**2,axis=1))[np.newaxis,:]
-        return np.dot(A,B.T) / (p1*p2)
+        p1 = np.sqrt(np.sum(A**2, axis=1))[:, np.newaxis]
+        p2 = np.sqrt(np.sum(B**2, axis=1))[np.newaxis, :]
+        return np.dot(A, B.T) / (p1*p2)
+
+    @staticmethod
+    def _cosine(A: NDArray, B: NDArray) -> NDArray:
+        """Compute the cosine distance between two matrices"""
+        return np.abs(1 - FCM._cosine_similarity(A, B))
 
     @staticmethod
     def _next_centers(X: NDArray, u: NDArray, m: float):