Merge pull request #18 from GeoOcean/clean/datamining

[JTH] merge last changes in datamining

Author: tausiaj

bluemath_tk/core/decorators.py

@@ -72,8 +72,6 @@ def wrapper(
         directional_variables: List[str] = [],
         custom_scale_factor: dict = {},
     ):
-        # NOTE: Default custom scale factors are defined below
-        _default_custom_scale_factor = {}
         if data is None:
             raise ValueError("Data cannot be None")
         elif not isinstance(data, pd.DataFrame):
@@ -82,19 +80,6 @@ def wrapper(
             raise TypeError("Directional variables must be a list")
         if not isinstance(custom_scale_factor, dict):
             raise TypeError("Custom scale factor must be a dict")
-        for directional_variable in directional_variables:
-            if directional_variable not in custom_scale_factor:
-                if directional_variable in _default_custom_scale_factor:
-                    custom_scale_factor[directional_variable] = (
-                        _default_custom_scale_factor[directional_variable]
-                    )
-                    self.logger.warning(
-                        f"Using default custom scale factor for {directional_variable}"
-                    )
-                else:
-                    self.logger.warning(
-                        f"No custom scale factor provided for {directional_variable}, min and max values will be used"
-                    )
         return func(self, data, directional_variables, custom_scale_factor)
 
     return wrapper
@@ -119,11 +104,9 @@ def validate_data_kma(func):
     def wrapper(
         self,
         data: pd.DataFrame,
-        directional_variables: List[str],
-        custom_scale_factor: dict,
+        directional_variables: List[str] = [],
+        custom_scale_factor: dict = {},
     ):
-        # NOTE: Default custom scale factors are defined below
-        _default_custom_scale_factor = {}
         if data is None:
             raise ValueError("Data cannot be None")
         elif not isinstance(data, pd.DataFrame):
@@ -132,24 +115,46 @@ def wrapper(
             raise TypeError("Directional variables must be a list")
         if not isinstance(custom_scale_factor, dict):
             raise TypeError("Custom scale factor must be a dict")
-        for directional_variable in directional_variables:
-            if directional_variable not in custom_scale_factor:
-                if directional_variable in _default_custom_scale_factor:
-                    custom_scale_factor[directional_variable] = (
-                        _default_custom_scale_factor[directional_variable]
-                    )
-                    self.logger.warning(
-                        f"Using default custom scale factor for {directional_variable}"
-                    )
-                else:
-                    self.logger.warning(
-                        f"No custom scale factor provided for {directional_variable}, min and max values will be used"
-                    )
         return func(self, data, directional_variables, custom_scale_factor)
 
     return wrapper
 
 
+def validate_data_som(func):
+    """
+    Decorator to validate data in SOM class fit method.
+
+    Parameters
+    ----------
+    func : callable
+        The function to be decorated
+
+    Returns
+    -------
+    callable
+        The decorated function
+    """
+
+    @functools.wraps(func)
+    def wrapper(
+        self,
+        data: pd.DataFrame,
+        directional_variables: List[str] = [],
+        num_iteration: int = 1000,
+    ):
+        if data is None:
+            raise ValueError("Data cannot be None")
+        elif not isinstance(data, pd.DataFrame):
+            raise TypeError("Data must be a pandas DataFrame")
+        if not isinstance(directional_variables, list):
+            raise TypeError("Directional variables must be a list")
+        if not isinstance(num_iteration, int) or num_iteration <= 0:
+            raise ValueError("Number of iterations must be integer and > 0")
+        return func(self, data, directional_variables, num_iteration)
+
+    return wrapper
+
+
 def validate_data_pca(func):
     """
     Decorator to validate data in PCA class fit method.

bluemath_tk/core/models.py

@@ -270,3 +270,85 @@ def get_metrics(
             }
 
         return pd.DataFrame(metrics).T
+
+    @staticmethod
+    def _get_uv_components(x_deg: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
+        """
+        This method calculates the u and v components for the given directional data.
+
+        Here, we assume that the directional data is in degrees,
+            beign 0° the North direction,
+            and increasing clockwise.
+
+                   0° N
+                    |
+                    |
+        270° W <---------> 90° E
+                    |
+                    |
+                  90° S
+
+        Parameters
+        ----------
+        x_deg : np.ndarray
+            The directional data in degrees.
+
+        Returns
+        -------
+        Tuple[np.ndarray, np.ndarray]
+            The u and v components.
+
+        Notes
+        -----
+        - TODO: This method can be moved to a separate utility module.
+        """
+
+        # Convert degrees to radians and adjust by subtracting from π/2
+        x_rad = x_deg * np.pi / 180
+
+        # Calculate x and y components using cosine and sine
+        xu = np.sin(x_rad)
+        xv = np.cos(x_rad)
+
+        # Return the u and v components
+        return xu, xv
+
+    @staticmethod
+    def _get_degrees_from_uv(xu: np.ndarray, xv: np.ndarray) -> np.ndarray:
+        """
+        This method calculates the degrees from the u and v components.
+
+        Here, we assume u and v represent angles between 0 and 360 degrees,
+            where 0° is the North direction,
+            and increasing clockwise.
+
+                     (u=0, v=1)
+                         |
+                         |
+        (u=-1, v=0) <---------> (u=1, v=0)
+                         |
+                         |
+                     (u=0, v=-1)
+
+        Parameters
+        ----------
+        xu : np.ndarray
+            The u component.
+        xv : np.ndarray
+            The v component.
+
+        Returns
+        -------
+        np.ndarray
+            The degrees.
+
+        Notes
+        -----
+        - TODO: This method can be moved to a separate utility module.
+        """
+
+        # Calculate the degrees using the arctangent function
+        x_deg = np.arctan2(xu, xv) * 180 / np.pi % 360
+
+        # Return the degrees
+        return x_deg

bluemath_tk/core/operations.py

@@ -201,6 +201,7 @@ def denormalize(
     return data
 
 
+# TODO: Return pd.DataFrame or xr.Dataset depending on input type
 def standarize(
     data: Union[np.ndarray, pd.DataFrame, xr.Dataset],
     scaler: StandardScaler = None,

bluemath_tk/core/plotting/base_plotting.py

@@ -129,6 +129,9 @@ def plot_scatter(self, ax, **kwargs):
         ax.scatter(**kwargs)
         self.set_grid(ax)
 
+    def plot_pie(self, ax, **kwargs):
+        ax.pie(**kwargs)
+
     def plot_map(self, ax, **kwargs):
         ax.set_global()
         ax.coastlines()

bluemath_tk/datamining/_base_datamining.py

@@ -1,5 +1,5 @@
 from abc import abstractmethod
-from typing import Tuple
+from typing import Tuple, List
 import numpy as np
 import pandas as pd
 from matplotlib import pyplot as plt
@@ -209,7 +209,7 @@ def plot_selected_centroids(
         """
 
         if (
-            list(self.data.columns) == list(self.centroids.columns)
+            len(self.data.columns) == len(self.centroids.columns)
             and list(self.data.columns) != []
         ):
             variables_names = list(self.data.columns)
@@ -350,3 +350,60 @@ class BaseReduction(BlueMathModel):
     @abstractmethod
     def __init__(self) -> None:
         super().__init__()
+
+
+class ClusteringComparator:
+    """
+    Class for comparing clustering models.
+    """
+
+    def __init__(self, list_of_models: List[BaseClustering]) -> None:
+        """
+        Initializes the ClusteringComparator class.
+        """
+
+        self.list_of_models = list_of_models
+
+    def fit(
+        self,
+        data: pd.DataFrame,
+        directional_variables: List[str] = [],
+        custom_scale_factor: dict = {},
+    ) -> None:
+        """
+        Fits the clustering models.
+        """
+
+        for model in self.list_of_models:
+            if model.__class__.__name__ == "SOM":
+                model.fit(
+                    data=data,
+                    directional_variables=directional_variables,
+                )
+            else:
+                model.fit(
+                    data=data,
+                    directional_variables=directional_variables,
+                    custom_scale_factor=custom_scale_factor,
+                )
+
+    def plot_selected_centroids(self) -> None:
+        """
+        Plots the selected centroids for the clustering models.
+        """
+
+        for model in self.list_of_models:
+            fig, axes = model.plot_selected_centroids()
+            fig.suptitle(f"Selected centroids for {model.__class__.__name__}")
+
+    def plot_data_as_clusters(self, data: pd.DataFrame) -> None:
+        """
+        Plots the data as clusters for the clustering models.
+        """
+
+        for model in self.list_of_models:
+            nearest_centroids, _ = model.predict(data=data)
+            fig, axes = model.plot_data_as_clusters(
+                data=data, nearest_centroids=nearest_centroids
+            )
+            fig.suptitle(f"Data as clusters for {model.__class__.__name__}")