Critical path integration (#94)

* added critical path calculation
* removed NONE type of lag optimization
* added contractor auto-generation to pipeline

Author: StannisMod

.github/workflows/test.yaml

@@ -24,7 +24,7 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install pytest pytest-rerunfailures
+          pip install pytest==7.2.0 pytest-rerunfailures==12.0
           if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
 #      - name: Lint with flake8
 #        run: |

experiments/critical_path.py

@@ -0,0 +1,29 @@
+from sampo.scheduler import GeneticScheduler
+from sampo.generator.base import SimpleSynthetic
+from sampo.generator.environment.contractor_by_wg import get_contractor_by_wg, ContractorGenerationMethod
+from sampo.generator import SyntheticGraphType
+from sampo.pipeline import SchedulingPipeline
+from sampo.scheduler.heft.base import HEFTScheduler
+
+scheduler = HEFTScheduler()
+
+r_seed = 231
+ss = SimpleSynthetic(r_seed)
+
+simple_wg = ss.work_graph(mode=SyntheticGraphType.GENERAL,
+                          cluster_counts=10,
+                          bottom_border=100,
+                          top_border=200)
+
+contractors = [get_contractor_by_wg(simple_wg, 10, ContractorGenerationMethod.MAX)]
+
+project = SchedulingPipeline.create() \
+    .wg('9-1-ukpg-full-with-priority.csv', sep=';', all_connections=True) \
+    .contractors((ContractorGenerationMethod.MAX, 1, 1000)) \
+    .schedule(scheduler) \
+    .visualization('2022-01-01')[0] \
+    .shape((14, 14)) \
+    .color_type('critical_path') \
+    .show_gant_chart()
+
+schedule = project.schedule

experiments/critical_path_functions.py

@@ -0,0 +1,31 @@
+from sampo.generator.base import SimpleSynthetic
+from sampo.generator.environment.contractor_by_wg import get_contractor_by_wg, ContractorGenerationMethod
+from sampo.generator import SyntheticGraphType
+from sampo.pipeline import SchedulingPipeline
+from sampo.scheduler.heft.base import HEFTScheduler
+from sampo.scheduler.utils.critical_path import critical_path_graph
+from sampo.schemas.time_estimator import DefaultWorkEstimator
+
+scheduler = HEFTScheduler()
+
+r_seed = 231
+ss = SimpleSynthetic(r_seed)
+
+simple_wg = ss.work_graph(mode=SyntheticGraphType.GENERAL,
+                          cluster_counts=10,
+                          bottom_border=100,
+                          top_border=200)
+
+contractors = [get_contractor_by_wg(simple_wg, 10, ContractorGenerationMethod.MAX)]
+
+project = SchedulingPipeline.create() \
+    .wg('9-1-ukpg-full-with-priority.csv', sep=';', all_connections=True) \
+    .contractors((ContractorGenerationMethod.MAX, 1, 1000)) \
+    .schedule(scheduler) \
+    .finish()[0]
+
+schedule = project.schedule
+
+graph_cp = critical_path_graph(project.wg.nodes, work_estimator=DefaultWorkEstimator())
+
+schedule_cp = project.critical_path()

sampo/pipeline/base.py

@@ -2,6 +2,7 @@
 
 import pandas as pd
 
+from sampo.generator.environment import ContractorGenerationMethod
 from sampo.pipeline.lag_optimization import LagOptimizationStrategy
 from sampo.scheduler.base import Scheduler
 from sampo.scheduler.utils.local_optimization import OrderLocalOptimizer, ScheduleLocalOptimizer
@@ -28,7 +29,7 @@ def wg(self, wg: WorkGraph | pd.DataFrame | str,
         ...
 
     @abstractmethod
-    def contractors(self, contractors: list[Contractor] | pd.DataFrame | str) -> 'InputPipeline':
+    def contractors(self, contractors: list[Contractor] | pd.DataFrame | str | tuple[ContractorGenerationMethod, int, int]) -> 'InputPipeline':
         ...
 
     @abstractmethod

sampo/pipeline/default.py

@@ -1,6 +1,6 @@
 import pandas as pd
 
-from sampo.generator.environment import ContractorGenerationMethod
+from sampo.generator.environment import ContractorGenerationMethod, get_contractor_by_wg
 from sampo.pipeline.base import InputPipeline, SchedulePipeline
 from sampo.pipeline.delegating import DelegatingScheduler
 from sampo.pipeline.lag_optimization import LagOptimizationStrategy
@@ -55,11 +55,11 @@ class DefaultInputPipeline(InputPipeline):
 
     def __init__(self):
         self._wg: WorkGraph | pd.DataFrame | str | None = None
-        self._contractors: list[Contractor] | pd.DataFrame | str | tuple[ContractorGenerationMethod, int] | None \
-            = ContractorGenerationMethod.AVG, 1
+        self._contractors: list[Contractor] | pd.DataFrame | str | tuple[ContractorGenerationMethod, int, int] | None \
+            = ContractorGenerationMethod.AVG, 1, 1
         self._work_estimator: WorkTimeEstimator = DefaultWorkEstimator()
         self._node_orders: list[list[GraphNode]] | None = None
-        self._lag_optimize: LagOptimizationStrategy = LagOptimizationStrategy.NONE
+        self._lag_optimize: LagOptimizationStrategy = LagOptimizationStrategy.FALSE
         self._spec: ScheduleSpec | None = ScheduleSpec()
         self._assigned_parent_time: Time | None = Time(0)
         self._local_optimize_stack: ApplyQueue = ApplyQueue()
@@ -98,7 +98,7 @@ def wg(self,
         self.sep_wg = sep
         return self
 
-    def contractors(self, contractors: list[Contractor] | pd.DataFrame | str | tuple[ContractorGenerationMethod, int]) \
+    def contractors(self, contractors: list[Contractor] | pd.DataFrame | str | tuple[ContractorGenerationMethod, int, float]) \
             -> 'InputPipeline':
         """
         Mandatory argument.
@@ -206,6 +206,15 @@ def schedule(self, scheduler: Scheduler, validate: bool = False) -> 'SchedulePip
 
         check_and_correct_priorities(self._wg)
 
+        if not isinstance(self._contractors, list):
+            generation_method, contractors_number, scaler = self._contractors
+            self._contractors = [get_contractor_by_wg(self._wg,
+                                                      method=generation_method,
+                                                      contractor_id=str(i),
+                                                      contractor_name='Contractor' + ' ' + str(i + 1),
+                                                      scaler=scaler)
+                           for i in range(contractors_number)]
+
         if not contractors_can_perform_work_graph(self._contractors, self._wg):
             raise NoSufficientContractorError('Contractors are not able to perform the graph of works')
 
@@ -236,17 +245,6 @@ def prioritization(head_nodes: list[GraphNode],
             print('Trying to apply local optimizations to non-generic scheduler, ignoring it')
 
         match self._lag_optimize:
-            case LagOptimizationStrategy.NONE:
-                wg = self._wg
-                schedules = scheduler.schedule_with_cache(wg, self._contractors,
-                                                          self._spec,
-                                                          landscape=self._landscape_config,
-                                                          assigned_parent_time=self._assigned_parent_time,
-                                                          validate=validate)
-                node_orders = [node_order for _, _, _, node_order in schedules]
-                schedules = [schedule for schedule, _, _, _ in schedules]
-                self._node_orders = node_orders
-
             case LagOptimizationStrategy.AUTO:
                 # Searching the best
                 wg1 = graph_restructuring(self._wg, False)
@@ -278,7 +276,7 @@ def prioritization(head_nodes: list[GraphNode],
                     wg = wg2
                     schedules = schedules2
 
-            case _:
+            case LagOptimizationStrategy.TRUE, LagOptimizationStrategy.FALSE:
                 wg = graph_restructuring(self._wg, self._lag_optimize.value)
                 schedules = scheduler.schedule_with_cache(wg, self._contractors,
                                                           self._spec,
@@ -288,6 +286,16 @@ def prioritization(head_nodes: list[GraphNode],
                 node_orders = [node_order for _, _, _, node_order in schedules]
                 schedules = [schedule for schedule, _, _, _ in schedules]
                 self._node_orders = node_orders
+            case _:
+                wg = self._wg
+                schedules = scheduler.schedule_with_cache(wg, self._contractors,
+                                                          self._spec,
+                                                          landscape=self._landscape_config,
+                                                          assigned_parent_time=self._assigned_parent_time,
+                                                          validate=validate)
+                node_orders = [node_order for _, _, _, node_order in schedules]
+                schedules = [schedule for schedule, _, _, _ in schedules]
+                self._node_orders = node_orders
 
         return DefaultSchedulePipeline(self, wg, schedules)
 

sampo/pipeline/lag_optimization.py

@@ -5,4 +5,3 @@ class LagOptimizationStrategy(Enum):
     TRUE = True
     FALSE = False
     AUTO = None
-    NONE = None

sampo/scheduler/heft/prioritization.py

@@ -1,57 +1,12 @@
 from operator import itemgetter
 
+from sampo.scheduler.utils.critical_path import ford_bellman
 from sampo.utilities.priority import extract_priority_groups_from_nodes
 from sampo.scheduler.utils.time_computaion import work_priority, calculate_working_time_cascade
 from sampo.schemas.graph import GraphNode
 from sampo.schemas.time_estimator import WorkTimeEstimator
 
 
-def ford_bellman(nodes: list[GraphNode],
-                 weights: dict[str, float],
-                 node_id2parent_ids: dict[str, set[str]]) -> dict[str, float]:
-    """
-    Runs heuristic ford-bellman algorithm for given graph and weights.
-    """
-    path_weights: dict[str, float] = {node.id: 0 for node in nodes}
-    # cache graph edges
-    edges: list[tuple[str, str, float]] = sorted([(finish, start.id, weights[finish])
-                                                  for start in nodes
-                                                  for finish in node_id2parent_ids[start.id]
-                                                  if finish in path_weights])
-    if not edges:
-        return path_weights
-
-    # for changes heuristic
-    changed = False
-    # run standard ford-bellman on reversed edges
-    # optimize dict access to finish weight
-    for i in range(len(nodes)):
-        cur_finish = edges[0][0]
-        cur_finish_weight = path_weights[cur_finish]
-        # relax on edges
-        for finish, start, weight in edges:
-            # we are running through the equality class by finish node
-            # so if it changes renew the parameters of current equality class
-            if cur_finish != finish:
-                path_weights[cur_finish] = cur_finish_weight
-                cur_finish = finish
-                cur_finish_weight = path_weights[cur_finish]
-            new_weight = path_weights[start] + weight
-            if new_weight < cur_finish_weight:
-                cur_finish_weight = new_weight
-                changed = True
-        # if we were done completely nothing with actual minimum weights, the algorithm ends
-        if not changed:
-            break
-        # we go here if changed = True
-        # so the last equality class weight can be changed, save it
-        path_weights[cur_finish] = cur_finish_weight
-        # next iteration should start without change info from previous
-        changed = False
-
-    return path_weights
-
-
 def prioritization_nodes(nodes: list[GraphNode],
                          node_id2parent_ids: dict[str, set[str]],
                          work_estimator: WorkTimeEstimator) -> list[GraphNode]:

sampo/scheduler/utils/__init__.py

@@ -38,6 +38,35 @@ def get_head_nodes_with_connections_mappings(wg: WorkGraph
     Args:
         wg: The `WorkGraph` to analyze.
 
+    Returns:
+        A tuple containing:
+            - A list of `GraphNode` objects representing the head nodes,
+              sorted in topological order based on their reconstructed dependencies.
+            - A dictionary mapping the ID of each head node to a set of IDs of
+              its new 'parent' head nodes. These represent external dependencies
+              where a parent of any node within the current head node's inseparable
+              chain belongs to another head node's chain.
+            - A dictionary mapping the ID of each head node to a set of IDs of
+              its new 'child' head nodes. Similar to parents, these represent
+              external dependencies where a child of any node within the current
+              head node's inseparable chain belongs to another head node's chain.
+    """
+    return get_head_nodes_with_connections_mappings_nodes(wg.nodes)
+
+
+def get_head_nodes_with_connections_mappings_nodes(nodes: list[GraphNode]) -> tuple[list[GraphNode], dict[str, set[str]], dict[str, set[str]]]:
+    """
+    Identifies 'head nodes' in a WorkGraph and reconstructs their inter-node dependencies.
+
+    Head nodes are defined as the first nodes of inseparable chains or standalone nodes
+    that are not part of an inseparable chain (i.e., they are not 'inseparable sons').
+    This function effectively flattens the graph by treating inseparable chains as
+    single logical entities represented by their head node, and then re-establishes
+    parent-child relationships between these head nodes.
+
+    Args:
+        nodes: The `WorkGraph` to analyze.
+
     Returns:
         A tuple containing:
             - A list of `GraphNode` objects representing the head nodes,
@@ -54,7 +83,8 @@ def get_head_nodes_with_connections_mappings(wg: WorkGraph
     # Filter the work graph nodes to identify all 'head nodes'.
     # A head node is one that is not an 'inseparable son', meaning it's either
     # the start of an inseparable chain or a standalone node.
-    nodes = [node for node in wg.nodes if not node.is_inseparable_son()]
+    nodes = [node for node in nodes if not node.is_inseparable_son()]
+    node_dict = {node.id: node for node in nodes}
 
     # Construct a mapping from any node within an inseparable chain to its
     # corresponding head node.
@@ -87,6 +117,6 @@ def get_head_nodes_with_connections_mappings(wg: WorkGraph
     tsorted_nodes_ids = toposort_flatten(node_id2parent_ids, sort=True)
 
     # Map the sorted head node IDs back to their corresponding GraphNode objects.
-    tsorted_nodes = [wg[node_id] for node_id in tsorted_nodes_ids]
+    tsorted_nodes = [node_dict[node_id] for node_id in tsorted_nodes_ids]
 
     return tsorted_nodes, node_id2parent_ids, node_id2child_ids