The first hybridisation (#83)

Extended all schedulers with the ability of participate in population-based hybridisation schemes.

Implemented cyclic hybridisation scheme, made experiments.

experiments/hybridisation.py

@@ -0,0 +1,65 @@
+import json
+
+import pathos
+from tqdm import tqdm
+
+import sampo.scheduler
+from sampo.backend.multiproc import MultiprocessingComputationalBackend
+
+from sampo.hybrid.population_tabu import TabuPopulationScheduler
+
+from sampo.hybrid.cycle import CycleHybridScheduler
+from sampo.api.genetic_api import ScheduleGenerationScheme
+from sampo.scheduler import HEFTScheduler, HEFTBetweenScheduler, TopologicalScheduler, GeneticScheduler
+from sampo.hybrid.population import HeuristicPopulationScheduler, GeneticPopulationScheduler
+
+from sampo.generator.environment import get_contractor_by_wg
+from sampo.generator import SimpleSynthetic
+
+from sampo.base import SAMPO
+from sampo.schemas import WorkGraph
+
+def run_experiment(args):
+    graph_size, iteration = args
+
+    heuristics = HeuristicPopulationScheduler([HEFTScheduler(), HEFTBetweenScheduler(), TopologicalScheduler()])
+    # genetic1 = TabuPopulationScheduler()
+    genetic1 = GeneticPopulationScheduler(GeneticScheduler(mutate_order=0.2,
+                                                           mutate_resources=0.2,
+                                                           sgs_type=ScheduleGenerationScheme.Parallel))
+    genetic2 = GeneticPopulationScheduler(GeneticScheduler(mutate_order=0.001,
+                                                           mutate_resources=0.001,
+                                                           sgs_type=ScheduleGenerationScheme.Parallel))
+
+    hybrid_combine = CycleHybridScheduler(heuristics, [genetic1, genetic2], max_plateau_size=1)
+    hybrid_genetic1 = CycleHybridScheduler(heuristics, [genetic1], max_plateau_size=1)
+    hybrid_genetic2 = CycleHybridScheduler(heuristics, [genetic2], max_plateau_size=1)
+
+    wg = WorkGraph.load('wgs', f'{graph_size}_{iteration}')
+    contractors = [get_contractor_by_wg(wg)]
+
+    # SAMPO.backend = MultiprocessingComputationalBackend(n_cpus=10)
+    SAMPO.backend.cache_scheduler_info(wg, contractors)
+    SAMPO.backend.cache_genetic_info()
+
+    schedule_hybrid_combine = hybrid_combine.schedule(wg, contractors)
+    schedule_genetic1 = hybrid_genetic1.schedule(wg, contractors)
+    schedule_genetic2 = hybrid_genetic2.schedule(wg, contractors)
+
+    # print(f'Hybrid combine: {schedule_hybrid_combine.execution_time}')
+    # print(f'Scheduler 1 cycled: {schedule_genetic1.execution_time}')
+    # print(f'Scheduler 2 cycled: {schedule_genetic2.execution_time}')
+    return schedule_hybrid_combine.execution_time, schedule_genetic1.execution_time, schedule_genetic2.execution_time
+
+if __name__ == '__main__':
+    arguments = [(graph_size, iteration) for graph_size in [100, 200, 300, 400, 500] for iteration in range(5)]
+    results = {graph_size: [] for graph_size in [100, 200, 300, 400, 500]}
+
+    with pathos.multiprocessing.Pool(processes=11) as p:
+        r = p.map(run_experiment, arguments)
+
+        for (graph_size, _), (combined_time, time1, time2) in zip(arguments, r):
+            results[graph_size].append((combined_time / time1, combined_time / time2))
+
+    with open('hybrid_results.json', 'w') as f:
+        json.dump(results, f)

experiments/hybridisation_results_proc.py

@@ -0,0 +1,20 @@
+import json
+from itertools import chain
+from random import Random
+
+import matplotlib.pyplot as plt
+
+with open('hybrid_results.json', 'r') as f:
+    results = json.load(f)
+
+results = {int(graph_size): [100 * (1 - res) for res in chain(*result)]
+           for graph_size, result in results.items()}
+graph_sizes = results.keys()
+
+rand = Random()
+
+plt.title('Прирост качества планов\nот применения гибридизации', fontsize=16)
+plt.xlabel('Размер графа')
+plt.ylabel('% прироста качества относительно базового алгоритма')
+plt.boxplot(results.values(), labels=graph_sizes)
+plt.show()

experiments/wg_generate.py

@@ -0,0 +1,14 @@
+from sampo.generator import SimpleSynthetic
+
+from tqdm import tqdm
+
+ss = SimpleSynthetic(rand=231)
+
+for size in range(100, 500 + 1, 100):
+    for i in tqdm(range(100)):
+        wg = ss.work_graph(bottom_border=size - 5,
+                           top_border=size)
+        while not (size - 5 <= wg.vertex_count <= size):
+            wg = ss.work_graph(bottom_border=size - 20,
+                               top_border=size)
+        wg.dump('wgs', f'{size}_{i}')

sampo/backend/__init__.py

@@ -7,6 +7,7 @@
 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
+from sampo.schemas.time_estimator import DefaultWorkEstimator
 
 T = TypeVar('T')
 R = TypeVar('R')
@@ -20,7 +21,7 @@ def __init__(self):
         self._landscape = None
         self._spec = None
         self._rand = Random()
-        self._work_estimator = None
+        self._work_estimator = DefaultWorkEstimator()
 
         # additional genetic parameters
         self._toolbox = None
@@ -41,26 +42,27 @@ def __init__(self):
     def cache_scheduler_info(self,
                              wg: WorkGraph,
                              contractors: list[Contractor],
-                             landscape: LandscapeConfiguration,
-                             spec: ScheduleSpec,
+                             landscape: LandscapeConfiguration = LandscapeConfiguration(),
+                             spec: ScheduleSpec = ScheduleSpec(),
                              rand: Random | None = None,
-                             work_estimator: WorkTimeEstimator | None = None):
+                             work_estimator: WorkTimeEstimator = DefaultWorkEstimator()):
         ...
 
     @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):
