fix: mypy

Author: iraedeus

.github/workflows/check.yml

@@ -19,18 +19,19 @@ jobs:
           python-version: ${{ matrix.python-version }}
 
       - name: Install poetry
-        run: |
-          pipx install poetry
+        run: pipx install poetry
 
       - name: Install dependencies
-        run: |
-          poetry install --with dev
+        run: poetry install --with dev
 
       - name: Ruff
         run: poetry run ruff check
 
+      - name: Install stubs
+        run: poetry run pip install types-PyYAML types-tqdm
+
       - name: Mypy
-        run: poetry run mypy .
+        run: poetry run mypy
 
       - name: Pytest
         run:  poetry run python -m pytest tests

experimental_env/analysis/analyze_summarizers/error_summarizer.py

@@ -39,7 +39,7 @@ def calculate(self, results: list[ExperimentDescription]) -> tuple:
             errors.append(error)
 
         mean = np.sum(errors) / len(errors)
-        standart_deviation = np.sqrt(np.sum((x - mean) ** 2 for x in errors) / len(errors))
+        standart_deviation = np.sqrt(np.sum([(x - mean) ** 2 for x in errors]) / len(errors))
 
         errors.sort()
         median = errors[len(errors) // 2]

experimental_env/analysis/analyze_summarizers/time_summarizer.py

@@ -23,11 +23,11 @@ def calculate(self, results: list[ExperimentDescription]) -> tuple:
         """
         times = []
         for result in results:
-            time = np.sum(step.time for step in result.steps)
+            time = np.sum([step.time for step in result.steps])
             times.append(time)
 
         mean = np.sum(times) / len(times)
-        deviation = np.sqrt(np.sum((x - mean) ** 2 for x in times) / len(times))
+        deviation = np.sqrt(np.sum([(x - mean) ** 2 for x in times]) / len(times))
         return float(mean), float(deviation)
 
     def analyze_method(self, results: list[ExperimentDescription], method: str):

experimental_env/experiment/estimators.py

@@ -16,7 +16,7 @@
 from mpest.optimizers import ALL_OPTIMIZERS
 from mpest.utils import ANamed, Factory, ResultWithLog
 
-METHODS = {
+METHODS: dict = {
     "Likelihood": [[Factory(BayesEStep), Factory(LikelihoodMStep, optimizer)] for optimizer in ALL_OPTIMIZERS],
     "L-moments": [Factory(IndicatorEStep), Factory(LMomentsMStep)],
 }

experimental_env/mixture_generators/abstract_generator.py

@@ -16,7 +16,7 @@ def __init__(self, seed: int = 42):
         random.seed(seed)
 
     @abstractmethod
-    def generate_priors(self, models: list[type[AModel]]) -> list[float]:
+    def generate_priors(self, models: list[type[AModel]]) -> list[float | None]:
         """
         A method for choosing how a prior probabilities will be generated.
         The function is needed so that, with a uniform distribution, degenerate distributions are not generated.

experimental_env/preparation/dataset_description.py

@@ -1,6 +1,7 @@
 """Module which describes dataset"""
 
 import copy
+from typing import Any
 
 from mpest.annotations import Samples
 from mpest.core.mixture_distribution import MixtureDistribution
@@ -61,9 +62,9 @@ def to_yaml_format(self) -> dict:
         """
         Convert info about mixture to yaml format
         """
-        # pylint: disable=duplicate-code
 
-        output = {}
+        output: dict[str, Any] = {}
+
         # Add name
         output["name"] = self.get_dataset_name()
         output["samples_size"] = self._samples_size

experimental_env/preparation/dataset_generator.py

@@ -60,8 +60,8 @@ class ConcreteDatasetGenerator:
 
     def __init__(self, seed: int = 42):
         np.random.seed(seed)
-        self._dists = []
-        self._priors = []
+        self._dists: list[Distribution] = []
+        self._priors: list[float | None] = []
 
     def add_distribution(self, model: type[AModel], params: list[float], prior: float) -> None:
         """

mm.py

@@ -0,0 +1,66 @@
+"""The script implements the first step of the experiment"""
+
+from pathlib import Path
+
+import numpy as np
+
+from experimental_env.preparation.dataset_generator import (
+    ConcreteDatasetGenerator,
+    RandomDatasetGenerator,
+)
+from mpest.models import ExponentialModel, GaussianModel, WeibullModelExp
+
+WORKING_DIR = Path("/home/iraedeus/Projects/stage_1")
+SAMPLES_SIZE = 1000
+
+np.random.seed(42)
+
+r_generator = RandomDatasetGenerator()
+mixtures = [
+    [ExponentialModel],
+    [GaussianModel],
+    [WeibullModelExp],
+    [WeibullModelExp, GaussianModel],
+    [ExponentialModel, GaussianModel],
+    [WeibullModelExp, WeibullModelExp],
+    [ExponentialModel, ExponentialModel],
+]
+for models in mixtures:
+    r_generator.generate(SAMPLES_SIZE, models, Path(WORKING_DIR), exp_count=100)
+
+c_generator2 = ConcreteDatasetGenerator()
+models = [ExponentialModel]
+c_generator2.add_distribution(models[0], [1.0], 1.0)
+c_generator2.generate(SAMPLES_SIZE, Path(WORKING_DIR), 5)
+
+c_generator3 = ConcreteDatasetGenerator()
+models = [GaussianModel]
+c_generator3.add_distribution(models[0], [0, 1.0], 1.0)
+c_generator3.generate(SAMPLES_SIZE, Path(WORKING_DIR), 5)
+
+c_generator4 = ConcreteDatasetGenerator()
+models = [WeibullModelExp]
+c_generator4.add_distribution(models[0], [1.0, 1.0], 1.0)
+c_generator4.generate(SAMPLES_SIZE, Path(WORKING_DIR), 5)
+
+c_generator5 = ConcreteDatasetGenerator()
+models = [WeibullModelExp]
+c_generator5.add_distribution(models[0], [1.0, 1.0], 1.0)
+c_generator5.generate(SAMPLES_SIZE, Path(WORKING_DIR), 5)
+
+c_generator6 = ConcreteDatasetGenerator()
+models = [WeibullModelExp]
+c_generator6.add_distribution(models[0], [1.0, 0.5], 1.0)
+c_generator6.generate(SAMPLES_SIZE, Path(WORKING_DIR), 5)
+
+c_generator7 = ConcreteDatasetGenerator()
+models = [GaussianModel, GaussianModel]
+c_generator7.add_distribution(models[0], [-1.0, 2.5], 0.3)
+c_generator7.add_distribution(models[1], [1.0, 0.5], 0.7)
+c_generator7.generate(SAMPLES_SIZE, Path(WORKING_DIR), 10)
+
+c_generator8 = ConcreteDatasetGenerator()
+models = [GaussianModel, GaussianModel]
+c_generator8.add_distribution(models[0], [0.0, 1.5], 0.6)
+c_generator8.add_distribution(models[1], [1.0, 1.0], 0.4)
+c_generator8.generate(SAMPLES_SIZE, Path(WORKING_DIR), 10)

mpest/core/mixture_distribution.py

@@ -122,11 +122,11 @@ def generate(self, size=1) -> Params:
             )
         )
 
-        x = []
+        temp = []
         for i, model in enumerate(models):
-            x += list(model[0].generate(x_models.count(i)))
+            temp += list(model[0].generate(x_models.count(i)))
 
-        x = np.array(x)
+        x = np.array(temp)
         np.random.shuffle(x)
 
         return x

mpest/em/em.py

@@ -246,7 +246,7 @@ def postprocess_result(result: ResultWithError) -> ResultWithError:
 
             return ResultWithError(new_mixture, result.error)
 
-        history = []
+        history: list = []
 
         def log_map(distributions: ResultWithError[EM._DistributionMixtureAlive]):
             return ResultWithError(