Change Dict and List typing to dict and list built-in types

Author: jaime-cespedes-sisniega

frouros/callbacks/batch/permutation_test.py

@@ -1,7 +1,7 @@
 """Permutation test batch callback module."""
 
 import multiprocessing
-from typing import Any, Callable, Dict, List, Optional, Tuple
+from typing import Any, Callable, Optional, Tuple
 
 import numpy as np  # type: ignore
 
@@ -154,14 +154,14 @@ def _calculate_p_value(  # pylint: disable=too-many-arguments
         X_ref: np.ndarray,  # noqa: N803
         X_test: np.ndarray,
         statistic: Callable,
-        statistic_args: Dict[str, Any],
+        statistic_args: dict[str, Any],
         observed_statistic: float,
         num_permutations: int,
         num_jobs: int,
         conservative: bool,
         random_state: Optional[int],
         verbose: bool,
-    ) -> Tuple[List[float], float]:
+    ) -> Tuple[list[float], float]:
         permuted_statistic = permutation(
             X=X_ref,
             Y=X_test,

frouros/callbacks/streaming/history.py

@@ -1,6 +1,6 @@
 """History callback module."""
 
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Optional, Union
 
 from frouros.callbacks.streaming.base import BaseCallbackStreaming
 from frouros.utils.stats import BaseStat
@@ -46,18 +46,18 @@ def __init__(  # noqa: D107
         name: Optional[str] = None,
     ) -> None:
         super().__init__(name=name)
-        self.additional_vars: List[str] = []
-        self.history: Dict[str, List[Any]] = {
+        self.additional_vars: list[str] = []
+        self.history: dict[str, list[Any]] = {
             "value": [],
             "num_instances": [],
             "drift": [],
         }
 
-    def add_additional_vars(self, vars_: List[str]) -> None:
+    def add_additional_vars(self, vars_: list[str]) -> None:
         """Add additional variables to track.
 
         :param vars_: list of variables
-        :type vars_: List[str]
+        :type vars_: list[str]
         """
         self.additional_vars.extend(vars_)
         self.history = {**self.history, **{var: [] for var in self.additional_vars}}

frouros/datasets/base.py

@@ -4,7 +4,7 @@
 import tempfile
 import urllib.parse
 from pathlib import Path
-from typing import Any, List, Optional, Union
+from typing import Any, Optional, Union
 
 import numpy as np  # type: ignore
 import requests
@@ -22,13 +22,13 @@ class BaseDatasetDownload(abc.ABC):
 
     def __init__(
         self,
-        url: Union[str, List[str]],
+        url: Union[str, list[str]],
         file_path: Optional[str] = None,
     ) -> None:
         """Init method.
 
         :param url: url or url mirrors from where dataset will be downloaded
-        :type url: Union[str, List[str]]
+        :type url: Union[str, list[str]]
         :param file_path: file path for the downloaded file
         :type file_path: str
         """
@@ -58,20 +58,20 @@ def file_path(self, value: Optional[Path]) -> None:
         self._file_path = value
 
     @property
-    def url(self) -> Union[str, List[str]]:
+    def url(self) -> Union[str, list[str]]:
         """URL property.
 
         :return: URL from where dataset will be downloaded
-        :rtype: Union[str, List[str]]
+        :rtype: Union[str, list[str]]
         """
         return self._url
 
     @url.setter
-    def url(self, value: Union[str, List[str]]) -> None:
+    def url(self, value: Union[str, list[str]]) -> None:
         """URL setter.
 
         :param value: value to be set
-        :type value: Union[str, List[str]]
+        :type value: Union[str, list[str]]
         :raises InvalidURLError: Invalid URL exception
         """
         urls = [value] if isinstance(value, str) else value

frouros/detectors/base.py

@@ -1,7 +1,7 @@
 """Base detector module."""
 
 import abc
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Optional, Union
 
 import numpy as np  # type: ignore
 
@@ -13,33 +13,33 @@ class BaseDetector(abc.ABC):
 
     def __init__(
         self,
-        callbacks: Optional[Union[BaseCallback, List[BaseCallback]]] = None,
+        callbacks: Optional[Union[BaseCallback, list[BaseCallback]]] = None,
     ) -> None:
         """Init method.
 
         :param callbacks: callbacks
-        :type callbacks: Optional[Union[BaseCallback, List[Callback]]]
+        :type callbacks: Optional[Union[BaseCallback, list[Callback]]]
         """
         self.callbacks = callbacks  # type: ignore
 
     @property
-    def callbacks(self) -> Optional[List[BaseCallback]]:
+    def callbacks(self) -> Optional[list[BaseCallback]]:
         """Callbacks property.
 
         :return: callbacks
-        :rtype: Optional[List[BaseCallback]]
+        :rtype: Optional[list[BaseCallback]]
         """
         return self._callbacks  # type: ignore
 
     @callbacks.setter
     def callbacks(
         self,
-        value: Optional[Union[BaseCallback, List[BaseCallback]]],
+        value: Optional[Union[BaseCallback, list[BaseCallback]]],
     ) -> None:
         """Callbacks setter.
 
         :param value: value to be set
-        :type value: Optional[Union[BaseCallback, List[Callback]]]
+        :type value: Optional[Union[BaseCallback, list[Callback]]]
         :raises TypeError: Type error exception
         """
         if value is not None:
@@ -58,7 +58,7 @@ def callbacks(
     def reset(self) -> None:
         """Reset method."""
 
-    def _get_callbacks_logs(self) -> Dict[str, Any]:
+    def _get_callbacks_logs(self) -> dict[str, Any]:
         logs = {
             callback.name: callback.logs for callback in self.callbacks  # type: ignore
         }

frouros/detectors/concept_drift/base.py

@@ -1,7 +1,7 @@
 """Base concept drift module."""
 
 import abc
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Optional, Union
 
 from frouros.callbacks import HistoryConceptDrift
 from frouros.callbacks.streaming.base import BaseCallbackStreaming
@@ -66,7 +66,7 @@ def __init__(
         self,
         config: Optional[BaseConceptDriftConfig] = None,
         callbacks: Optional[
-            Union[BaseCallbackStreaming, List[BaseCallbackStreaming]]
+            Union[BaseCallbackStreaming, list[BaseCallbackStreaming]]
         ] = None,
     ) -> None:
         """Init method.
@@ -75,7 +75,7 @@ def __init__(
         :type config: Optional[BaseConceptDriftConfig]
         :param callbacks: callbacks
         :type callbacks: Optional[Union[BaseCallbackStreaming,
-        List[BaseCallbackStreaming]]]
+        list[BaseCallbackStreaming]]]
         """
         check_callbacks(
             callbacks=callbacks,
@@ -99,20 +99,20 @@ def _set_additional_vars_callback(self) -> None:
                 )
 
     @property
-    def additional_vars(self) -> Optional[Dict[str, Any]]:
+    def additional_vars(self) -> Optional[dict[str, Any]]:
         """Additional variables property.
 
         :return: additional variables
-        :rtype: Optional[Dict[str, Any]]
+        :rtype: Optional[dict[str, Any]]
         """
         return self._additional_vars
 
     @additional_vars.setter
-    def additional_vars(self, value: Optional[Dict[str, Any]]) -> None:
+    def additional_vars(self, value: Optional[dict[str, Any]]) -> None:
         """Additional variables setter.
 
         :param value: value to be set
-        :type value: Optional[Dict[str, Any]]
+        :type value: Optional[dict[str, Any]]
         """
         self._additional_vars = value if value is not None else {}
 
@@ -171,21 +171,21 @@ def reset(self) -> None:
             callback.reset()
 
     @property
-    def status(self) -> Dict[str, bool]:
+    def status(self) -> dict[str, bool]:
         """Status property.
 
         :return: status dict
-        :rtype: Dict[str, bool]
+        :rtype: dict[str, bool]
         """
         return {"drift": self.drift}
 
-    def update(self, value: Union[int, float], **kwargs) -> Dict[str, Any]:
+    def update(self, value: Union[int, float], **kwargs) -> dict[str, Any]:
         """Update method.
 
         :param value: value to update detector
         :type value: Union[int, float]
         :return: callbacks logs
-        :rtype: Dict[str, Any]]
+        :rtype: dict[str, Any]]
         """
         for callback in self.callbacks:  # type: ignore
             callback.on_update_start(  # type: ignore
@@ -200,7 +200,7 @@ def update(self, value: Union[int, float], **kwargs) -> Dict[str, Any]:
         callbacks_logs = self._get_callbacks_logs()
         return callbacks_logs
 
-    def _get_callbacks_logs(self) -> Dict[str, Any]:
+    def _get_callbacks_logs(self) -> dict[str, Any]:
         logs = {
             callback.name: callback.logs for callback in self.callbacks  # type: ignore
         }

frouros/detectors/concept_drift/streaming/change_detection/base.py

@@ -1,7 +1,7 @@
 """Base concept drift ChangeDetection based module."""
 
 import abc
-from typing import List, Optional, Union
+from typing import Optional, Union
 
 from frouros.callbacks.streaming.base import BaseCallbackStreaming
 from frouros.detectors.concept_drift.streaming.base import (
@@ -142,7 +142,7 @@ class BaseCUSUM(BaseChangeDetection):
     :param config: configuration parameters, defaults to None
     :type config: Optional[BaseCUSUMConfig]
     :param callbacks: callbacks, defaults to None
-    :type callbacks: Optional[Union[BaseCallbackStreaming, List[BaseCallbackStreaming]]]
+    :type callbacks: Optional[Union[BaseCallbackStreaming, list[BaseCallbackStreaming]]]
     """  # noqa: E501
 
     config_type = BaseCUSUMConfig
@@ -151,7 +151,7 @@ def __init__(  # noqa: D107
         self,
         config: Optional[BaseCUSUMConfig] = None,
         callbacks: Optional[
-            Union[BaseCallbackStreaming, List[BaseCallbackStreaming]]
+            Union[BaseCallbackStreaming, list[BaseCallbackStreaming]]
         ] = None,
     ) -> None:
         super().__init__(

frouros/detectors/concept_drift/streaming/change_detection/bocd.py

@@ -2,7 +2,7 @@
 
 import abc
 import copy
-from typing import List, Union, Optional
+from typing import Union, Optional
 
 import numpy as np  # type: ignore
 from scipy.special import logsumexp  # type: ignore
@@ -185,7 +185,7 @@ class BOCD(BaseChangeDetection):
     :param config: configuration object of the detector, defaults to None. If None, the default configuration of :class:`BOCDConfig` is used.
     :type config: Optional[BOCDConfig]
     :param callbacks: callbacks, defaults to None
-    :type callbacks: Optional[Union[BaseCallbackStreaming, List[BaseCallbackStreaming]]]
+    :type callbacks: Optional[Union[BaseCallbackStreaming, list[BaseCallbackStreaming]]]
 
     :Note:
      Adapted from the implementation in https://github.com/gwgundersen/bocd.
@@ -219,7 +219,7 @@ def __init__(  # noqa: D107
         self,
         config: Optional[BOCDConfig] = None,
         callbacks: Optional[
-            Union[BaseCallbackStreaming, List[BaseCallbackStreaming]]
+            Union[BaseCallbackStreaming, list[BaseCallbackStreaming]]
         ] = None,
     ) -> None:
         super().__init__(

frouros/detectors/concept_drift/streaming/change_detection/cusum.py

@@ -1,6 +1,6 @@
 """CUSUM module."""
 
-from typing import Optional, Union, List
+from typing import Optional, Union
 
 import numpy as np  # type: ignore
 
@@ -47,7 +47,7 @@ class CUSUM(BaseCUSUM):
     :param config: configuration object of the detector, defaults to None. If None, the default configuration of :class:`CUSUMConfig` is used.
     :type config: Optional[CUSUMConfig]
     :param callbacks: callbacks, defaults to None
-    :type callbacks: Optional[Union[BaseCallbackStreaming, List[BaseCallbackStreaming]]]
+    :type callbacks: Optional[Union[BaseCallbackStreaming, list[BaseCallbackStreaming]]]
 
     :References:
 
@@ -78,7 +78,7 @@ def __init__(  # noqa: D107
         self,
         config: Optional[CUSUMConfig] = None,
         callbacks: Optional[
-            Union[BaseCallbackStreaming, List[BaseCallbackStreaming]]
+            Union[BaseCallbackStreaming, list[BaseCallbackStreaming]]
         ] = None,
     ) -> None:
         super().__init__(

frouros/detectors/concept_drift/streaming/change_detection/geometric_moving_average.py

@@ -1,6 +1,6 @@
 """Geometric Moving Average module."""
 
-from typing import Optional, List, Union
+from typing import Optional, Union
 
 from frouros.callbacks.streaming.base import BaseCallbackStreaming
 from frouros.detectors.concept_drift.streaming.change_detection.base import (
@@ -46,7 +46,7 @@ class GeometricMovingAverage(BaseCUSUM):
     :param config: configuration object of the detector, defaults to None. If None, the default configuration of :class:`GeometricMovingAverageConfig` is used.
     :type config: Optional[GeometricMovingAverageConfig]
     :param callbacks: callbacks, defaults to None
-    :type callbacks: Optional[Union[BaseCallbackStreaming, List[BaseCallbackStreaming]]]
+    :type callbacks: Optional[Union[BaseCallbackStreaming, list[BaseCallbackStreaming]]]
 
     :References:
 
@@ -78,7 +78,7 @@ def __init__(  # noqa: D107
         self,
         config: Optional[GeometricMovingAverageConfig] = None,
         callbacks: Optional[
-            Union[BaseCallbackStreaming, List[BaseCallbackStreaming]]
+            Union[BaseCallbackStreaming, list[BaseCallbackStreaming]]
         ] = None,
     ) -> None:
         super().__init__(

frouros/detectors/concept_drift/streaming/change_detection/page_hinkley.py

@@ -1,6 +1,6 @@
 """Page Hinkley module."""
 
-from typing import Optional, Union, List
+from typing import Optional, Union
 
 from frouros.callbacks.streaming.base import BaseCallbackStreaming
 from frouros.detectors.concept_drift.streaming.change_detection.base import (
@@ -50,7 +50,7 @@ class PageHinkley(BaseCUSUM):
     :param config: configuration object of the detector, defaults to None. If None, the default configuration of :class:`PageHinkleyConfig` is used.
     :type config: Optional[PageHinkleyConfig]
     :param callbacks: callbacks, defaults to None
-    :type callbacks: Optional[Union[BaseCallbackStreaming, List[BaseCallbackStreaming]]]
+    :type callbacks: Optional[Union[BaseCallbackStreaming, list[BaseCallbackStreaming]]]
 
     :References:
 
@@ -81,7 +81,7 @@ def __init__(  # noqa: D107
         self,
         config: Optional[PageHinkleyConfig] = None,
         callbacks: Optional[
-            Union[BaseCallbackStreaming, List[BaseCallbackStreaming]]
+            Union[BaseCallbackStreaming, list[BaseCallbackStreaming]]
         ] = None,
     ) -> None:
         super().__init__(