+                           population_size: int = 50,
+                           mutate_order: float = 0.1,
+                           mutate_resources: float = 0.05,
+                           mutate_zones: float = 0.05,
+                           deadline: Time | None = None,
+                           weights: list[int] | None = None,
+                           init_schedules: dict[
+                               str, tuple[Schedule, list[GraphNode] | None, ScheduleSpec, float]] = None,
+                           assigned_parent_time: Time = Time(0),
+                           fitness_weights: tuple[int | float, ...] = None,
+                           sgs_type: ScheduleGenerationScheme = ScheduleGenerationScheme.Parallel,
+                           only_lft_initialization: bool = False,
+                           is_multiobjective: bool = False):
         ...
 
     @abstractmethod

sampo/backend/default.py

@@ -18,10 +18,10 @@ def map(self, action: Callable[[T], R], values: list[T]) -> list[R]:
     def cache_scheduler_info(self,
                              wg: WorkGraph,
                              contractors: list[Contractor],
-                             landscape: LandscapeConfiguration,
-                             spec: ScheduleSpec,
+                             landscape: LandscapeConfiguration = LandscapeConfiguration(),
+                             spec: ScheduleSpec = ScheduleSpec(),
                              rand: Random | None = None,
-                             work_estimator: WorkTimeEstimator | None = None):
+                             work_estimator: WorkTimeEstimator = DefaultWorkEstimator()):
         self._wg = wg
         self._contractors = contractors
         self._landscape = landscape
@@ -31,18 +31,18 @@ def cache_scheduler_info(self,
         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):
+                           population_size: int = 50,
+                           mutate_order: float = 0.1,
+                           mutate_resources: float = 0.05,
+                           mutate_zones: float = 0.05,
+                           deadline: Time | None = None,
+                           weights: list[int] | None = None,
+                           init_schedules: dict[str, tuple[Schedule, list[GraphNode] | None, ScheduleSpec, float]] = None,
+                           assigned_parent_time: Time = Time(0),
+                           fitness_weights: tuple[int | float, ...] = None,
+                           sgs_type: ScheduleGenerationScheme = ScheduleGenerationScheme.Parallel,
+                           only_lft_initialization: bool = False,
+                           is_multiobjective: bool = False):
         self._selection_size = population_size
         self._mutate_order = mutate_order
         self._mutate_resources = mutate_resources
