Merge remote-tracking branch 'origin/main' into refactor/structurator

sampo/api/genetic_api.py

@@ -0,0 +1,51 @@
+from abc import ABC, abstractmethod
+from enum import Enum
+from functools import partial
+from typing import Callable
+
+import numpy as np
+from deap import base, creator
+
+from sampo.schemas import Schedule
+from sampo.schemas.schedule_spec import ScheduleSpec
+
+ChromosomeType = tuple[np.ndarray, np.ndarray, np.ndarray, ScheduleSpec, np.ndarray]
+
+class ScheduleGenerationScheme(Enum):
+    Parallel = 'Parallel'
+    Serial = 'Serial'
+
+
+class FitnessFunction(ABC):
+    """
+    Base class for description of different fitness functions.
+    """
+
+    @abstractmethod
+    def evaluate(self, chromosome: ChromosomeType, evaluator: Callable[[ChromosomeType], Schedule]) \
+            -> tuple[int | float]:
+        """
+        Calculate the value of fitness function of the chromosome.
+        It is better when value is less.
+        """
+        ...
+
+
+# create class FitnessMin, the weights = -1 means that fitness - is function for minimum
+
+# creator.create('FitnessMin', base.Fitness, weights=(-1.0,))
+# creator.create('Individual', list, fitness=creator.FitnessMin)
+# Individual = creator.Individual
+
+class Individual(list):
+    def __init__(self, individual_fitness_constructor: Callable[[], base.Fitness], chromosome: ChromosomeType):
+        super().__init__(chromosome)
+        self.fitness = individual_fitness_constructor()
+
+    @staticmethod
+    def prepare(individual_fitness_constructor: Callable[[], base.Fitness]) -> Callable[[ChromosomeType], list]:
+        """
+        Returns the constructor of Individual prepared to use in Genetic algorithm
+        """
+        return partial(Individual, individual_fitness_constructor)
+

sampo/backend/__init__.py

@@ -0,0 +1,74 @@
+from abc import ABC, abstractmethod
+from random import Random
+from typing import TypeVar
+
+# import sampo.scheduler
+
+from sampo.api.genetic_api import ChromosomeType, FitnessFunction, Individual, ScheduleGenerationScheme
+from sampo.schemas import WorkGraph, Contractor, LandscapeConfiguration, Schedule, GraphNode, Time, WorkTimeEstimator
+from sampo.schemas.schedule_spec import ScheduleSpec
+
+T = TypeVar('T')
+R = TypeVar('R')
+
+
+class ComputationalBackend(ABC):
+    def __init__(self):
+        # scheduler parameters
+        self._wg = None
+        self._contractors = None
+        self._landscape = None
+        self._spec = None
+        self._rand = Random()
+        self._work_estimator = None
+
+        # additional genetic parameters
+        self._toolbox = None
+        self._selection_size = None
+        self._mutate_order = None
+        self._mutate_resources = None
+        self._mutate_zones = None
+        self._deadline = None
+        self._weights = None
+        self._init_schedules = None
+        self._assigned_parent_time = None
+        self._fitness_weights = None
+        self._sgs_type = None
+        self._only_lft_initialization = None
+        self._is_multiobjective = None
+
+    @abstractmethod
+    def cache_scheduler_info(self,
+                             wg: WorkGraph,
+                             contractors: list[Contractor],
+                             landscape: LandscapeConfiguration,
+                             spec: ScheduleSpec,
+                             rand: Random | None = None,
+                             work_estimator: WorkTimeEstimator | None = None):
+        ...
+
+    @abstractmethod
+    def cache_genetic_info(self,
+                           population_size: int,
+                           mutate_order: float,
+                           mutate_resources: float,
+                           mutate_zones: float,
+                           deadline: Time | None,
+                           weights: list[int] | None,
+                           init_schedules: dict[str, tuple[Schedule, list[GraphNode] | None, ScheduleSpec, float]],
+                           assigned_parent_time: Time,
+                           fitness_weights: tuple[int | float, ...],
+                           sgs_type: ScheduleGenerationScheme,
+                           only_lft_initialization: bool,
+                           is_multiobjective: bool):
+        ...
+
+    @abstractmethod
+    def compute_chromosomes(self,
+                            fitness: FitnessFunction,
+                            chromosomes: list[ChromosomeType]) -> list[float]:
+        ...
+
+    @abstractmethod
+    def generate_first_population(self, size_population: int) -> list[Individual]:
+        ...

