Finished cyclic hybridisation, experiments implemented

Author: StannisMod

experiments/hybridisation.ipynb

@@ -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.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/hybridisation_results_proc.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}')

experiments/wg_generate.py

@@ -139,7 +139,11 @@ def cache_genetic_info(self,
         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/backend/multiproc.py

@@ -39,10 +39,12 @@ def run(self,
         cur_fitness = Time.inf().value
         plateau_steps = 0
 
-        while plateau_steps < self._max_plateau_size:
+        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
@@ -59,23 +61,7 @@ def schedule(self,
                  assigned_parent_time: Time = Time(0),
                  sgs_type: ScheduleGenerationScheme = ScheduleGenerationScheme.Parallel,
                  landscape: LandscapeConfiguration = LandscapeConfiguration()) -> Schedule:
-        pop = self._starting_scheduler.schedule([], wg, contractors, spec, assigned_parent_time, landscape)
-
-        cur_fitness = Time.inf().value
-        plateau_steps = 0
-
-        while plateau_steps < self._max_plateau_size:
-            pop_fitness = self._get_population_fitness(pop)
-            if pop_fitness == cur_fitness:
-                plateau_steps += 1
-            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)
-
-        best_ind = self._get_best_individual(pop)
+        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,

sampo/hybrid/cycle.py

@@ -0,0 +1,56 @@
+import numpy as np
+
+from sampo.api.genetic_api import ChromosomeType, FitnessFunction
+from sampo.base import SAMPO
+from sampo.hybrid.population import PopulationScheduler
+from sampo.scheduler.genetic import TimeFitness
+from sampo.scheduler.utils import get_worker_contractor_pool
+from sampo.schemas import WorkGraph, Contractor, Time, LandscapeConfiguration
+from sampo.schemas.schedule_spec import ScheduleSpec
+
+from sampo.scheduler.genetic.converter import ChromosomeType
+from tabusearch.experiments.scheduling.fixtures import setup_toolbox
+
+from tabusearch.experiments.scheduling.scheduling_utils import get_optimiser, OptimiserLifetime
+from tabusearch.utility.chromosome import ChromosomeRW
+
+
+class TabuPopulationScheduler(PopulationScheduler):
+
+    def __init__(self, fitness: FitnessFunction = TimeFitness()):
+        self._fitness = fitness
+
+    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], fitness: list[float] | None = None) -> ChromosomeType:
+        fitness = fitness or SAMPO.backend.compute_chromosomes(self._fitness, pop)
+        return pop[np.argmin(fitness)]
+
+    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]:
+        tabu_toolbox = setup_toolbox(wg, contractors, get_worker_contractor_pool(contractors))
+
+        leader = self._get_best_individual(initial_population)
+        fitness = SAMPO.backend.compute_chromosomes(TimeFitness(), [leader])
+        print(f'TABU initial fitness: {fitness}')
+
+        tabu_leader = ChromosomeRW.from_sampo_chromosome(leader)
+
+        opt_ord, opt_res = get_optimiser(tabu_toolbox,
+                                         use_vp=True,
+                                         optimisers_lifetime=OptimiserLifetime.Short)
+        tabu_leader = opt_ord.optimize(tabu_leader)
+        tabu_leader = opt_res.optimize(tabu_leader.position)
+
+        chromosome = tabu_leader.position.to_sampo_chromosome()
+        fitness = SAMPO.backend.compute_chromosomes(TimeFitness(), [chromosome])
+        print(f'TABU fitness: {fitness}')
+
+        return initial_population + [chromosome]

sampo/hybrid/population_tabu.py

@@ -102,6 +102,9 @@ def optimize_resources_using_spec(work_unit: WorkUnit, worker_team: list[Worker]
             data in its closure and run optimization process when receives `optimize_array`.
             Passing None or default value means this function should only apply spec.
         """
+        worker_reqs = set(wr.kind for wr in work_unit.worker_reqs)
+        worker_team = [worker for worker in worker_team if worker.name in worker_reqs]
+
         if len(work_spec.assigned_workers) == len(work_unit.worker_reqs):
             # all resources passed in spec, skipping optimize_resources step
             for worker in worker_team:

sampo/scheduler/base.py

@@ -130,7 +130,7 @@ def build_schedules_with_cache(wg: WorkGraph,
                                init_schedules: dict[str, tuple[Schedule, list[GraphNode] | None, ScheduleSpec, float]],
                                rand: random.Random,
                                spec: ScheduleSpec,
-                               weights: list[int],
+                               weights: list[int] = None,
                                pop: list[ChromosomeType] = None,
                                landscape: LandscapeConfiguration = LandscapeConfiguration(),
                                fitness_object: FitnessFunction = TimeFitness(),
@@ -176,15 +176,19 @@ def build_schedules_with_cache(wg: WorkGraph,
                                      init_schedules, assigned_parent_time, fitness_weights,
                                      sgs_type, only_lft_initialization, is_multiobjective)
 
-    # create population of a given size
-    pop = pop or SAMPO.backend.generate_first_population(population_size)
-
     SAMPO.logger.info(f'Toolbox initialization & first population took {(time.time() - start) * 1000} ms')
 
     have_deadline = deadline is not None
     fitness_f = fitness_object if not have_deadline else TimeFitness()
     if have_deadline:
         toolbox.register_individual_constructor((-1,))
+
+    # create population of a given size
+    if pop is None:
+        pop = SAMPO.backend.generate_first_population(population_size)
+    else:
+        pop = [toolbox.Individual(chromosome) for chromosome in pop]
+
     evaluation_start = time.time()
 
     hof = tools.ParetoFront(similar=compare_individuals)