@@ -61,8 +61,11 @@ 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)
+            if self._init_schedules:
+                init_chromosomes = init_chromosomes_f(self._wg, self._contractors, self._init_schedules,
+                                                      self._landscape)
+            else:
+                init_chromosomes = []
 
             rand = self._rand or Random()
             work_estimator = self._work_estimator or DefaultWorkEstimator()

sampo/backend/multiproc.py

@@ -115,30 +115,35 @@ def _ensure_pool_created(self):
     def cache_scheduler_info(self,
                              wg: WorkGraph,
                              contractors: list[Contractor],
-                             landscape: LandscapeConfiguration,
-                             spec: ScheduleSpec,
+                             landscape: LandscapeConfiguration = LandscapeConfiguration(),
+                             spec: ScheduleSpec = ScheduleSpec(),
                              rand: Random | None = None,
                              work_estimator: WorkTimeEstimator = DefaultWorkEstimator()):
         super().cache_scheduler_info(wg, contractors, landscape, spec, rand, work_estimator)
         self._pool = None
 
     def cache_genetic_info(self,
-                           selection_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):
-        super().cache_genetic_info(selection_size, mutate_order, mutate_resources, mutate_zones, deadline,
+                           population_size: int = 50,
+                           mutate_order: float = 0.1,
+                           mutate_resources: float = 0.05,
+                           mutate_zones: float = 0.05,
+                           deadline: Time | None = None,
+                           weights: list[int] | None = None,
+                           init_schedules: dict[
+                               str, tuple[Schedule, list[GraphNode] | None, ScheduleSpec, float]] = None,
+                           assigned_parent_time: Time = Time(0),
+                           fitness_weights: tuple[int | float, ...] = None,
+                           sgs_type: ScheduleGenerationScheme = ScheduleGenerationScheme.Parallel,
+                           only_lft_initialization: bool = False,
+                           is_multiobjective: bool = False):
+        super().cache_genetic_info(population_size, mutate_order, mutate_resources, mutate_zones, deadline,
                                    weights, init_schedules, assigned_parent_time, fitness_weights, sgs_type,
                                    only_lft_initialization, is_multiobjective)
-        self._init_chromosomes = init_chromosomes_f(self._wg, self._contractors, init_schedules, self._landscape)
+        if init_schedules:
+            self._init_chromosomes = init_chromosomes_f(self._wg, self._contractors, init_schedules,
+                                                  self._landscape)
+        else:
+            self._init_chromosomes = []
         self._pool = None
 
     def compute_chromosomes(self, fitness: FitnessFunction, chromosomes: list[ChromosomeType]) -> list[float]:

sampo/hybrid/cycle.py

