Generic Task Moduel:

- StageActions: This method stages the action_sequence for a proper execution order (in the task initialization, it is called if an action sequence is not provided). Currently, it takes an argument 'from_deps' to compute the sequence based on the inter-action dependencies but the idea is to enable other strategies later to be built to create the action_sequence in the Task for more autonomy.
- calcDuration - This organizes the actions to be done by this task into the proper order based on the action_sequence. This is used to fill out the self.actions_ti (the initial time of individual actions within the task), self.ti, and self.tf. Right now, when we're initializing the task, we assume a ti of zero.
- calcCost - This calculate the total cost of the task based on the costs of individual actions (a simple roll-up). This updates self.cost.
- Both calcDuration and calcCost assumes that the individual action.duration and action.cost have already been computed (using calcCostandDuration in the action class but I'm also thinking of dividing this method into two different methods - one for cost and one for duration to match with Task.py and also task-assets matrix in the scheduler.
- updateTaskTime: This method updates the starting and finishing time of the whole task and the starting time of individual actions based on a new start time passed to Task (could be helpful later during task scheduling using the sceduler.

Author: Yuksel-Rudy

famodel/irma/calwave_task1.py

@@ -7,7 +7,8 @@
 from famodel.project import Project
 from calwave_irma import Scenario
 import calwave_chart as chart
-from calwave_task import Task
+# from calwave_task import Task  # calwave_task module (Felipe)
+from task import Task as Task    # generic Task module ( Rudy )
 
 sc = Scenario()  # now sc exists in *this* session
 
@@ -183,21 +184,22 @@ def assign_actions(sc: Scenario, actions: dict):
     # 4) Assign (assign vessels/roles)
     assign_actions(sc, actions)
 
-    # 5) schedule once, in the Task
-    calwave_task1 = Task.from_scenario(
-        sc,
-        name='calwave_task1',
-        strategy='earliest',               # 'earliest' or 'levels'
-        enforce_resources=False,         # keep single-resource blocking if you want it
-        resource_roles=('vessel', 'carrier', 'operator'))
-    
-    # 6) Extract Task1 sequencing info
-    # calwave_task1.extractSeqYaml()
+    # # 5) schedule once, in the Task
+    # calwave_task1 = Task.from_scenario(
+    #     sc,
+    #     name='calwave_task1',
+    #     strategy='earliest',               # 'earliest' or 'levels'
+    #     enforce_resources=False,         # keep single-resource blocking if you want it
+    #     resource_roles=('vessel', 'carrier', 'operator'))
+
+
+    # 5) Build Task
+    task1 = Task(name='calwave_task1', actions=sc.actions)
+
+    # task1.updateTaskTime(newStart=10)
 
-    # 7) update Task1 if needed
-    calwave_task1.update_from_SeqYaml()  # uncomment to re-apply sequencing from YAML
-    # 8) build the chart input directly from the Task and plot
-    chart_view = chart.view_from_task(calwave_task1, sc, title='CalWave Task 1 - Anchor installation plan')
+    # 6) build the chart input directly from the Task and plot  #TODO: Rudy / Improve this later (maybe include it in Task.py/Scenario and let it plot the absolute time instead of relative time)
+    chart_view = chart.view_from_task(task1, sc, title='CalWave Task 1 - Anchor installation plan')
     chart.plot_task(chart_view, outpath='calwave_task1_chart.png')
 
 

famodel/irma/task.py

@@ -32,250 +32,140 @@ class Task():
     
     '''
 
-    def __init__(self, actions, action_sequence, name, **kwargs):
+    def __init__(self, name, actions, action_sequence=None, **kwargs):
         '''Create an action object...
         It must be given a name and a list of actions.
         The action list should be by default coherent with actionTypes dictionary.
         
         Parameters
         ----------
+        name : string
+            A name for the action. It may be appended with numbers if there
+            are duplicate names.
         actions : list
             A list of all actions that are part of this task.
-        action_sequence : dict
+        action_sequence : dict, optional
             A dictionary where each key is the name of each action, and the values are
             each a list of which actions (by name) must be completed before the current
-            one.
-        name : string
-            A name for the action. It may be appended with numbers if there
-            are duplicate names.
+            one. If None, the action_sequence will be built by calling self.stageActions(from_deps=True)
+            [building from the dependencies of each action].
         kwargs 
             Additional arguments may depend on the task type.
         
         '''
-        
-        # Make a dict by name of all actions that are carried out in this task
-        self.actions = {}
-        for act in actions:
-            self.actions[act.name] = act
+    
 
 
 
         self.name = name
-        
+        self.actions = {a.name: a for a in actions.values()}
+
+        if action_sequence is None:
+            self.stageActions(from_deps=True)
+        else:
+            self.action_sequence = {k: list(v) for k, v in action_sequence.items()}
+
+
         self.status = 0  # 0, waiting;  1=running;  2=finished
         self.actions_ti = {}  # relative start time of each action [h]
 
-        self.duration = 0  # duration must be calculated based on lengths of actions
-        self.cost     = 0  # cost must be calculated based on the cost of individual actions.
-        self.ti =0  # task start time [h?]
-        self.tf =0  # task end time [h?]
-        
-        # what else do we need to initialize the task?
-        
-        # Create a graph of the sequence of actions in this task based on action_sequence
-        self.getSequenceGraph(action_sequence, plot=True) # this also updates duration
-        
-        self.cost = sum(action.cost for action in self.actions.values())
+        self.duration = 0.0  # duration must be calculated based on lengths of actions
+        self.cost     = 0.0  # cost must be calculated based on the cost of individual actions.
+        self.ti       = 0.0  # task start time [h?]
+        self.tf       = 0.0  # task end time [h?]
 
-        print(f"---------------------- Initializing Task '{self.name} ----------------------")
+        # Calculate duration and cost
+        self.calcDuration()  # organizes actions and calculates duration
+        self.calcCost()
+
+        print(f"---------------------- Initializing Task '{self.name} ----------------------")        
         print(f"Task '{self.name}' initialized with duration = {self.duration:.2f} h.")
         print(f"Task '{self.name}' initialized with cost     = ${self.cost:.2f} ")
 
-    def organizeActions(self):
-        '''Organizes the actions to be done by this task into the proper order
-        based on the strategy of this type of task...
+    def stageActions(self, from_deps=True):
         '''
-        
-        if self.type == 'parallel_anchor_install':
-            
-            pass
-            # make a graph that reflects this strategy?
-    
-    
-    def calcDuration(self):
-        '''Calculates the duration of the task based on the durations of the
-        individual actions and their order of operation.'''
-        
-        # Does Rudy have graph-based code that can do this?
-        
-        
-        
-        
-        
-        
-    def getSequenceGraph(self, action_sequence, plot=True):
-        '''Generate a multi-directed graph that visalizes action sequencing within the task.
-                Build a MultiDiGraph with nodes:
-                Start -> CP1 -> CP2 -> ... -> End
-        
-        Checkpoints are computed from action "levels":
-          level(a) = 1 if no prerequisites.
-          level(a) = 1 + max(level(p) for p in prerequisites)      1 + the largest level among a’s prerequisites.
-          Number of checkpoints = max(level) - 1.           
-        '''
-
-        # Compute levels
-        levels: dict[str, int] = {}
-        def level_of(a: str, b: set[str]) -> int:
-            '''Return the level of action a. b is the set of actions currently being explored'''
+        This method stages the action_sequence for a proper execution order.
 
-            # If we have already computed the level, return it
-            if a in levels:
-                return levels[a]
+        Parameters
+        ----------
+        from_deps : bool
+            If True, builds the action_sequence from the dependencies of each action.
+            More options will be added in the future.
+        '''
+        if from_deps:
+            # build from dependencies
+            def getDeps(action):
+                deps = []
+                for dep in action.dependencies:
+                    deps.append(dep)
+                return deps
 
-            # The action cannot be its own prerequisite 
-            if a in b:
-                raise ValueError(f"Cycle detected in action sequence at '{a}' in task '{self.name}'. The action cannot be its own prerequisite.")
-
-            b.add(a)
-
-            # Look up prerequisites for action a.
-            pres = action_sequence.get(a, [])
-            if not pres:
-                lv = 1  # No prerequisites, level 1
-            else:
-                # If a prerequisites name is not in the dict, treat it as a root (level 1)
-                lv = 1 + max(level_of(p, b) if p in action_sequence else 1 for p in pres)
+            self.action_sequence = {self.actions[name].name: getDeps(self.actions[name]) for name in self.actions}
 
-            # b.remove(a)  # if you want to unmark a from the explored dictionary, b, uncomment this line. 
-            levels[a] = lv
-            return lv
-
-        for a in action_sequence:
-            level_of(a, set())
-        
-        max_level = max(levels.values(), default=1)
-        num_cps = max(0, max_level - 1)
-
-        H = nx.MultiDiGraph()
-
-        # Add the Start -> [checkpoints] -> End nodes
-        H.add_node("Start")
-        for i in range(1, num_cps + 1):
-            H.add_node(f"CP{i}")
-        H.add_node("End")
-
-        shells = [["Start"]]
-        if num_cps > 0:
-            # Middle shells
-            cps = [f"CP{i}" for i in range(1, num_cps + 1)]
-            shells.append(cps)
-        shells.append(["End"])
-
-        pos = nx.shell_layout(H, nlist=shells)
-
-        xmin, xmax = -2.0, 2.0  # maybe would need to change those later on.
-        pos["Start"] = (xmin, 0)
-        pos["End"]   = (xmax, 0)
-
-        # Add action edges
-        # Convention:
-        #   level 1 actions: Start -> CP1 (or Start -> End if no CPs)
-        #   level L actions (2 <= L < max_level): CP{L-1} -> CP{L}
-        #   level == max_level actions: CP{num_cps} -> End
-        for action, lv in levels.items():
-            action = self.actions[action]
-            if num_cps == 0:
-                # No checkpoints: all actions from Start to End
-                H.add_edge("Start", "End", key=action, duration=action.duration, cost=action.cost)
-            else:
-                if lv == 1:
-                    H.add_edge("Start", "CP1", key=action, duration=action.duration, cost=action.cost)
-                elif lv < max_level:
-                    H.add_edge(f"CP{lv-1}", f"CP{lv}", key=action, duration=action.duration, cost=action.cost)
-                else:  # lv == max_level
-                    H.add_edge(f"CP{num_cps}", "End", key=action, duration=action.duration, cost=action.cost)
 
+    def calcDuration(self):
+        '''Organizes the actions to be done by this task into the proper order
+        based on the action_sequence. This is used to fill out 
+        self.actions_ti, self.ti, and self.tf. This method assumes that action.duration
+        have already been evaluated for each action in self.actions.
+        '''
+        # Initialize dictionaries to hold start and finish times
+        starts = {}
+        finishes = {}
 
-        # 3. Compute cumulative start time for each level
-        level_groups = {}
-        for action, lv in levels.items():
-            level_groups.setdefault(lv, []).append(action)
-        
-        level_durations = {lv: max(self.actions[a].duration for a in acts)
-                        for lv, acts in level_groups.items()}        
+        # Iterate through actions in the sequence
+        for action, dep_actions in self.action_sequence.items():
+            # Calculate start time as the max finish time of dependencies
+            starts[action] = max((finishes[dep] for dep in dep_actions), default=0)
 
-        task_duration = sum(level_durations.values())
+            # get duration from actions
+            duration = self.actions[action].duration  # in hours
 
-        level_start_time = {}
-        elapsed = 0.0
-        cp_string = []
-        for lv in range(1, max_level + 1):
-            level_start_time[lv] = elapsed
-            elapsed += level_durations.get(lv, 0.0)
-            # also collect all actions at this level for title
-            acts = [a for a, l in levels.items() if l == lv]
-            if acts and lv <= num_cps:
-                cp_string.append(f"CP{lv}: {', '.join(acts)}")
-            elif acts and lv > num_cps:
-                cp_string.append(f"End: {', '.join(acts)}")
+            # Calculate finish time
+            finishes[action] = starts[action] + duration
 
-        # Assign to self:
-        self.duration = task_duration
-        self.actions_ti = {a: level_start_time[lv] for a, lv in levels.items()}
-        self.sequence_graph = H
-        
-        title_str = f"Task {self.name}. Duration {self.duration:.2f} : " + " | ".join(cp_string)
+        # Update self.actions_ti with relative start times
+        self.actions_ti = starts
 
-        if plot:
-            fig, ax = plt.subplots()
-            # pos = nx.shell_layout(G)
-            nx.draw(H, pos, with_labels=True, node_size=500, node_color="lightblue", edge_color='white')
+        # Set task start time and finish time
+        self.ti = min(starts.values(), default=0)
+        self.tf = max(finishes.values(), default=0)
 
-            label_positions = {}  # to store label positions for each edge
-            # Group edges by unique (u, v) pairs
-            for (u, v) in set((u, v) for u, v, _ in H.edges(keys=True)):
-                # get all edges between u and v (dict keyed by edge key)
-                edge_dict = H.get_edge_data(u, v)  # {key: {attrdict}, ...}
-                n = len(edge_dict)
+        # Task duration
+        self.duration = self.tf - self.ti
 
-                # curvature values spread between -0.3 and +0.3 [helpful to visualize multiple edges]
-                if n==1:
-                    rads = [0]
-                    offsets = [0.5]
-                else:
-                    rads = np.linspace(-0.3, 0.3, n)
-                    offsets = np.linspace(0.2, 0.8, n)
-                
-                # draw each edge
-                durations = [d.get("duration", 0.0) for d in edge_dict.values()]
-                scale = max(max(durations), 0.0001)  # avoid div by zero
-                width_scale = 4.0 / scale  # normalize largest to ~4px
+    def calcCost(self):
+        '''Calculates the total cost of the task based on the costs of individual actions.
+        Updates self.cost accordingly. This method assumes that action.cost has 
+        already been evaluated for each action in self.actions.
+        '''
+        total_cost = 0.0
+        for action in self.actions.values():
+            total_cost += action.cost
+        self.cost = total_cost
+        return self.cost
 
-                for rad, offset, (k, d) in zip(rads, offsets, edge_dict.items()):
-                    nx.draw_networkx_edges(
-                        H, pos, edgelist=[(u, v)], ax=ax,
-                        connectionstyle=f"arc3,rad={rad}",
-                        arrows=True, arrowstyle="-|>",
-                        edge_color="gray",
-                        width=max(0.5, d.get("duration", []) * width_scale),
-                    )
-                    label_positions[(u, v, k)] = offset  # store position for edge label
+    def updateTaskTime(self, newStart=0.0):
+        '''Update the start time of all actions based on a new task start time.
 
-            ax.set_title(title_str, fontsize=12, fontweight="bold")
-            ax.axis("off")
-            plt.tight_layout()
+        Parameters
+        ----------
+        newStart : float
+            The new start time for the task. All action start times will be adjusted accordingly.
+        '''
+        # Calculate the time shift
+        time_shift = newStart - self.ti
 
-        return H
-                
-        
+        # Update task start and finish times
+        self.ti = newStart
+        self.tf += time_shift
 
-    def getTaskGraph(self, plot=True):
-        '''Generate a graph of the action dependencies.
-        '''
-        
-        # Create the graph
-        G = nx.DiGraph()
-        for item, data in self.actions.items():
-            for dep in data.dependencies:
-                G.add_edge(dep, item, duration=data.duration)  # Store duration as edge attribute
+        # Update action start times
+        for action in self.actions_ti:
+            self.actions_ti[action] += time_shift
+    
 
-        # Compute longest path & total duration
-        longest_path = nx.dag_longest_path(G, weight='duration')
-        longest_path_edges = list(zip(longest_path, longest_path[1:]))  # Convert path into edge pairs
 
-        total_duration = sum(self.actions[node].duration for node in longest_path) 
-        return G  
 
 
     def get_row(self, assets):