sampo/backend/default.py

@@ -0,0 +1,96 @@
+from random import Random
+from typing import Callable
+
+import sampo.scheduler
+
+from sampo.api.genetic_api import FitnessFunction, ChromosomeType, Individual, ScheduleGenerationScheme
+from sampo.backend import ComputationalBackend, T, R
+from sampo.schemas import WorkGraph, Contractor, LandscapeConfiguration, WorkTimeEstimator, Schedule, GraphNode, Time
+from sampo.schemas.schedule_spec import ScheduleSpec
+from sampo.schemas.time_estimator import DefaultWorkEstimator
+
+
+class DefaultComputationalBackend(ComputationalBackend):
+
+    def map(self, action: Callable[[T], R], values: list[T]) -> list[R]:
+        return [action(v) for v in values]
+
+    def cache_scheduler_info(self,
+                             wg: WorkGraph,
+                             contractors: list[Contractor],
+                             landscape: LandscapeConfiguration,
+                             spec: ScheduleSpec,
+                             rand: Random | None = None,
+                             work_estimator: WorkTimeEstimator | None = None):
+        self._wg = wg
+        self._contractors = contractors
+        self._landscape = landscape
+        self._spec = spec
+        self._rand = rand
+        self._work_estimator = work_estimator
+        self._toolbox = None
+
+    def cache_genetic_info(self,
+                           population_size: int,
+                           mutate_order: float,
+                           mutate_resources: float,
+                           mutate_zones: float,
+                           deadline: Time | None,
+                           weights: list[int] | None,
+                           init_schedules: dict[str, tuple[Schedule, list[GraphNode] | None, ScheduleSpec, float]],
+                           assigned_parent_time: Time,
+                           fitness_weights: tuple[int | float, ...],
+                           sgs_type: ScheduleGenerationScheme,
+                           only_lft_initialization: bool,
+                           is_multiobjective: bool):
+        self._selection_size = population_size
+        self._mutate_order = mutate_order
+        self._mutate_resources = mutate_resources
+        self._mutate_zones = mutate_zones
+        self._deadline = deadline
+        self._weights = weights
+        self._init_schedules = init_schedules
+        self._assigned_parent_time = assigned_parent_time
+        self._fitness_weights = fitness_weights
+        self._sgs_type = sgs_type
+        self._only_lft_initialization = only_lft_initialization
+        self._is_multiobjective = is_multiobjective
+        self._toolbox = None
+
+    def _ensure_toolbox_created(self):
+        if self._toolbox is None:
+            from sampo.scheduler.genetic.utils import init_chromosomes_f, create_toolbox_using_cached_chromosomes
+
+            init_chromosomes = init_chromosomes_f(self._wg, self._contractors, self._init_schedules,
+                                                  self._landscape)
+
+            rand = self._rand or Random()
+            work_estimator = self._work_estimator or DefaultWorkEstimator()
+            assigned_parent_time = self._assigned_parent_time or Time(0)
+
+            self._toolbox = create_toolbox_using_cached_chromosomes(self._wg,
+                                                                    self._contractors,
+                                                                    self._selection_size,
+                                                                    self._mutate_order,
+                                                                    self._mutate_resources,
+                                                                    self._mutate_zones,
+                                                                    init_chromosomes,
+                                                                    rand,
+                                                                    self._spec,
+                                                                    work_estimator,
+                                                                    assigned_parent_time,
+                                                                    self._fitness_weights,
+                                                                    self._landscape,
+                                                                    self._sgs_type,
+                                                                    self._only_lft_initialization,
+                                                                    self._is_multiobjective)
+
+    def compute_chromosomes(self,
+                            fitness: FitnessFunction,
+                            chromosomes: list[ChromosomeType]) -> list[tuple[int | float]]:
+        self._ensure_toolbox_created()
+        return [fitness.evaluate(chromosome, self._toolbox.evaluate_chromosome) for chromosome in chromosomes]
+
+    def generate_first_population(self, size_population: int) -> list[Individual]:
+        self._ensure_toolbox_created()
+        return self._toolbox.population(size_population)