@@ -0,0 +1,71 @@
+import numpy as np
+
+from sampo.api.genetic_api import FitnessFunction, ChromosomeType, ScheduleGenerationScheme
+from sampo.base import SAMPO
+from sampo.hybrid.population import PopulationScheduler
+from sampo.scheduler.genetic import TimeFitness
+from sampo.scheduler.genetic.schedule_builder import create_toolbox
+from sampo.schemas import WorkGraph, Contractor, Time, LandscapeConfiguration, Schedule
+from sampo.schemas.schedule_spec import ScheduleSpec
+
+
+class CycleHybridScheduler:
+    def __init__(self,
+                 starting_scheduler: PopulationScheduler,
+                 cycle_schedulers: list[PopulationScheduler],
+                 fitness: FitnessFunction = TimeFitness(),
+                 max_plateau_size: int = 2):
+        self._starting_scheduler = starting_scheduler
+        self._cycle_schedulers = cycle_schedulers
+        self._fitness = fitness
+        self._max_plateau_size = max_plateau_size
+
+    def _get_population_fitness(self, pop: list[ChromosomeType]):
+        # return best chromosome's fitness
+        return min(SAMPO.backend.compute_chromosomes(self._fitness, pop))
+
+    def _get_best_individual(self, pop: list[ChromosomeType]) -> ChromosomeType:
+        fitness = SAMPO.backend.compute_chromosomes(self._fitness, pop)
+        return pop[np.argmin(fitness)]
+
+    def run(self,
+            wg: WorkGraph,
+            contractors: list[Contractor],
+            spec: ScheduleSpec = ScheduleSpec(),
+            assigned_parent_time: Time = Time(0),
+            landscape: LandscapeConfiguration = LandscapeConfiguration()) -> ChromosomeType:
+        pop = self._starting_scheduler.schedule([], wg, contractors, spec, assigned_parent_time, landscape)
+
+        cur_fitness = Time.inf().value
+        plateau_steps = 0
+
+        while True:
+            pop_fitness = self._get_population_fitness(pop)
+            if pop_fitness == cur_fitness:
+                plateau_steps += 1
+                if plateau_steps == self._max_plateau_size:
+                    break
+            else:
+                plateau_steps = 0
+                cur_fitness = pop_fitness
+
+            for scheduler in self._cycle_schedulers:
+                pop = scheduler.schedule(pop, wg, contractors, spec, assigned_parent_time, landscape)
+
+        return self._get_best_individual(pop)
+
+    def schedule(self,
+                 wg: WorkGraph,
+                 contractors: list[Contractor],
+                 spec: ScheduleSpec = ScheduleSpec(),
+                 assigned_parent_time: Time = Time(0),
+                 sgs_type: ScheduleGenerationScheme = ScheduleGenerationScheme.Parallel,
+                 landscape: LandscapeConfiguration = LandscapeConfiguration()) -> Schedule:
+        best_ind = self.run(wg, contractors, spec, assigned_parent_time, landscape)
+
+        toolbox = create_toolbox(wg=wg, contractors=contractors, landscape=landscape,
+                                 assigned_parent_time=assigned_parent_time, spec=spec,
+                                 sgs_type=sgs_type)
+        node2swork = toolbox.chromosome_to_schedule(best_ind)[0]
+
+        return Schedule.from_scheduled_works(node2swork.values(), wg)

sampo/hybrid/population.py

@@ -0,0 +1,54 @@
+from abc import ABC, abstractmethod
+
+from sampo.api.genetic_api import ChromosomeType, Individual
+from sampo.scheduler import GeneticScheduler, Scheduler
+from sampo.scheduler.genetic.schedule_builder import create_toolbox
+from sampo.schemas import WorkGraph, Contractor, Time, LandscapeConfiguration
+from sampo.schemas.schedule_spec import ScheduleSpec
+
+
+class PopulationScheduler(ABC):
+
+    @abstractmethod
+    def schedule(self,
+                 initial_population: list[ChromosomeType],
+                 wg: WorkGraph,
+                 contractors: list[Contractor],
+                 spec: ScheduleSpec = ScheduleSpec(),
+                 assigned_parent_time: Time = Time(0),
+                 landscape: LandscapeConfiguration = LandscapeConfiguration()) -> list[ChromosomeType]:
+        ...
+
+
+class GeneticPopulationScheduler(PopulationScheduler):
+    def __init__(self, genetic: GeneticScheduler):
+        self._genetic = genetic
+
+    def schedule(self,
+                 initial_population: list[ChromosomeType],
+                 wg: WorkGraph,
+                 contractors: list[Contractor],
+                 spec: ScheduleSpec = ScheduleSpec(),
+                 assigned_parent_time: Time = Time(0),
+                 landscape: LandscapeConfiguration = LandscapeConfiguration()) -> list[ChromosomeType]:
+        return self._genetic.upgrade_pop(wg, contractors, initial_population, spec,
+                                         assigned_parent_time, landscape=landscape)
+
+
+class HeuristicPopulationScheduler(PopulationScheduler):
+    def __init__(self, schedulers: list[Scheduler]):
+        self._schedulers = schedulers
+
+    def schedule(self,
+                 initial_population: list[ChromosomeType],
+                 wg: WorkGraph,
+                 contractors: list[Contractor],
+                 spec: ScheduleSpec = ScheduleSpec(),
+                 assigned_parent_time: Time = Time(0),
+                 landscape: LandscapeConfiguration = LandscapeConfiguration()) -> list[ChromosomeType]:
+        toolbox = create_toolbox(wg=wg, contractors=contractors,
+                                 spec=spec, assigned_parent_time=assigned_parent_time,
+                                 landscape=landscape)
+        return [toolbox.Individual(toolbox.schedule_to_chromosome(schedule=schedule))
+                for scheduler in self._schedulers
+                for schedule in scheduler.schedule(wg, contractors, spec, landscape=landscape)]