Fix: fixed non-strict typing errors

Author: Wenderlog

pysatl_cpd/core/algorithms/entropies/KLDivergence_entropy.py

@@ -102,8 +102,8 @@ def _calculate_kl_divergence(self) -> float:
     def _calculate_kl_divergence_histogram(
         self, ref_data: npt.NDArray[np.float64], curr_data: npt.NDArray[np.float64]
     ) -> float:
-        data_min = min(np.min(ref_data), np.min(curr_data))
-        data_max = max(np.max(ref_data), np.max(curr_data))
+        data_min = float(np.min(ref_data))
+        data_max = float(np.max(ref_data))
 
         margin = (data_max - data_min) * 0.01
         bin_edges = np.linspace(data_min - margin, data_max + margin, self._num_bins + 1)
@@ -120,39 +120,39 @@ def _calculate_kl_divergence_histogram(
         ref_prob = ref_prob / np.sum(ref_prob)
         curr_prob = curr_prob / np.sum(curr_prob)
 
-        kl_pq = np.sum(ref_prob * np.log(ref_prob / curr_prob))
+        kl_pq = float(np.sum(ref_prob * np.log(ref_prob / curr_prob)))
 
         if self._symmetric:
-            kl_qp = np.sum(curr_prob * np.log(curr_prob / ref_prob))
+            kl_qp = float(np.sum(curr_prob * np.log(curr_prob / ref_prob)))
             return (kl_pq + kl_qp) / 2
         else:
             return kl_pq
 
     def _calculate_kl_divergence_kde(
         self, ref_data: npt.NDArray[np.float64], curr_data: npt.NDArray[np.float64]
     ) -> float:
-        ref_kde = stats.gaussian_kde(ref_data)
-        curr_kde = stats.gaussian_kde(curr_data)
-
-        data_min = min(np.min(ref_data), np.min(curr_data))
-        data_max = max(np.max(ref_data), np.max(curr_data))
+        data_min = float(np.min(np.array([ref_data.min(), curr_data.min()])))
+        data_max = float(np.max(np.array([ref_data.max(), curr_data.max()])))
         margin = (data_max - data_min) * 0.1
         x_eval = np.linspace(data_min - margin, data_max + margin, 1000)
 
+        ref_kde = stats.gaussian_kde(ref_data)
+        curr_kde = stats.gaussian_kde(curr_data)
+
         ref_density = ref_kde(x_eval)
         curr_density = curr_kde(x_eval)
 
-        ref_density = ref_density + self._smoothing_factor
-        curr_density = curr_density + self._smoothing_factor
+        ref_density += self._smoothing_factor
+        curr_density += self._smoothing_factor
 
         dx = x_eval[1] - x_eval[0]
-        ref_density = ref_density / (np.sum(ref_density) * dx)
-        curr_density = curr_density / (np.sum(curr_density) * dx)
+        ref_density /= np.sum(ref_density) * dx
+        curr_density /= np.sum(curr_density) * dx
 
-        kl_pq = np.sum(ref_density * np.log(ref_density / curr_density)) * dx
+        kl_pq = float(np.sum(ref_density * np.log(ref_density / curr_density)) * dx)
 
         if self._symmetric:
-            kl_qp = np.sum(curr_density * np.log(curr_density / ref_density)) * dx
+            kl_qp = float(np.sum(curr_density * np.log(curr_density / ref_density)) * dx)
             return (kl_pq + kl_qp) / 2
         else:
             return kl_pq
@@ -166,7 +166,7 @@ def _update_reference_distribution(self) -> None:
     def get_kl_history(self) -> list[float]:
         return self._kl_values.copy()
 
-    def get_current_parameters(self) -> dict:
+    def get_current_parameters(self) -> dict[str, float | int | bool]:
         return {
             "window_size": self._window_size,
             "reference_window_size": self._reference_window_size,
@@ -200,7 +200,7 @@ def set_parameters(
         if smoothing_factor is not None:
             self._smoothing_factor = smoothing_factor
 
-    def get_distribution_comparison(self) -> dict:
+    def get_distribution_comparison(self) -> dict[str, float]:
         if len(self._reference_buffer) < self._reference_window_size or len(self._current_buffer) < self._window_size:
             return {}
 
@@ -225,7 +225,7 @@ def get_distribution_comparison(self) -> dict:
             "ks_pvalue": ks_pvalue,
         }
 
-    def analyze_distributions(self) -> dict:
+    def analyze_distributions(self) -> dict[str, float]:
         if len(self._reference_buffer) < self._reference_window_size or len(self._current_buffer) < self._window_size:
             return {}
 

pysatl_cpd/core/algorithms/entropies/approximate_entropy.py

@@ -85,11 +85,13 @@ def _calculate_approximate_entropy(self, time_series: npt.NDArray[np.float64]) -
 
         r = self._r
         if r is None:
-            std_dev = np.std(time_series)
+            std_dev = float(np.std(time_series))
             if std_dev == 0:
                 return 0.0
             r = self._r_factor * std_dev
 
+        assert r is not None
+
         phi_m = self._calculate_phi(time_series, self._m, r)
         phi_m_plus_1 = self._calculate_phi(time_series, self._m + 1, r)
 

pysatl_cpd/core/algorithms/entropies/dispersion_entropy.py

@@ -1,5 +1,5 @@
 from collections import deque
-from typing import Optional
+from typing import Optional, Union
 
 import numpy as np
 import numpy.typing as npt
@@ -125,7 +125,7 @@ def _discretize_to_classes(self, y_series: npt.NDArray[np.float64]) -> npt.NDArr
         z_series = np.clip(z_series, 1, self._num_classes)
         return z_series
 
-    def _create_dispersion_patterns(self, z_series: npt.NDArray[np.int32]) -> list[tuple]:
+    def _create_dispersion_patterns(self, z_series: npt.NDArray[np.int32]) -> list[tuple[int, ...]]:
         N = len(z_series)
         patterns = []
 
@@ -140,22 +140,22 @@ def _create_dispersion_patterns(self, z_series: npt.NDArray[np.int32]) -> list[t
 
         return patterns
 
-    def _calculate_pattern_probabilities(self, patterns: list[tuple]) -> dict[tuple, float]:
+    def _calculate_pattern_probabilities(self, patterns: list[tuple[int, ...]]) -> dict[tuple[int, ...], float]:
         if not patterns:
             return {}
 
-        pattern_counts = {}
+        pattern_counts: dict[tuple[int, ...], int] = {}
         for pattern in patterns:
             pattern_counts[pattern] = pattern_counts.get(pattern, 0) + 1
 
         total_patterns = len(patterns)
-        pattern_probs = {}
+        pattern_probs: dict[tuple[int, ...], float] = {}
         for pattern, count in pattern_counts.items():
             pattern_probs[pattern] = count / total_patterns
 
         return pattern_probs
 
-    def _calculate_shannon_entropy(self, pattern_probs: dict[tuple, float]) -> float:
+    def _calculate_shannon_entropy(self, pattern_probs: dict[tuple[int, ...], float]) -> float:
         if not pattern_probs:
             return 0.0
 
@@ -169,7 +169,7 @@ def _calculate_shannon_entropy(self, pattern_probs: dict[tuple, float]) -> float
     def get_entropy_history(self) -> list[float]:
         return self._entropy_values.copy()
 
-    def get_current_parameters(self) -> dict:
+    def get_current_parameters(self) -> dict[str, int | float | bool]:
         return {
             "window_size": self._window_size,
             "embedding_dim": self._embedding_dim,
@@ -204,7 +204,7 @@ def set_parameters(
                 f"c^w ({self._num_classes}^{self._embedding_dim}) should be less than window_size ({self._window_size})"
             )
 
-    def get_pattern_distribution(self) -> dict[tuple, int]:
+    def get_pattern_distribution(self) -> dict[tuple[int, ...], int]:
         if len(self._buffer) < self._window_size:
             return {}
 
@@ -218,13 +218,13 @@ def get_pattern_distribution(self) -> dict[tuple, int]:
         z_series = self._discretize_to_classes(y_series)
         patterns = self._create_dispersion_patterns(z_series)
 
-        pattern_counts = {}
+        pattern_counts: dict[tuple[int, ...], int] = {}
         for pattern in patterns:
             pattern_counts[pattern] = pattern_counts.get(pattern, 0) + 1
 
         return pattern_counts
 
-    def analyze_complexity(self) -> dict:
+    def analyze_complexity(self) -> dict[str, Union[float, int]]:
         if len(self._buffer) < self._window_size:
             return {}
 

pysatl_cpd/core/algorithms/entropies/renyi_entropy.py

@@ -112,7 +112,8 @@ def _compute_probabilities(self, time_series: npt.NDArray[np.float64]) -> list[f
 
         probabilities = []
         for i in range(1, len(bin_edges)):
-            count = bin_counts.get(i, 0)
+            key = np.int64(i)
+            count = bin_counts.get(key, 0)
             prob = count / total_count if total_count > 0 else 0.0
             probabilities.append(prob)
 

pysatl_cpd/core/algorithms/entropies/sample_entropy.py

@@ -95,21 +95,20 @@ def _calculate_sample_entropy(self, time_series: npt.NDArray[np.float64]) -> flo
 
         r = self._r
         if r is None:
-            std_dev = np.std(time_series)
+            std_dev = float(np.std(time_series))
             if std_dev == 0:
                 return float("inf")
             r = self._r_factor * std_dev
 
+        assert r is not None
+
         B = self._count_matches(time_series, self._m, r)
         A = self._count_matches(time_series, self._m + 1, r)
 
-        if B == 0:
-            return float("inf")
-        if A == 0:
+        if B == 0 or A == 0:
             return float("inf")
 
-        sample_entropy = -np.log(A / B)
-        return float(sample_entropy)
+        return float(-np.log(A / B))
 
     def _count_matches(self, time_series: npt.NDArray[np.float64], m: int, r: float) -> int:
         N = len(time_series)

pysatl_cpd/core/algorithms/entropies/slope_entropy.py

@@ -1,5 +1,5 @@
 from collections import deque
-from typing import Optional
+from typing import Any, Optional, Union
 
 import numpy as np
 import numpy.typing as npt
@@ -141,7 +141,7 @@ def _create_slope_pattern(self, subsequence: npt.NDArray[np.float64]) -> list[in
     def get_entropy_history(self) -> list[float]:
         return self._entropy_values.copy()
 
-    def get_current_parameters(self) -> dict:
+    def get_current_parameters(self) -> dict[str, Any]:
         return {
             "window_size": self._window_size,
             "embedding_dim": self._embedding_dim,
@@ -196,7 +196,7 @@ def get_pattern_distribution(self) -> dict[tuple[int, ...], float]:
 
         return pattern_probs
 
-    def analyze_slope_characteristics(self) -> dict:
+    def analyze_slope_characteristics(self) -> dict[str, Any]:
         if len(self._buffer) < self._window_size:
             return {}
 
@@ -232,7 +232,9 @@ def get_symbol_meanings(self) -> dict[int, str]:
             -2: f"Steep negative slope (< -{self._gamma})",
         }
 
-    def demonstrate_encoding(self, sample_data: list[float]) -> dict:
+    def demonstrate_encoding(
+        self, sample_data: list[float]
+    ) -> dict[str, Union[str, float, int, list[float], list[int], list[list[int]], dict[int, str]]]:
         if len(sample_data) < self._embedding_dim:
             return {"error": "Sample data too short"}
 

pysatl_cpd/core/algorithms/entropies/tsallis_entropy.py

@@ -1,8 +1,9 @@
 from collections import deque
-from typing import Optional
+from typing import Any, Optional, Union
 
 import numpy as np
 import numpy.typing as npt
+from numpy.typing import NDArray
 from scipy import stats
 from scipy.integrate import quad
 
@@ -143,19 +144,21 @@ def _calculate_continuous_tsallis_entropy(self, time_series: npt.NDArray[np.floa
             data_min, data_max = np.min(time_series), np.max(time_series)
             margin = (data_max - data_min) * 0.2
 
-            def integrand(x):
-                p_x = kde(x)[0] if np.isscalar(x) else kde(x)
-                return p_x**q
+            def integrand(x: Union[float, NDArray[np.float64]]) -> float:
+                x_array: NDArray[np.float64] = np.atleast_1d(x).astype(np.float64)
+                p_x = kde(x_array)[0]
+                return float(p_x**q)
 
             integral_result, _ = quad(
                 integrand, data_min - margin, data_max + margin, limit=100, epsabs=1e-8, epsrel=1e-8
             )
             v = 1e-10
             if abs(q - 1.0) < v:
 
-                def shannon_integrand(x):
-                    p_x = kde(x)[0] if np.isscalar(x) else kde(x)
-                    return p_x * np.log(p_x + 1e-10)
+                def shannon_integrand(x: Union[float, NDArray[np.float64]]) -> float:
+                    x_array: NDArray[np.float64] = np.atleast_1d(x).astype(np.float64)
+                    p_x = kde(x_array)[0]
+                    return float(p_x * np.log(p_x + 1e-10))
 
                 shannon_integral, _ = quad(
                     shannon_integrand, data_min - margin, data_max + margin, limit=100, epsabs=1e-8, epsrel=1e-8
@@ -180,9 +183,9 @@ def _calculate_max_discrete_tsallis_entropy(self, n_states: int, q: float) -> fl
         p_uniform = 1.0 / n_states
 
         if abs(q - 1.0) < v:
-            return self._k_constant * np.log(n_states)
+            return float(self._k_constant * np.log(n_states))
         else:
-            return self._k_constant * (1.0 / (q - 1.0)) * (1.0 - n_states * (p_uniform**q))
+            return float(self._k_constant * (1.0 / (q - 1.0)) * (1.0 - n_states * (p_uniform**q)))
 
     def _combine_multi_q_entropies(self, entropies: dict[float, float]) -> float:
         weights = {}
@@ -245,7 +248,7 @@ def get_entropy_history(self) -> list[float]:
     def get_multi_q_history(self) -> dict[float, list[float]]:
         return {q: values.copy() for q, values in self._multi_entropy_values.items()}
 
-    def get_current_parameters(self) -> dict:
+    def get_current_parameters(self) -> dict[str, Any]:
         return {
             "window_size": self._window_size,
             "q_parameter": self._q_parameter,
@@ -268,20 +271,20 @@ def set_parameters(
         normalize: Optional[bool] = None,
         multi_q: Optional[bool] = None,
     ) -> None:
-        def set_q_param(q):
+        def set_q_param(q: float) -> None:
             v = 1e-10
             if abs(q - 1.0) < v:
                 raise ValueError("q parameter cannot be 1")
             self._q_parameter = q
             if not self._multi_q:
                 self._q_values = [q]
 
-        def set_k_const(k):
+        def set_k_const(k: float) -> None:
             if k <= 0:
                 raise ValueError("k constant must be positive")
             self._k_constant = k
 
-        def set_num_bins_func(bins):
+        def set_num_bins_func(bins: int) -> None:
             if bins <= 1:
                 raise ValueError("Number of bins must be greater than 1")
             self._num_bins = bins
@@ -312,7 +315,7 @@ def set_num_bins_func(bins):
             else:
                 self._q_values = [self._q_parameter]
 
-    def analyze_q_sensitivity(self) -> dict:
+    def analyze_q_sensitivity(self) -> dict[str, Any]:
         if len(self._buffer) < self._window_size:
             return {}
 
@@ -341,7 +344,7 @@ def analyze_q_sensitivity(self) -> dict:
             "entropy_variance_across_q": np.var(list(q_entropies.values())) if q_entropies else 0,
         }
 
-    def get_complexity_metrics(self) -> dict:
+    def get_complexity_metrics(self) -> dict[str, Any]:
         if len(self._buffer) < self._window_size:
             return {}