Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor: upgrade for Pydantic V2; upgrade Typer #77

Closed
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 24 additions & 24 deletions fcmeans/main.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,18 @@
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):
euclidean = 'euclidean'
minkowski = 'minkowski'
cosine = 'cosine'
euclidean = "euclidean"
minkowski = "minkowski"
cosine = "cosine"


class FCM(BaseModel):
r"""Fuzzy C-means Model
Expand All @@ -38,23 +39,20 @@ class FCM(BaseModel):
Raises:
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_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

Expand All @@ -76,7 +74,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

Expand All @@ -87,15 +85,17 @@ 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)
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.

Expand Down Expand Up @@ -125,13 +125,13 @@ def _dist(A: NDArray, B: NDArray, distance: str, distance_params: str) -> NDArra
"""Compute the distance between two matrices"""
if isinstance(distance, Callable):
return distance(A, B, distance_params)
elif distance == 'minkowski':
elif distance == "minkowski":
return FCM._minkowski(A, B, distance_params.get("p", 1.0))
elif distance == 'cosine':
elif distance == "cosine":
return FCM._cosine_similarity(A, B)
else:
return FCM._euclidean(A, B)

@staticmethod
def _euclidean(A: NDArray, B: NDArray) -> NDArray:
"""Compute the euclidean distance between two matrices"""
Expand All @@ -140,14 +140,14 @@ def _euclidean(A: NDArray, B: NDArray) -> NDArray:
@staticmethod
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)
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 _next_centers(X: NDArray, u: NDArray, m: float):
Expand Down
Loading
Loading