Modifications to scheduled dates can now cascade through the task tree

2026-08-09 17:31:45 +00:00 · 2021-06-17 13:23:09 +10:00
parent 97d0286b57
commit 50608d9459
5 changed files with 192 additions and 17 deletions
@@ -0,0 +1,162 @@
+import datetime
+import ifcopenshell.util.date
+import ifcopenshell.util.sequence
+
+
+class Usecase:
+    def __init__(self, file, **settings):
+        self.file = file
+        self.settings = {"task": None}
+        for key, value in settings.items():
+            self.settings[key] = value
+
+    def execute(self):
+        self.calendar_cache = {}
+        self.cascade_task(self.settings["task"])
+
+    def cascade_task(self, task):
+        if not task.TaskTime:
+            return
+
+        duration = (
+            ifcopenshell.util.date.ifc2datetime(task.TaskTime.ScheduleDuration)
+            if task.TaskTime.ScheduleDuration
+            else datetime.timedelta()
+        )
+
+        finishes = []
+        starts = []
+
+        for rel in task.IsSuccessorFrom:
+            predecessor = rel.RelatingProcess
+            if rel.SequenceType == "FINISH_START":
+                finish = self.get_task_time_attribute(predecessor, "ScheduleFinish")
+                if not finish:
+                    continue
+                if rel.TimeLag:
+                    days = self.get_lag_time_days(rel.TimeLag)
+                    duration_type = rel.TimeLag.DurationType
+                    starts.append(self.offset_date(finish, days, duration_type, self.get_calendar(task)))
+                    starts.append(self.offset_date(finish, days, duration_type, self.get_calendar(predecessor)))
+                else:
+                    starts.append(finish)
+            elif rel.SequenceType == "START_START":
+                start = self.get_task_time_attribute(predecessor, "ScheduleStart")
+                if not start:
+                    continue
+                if rel.TimeLag:
+                    days = self.get_lag_time_days(rel.TimeLag)
+                    duration_type = rel.TimeLag.DurationType
+                    starts.append(self.offset_date(start, days, duration_type, self.get_calendar(task)))
+                    starts.append(self.offset_date(start, days, duration_type, self.get_calendar(predecessor)))
+                else:
+                    starts.append(start)
+            elif rel.SequenceType == "FINISH_FINISH":
+                finish = self.get_task_time_attribute(predecessor, "ScheduleFinish")
+                if not finish:
+                    continue
+                if rel.TimeLag:
+                    days = self.get_lag_time_days(rel.TimeLag)
+                    duration_type = rel.TimeLag.DurationType
+                    finishes.append(self.offset_date(finish, days, duration_type, self.get_calendar(task)))
+                    finishes.append(self.offset_date(finish, days, duration_type, self.get_calendar(predecessor)))
+                else:
+                    finishes.append(finish)
+            elif rel.SequenceType == "START_FINISH":
+                start = self.get_task_time_attribute(predecessor, "ScheduleStart")
+                if not start:
+                    continue
+                if rel.TimeLag:
+                    days = self.get_lag_time_days(rel.TimeLag)
+                    duration_type = rel.TimeLag.DurationType
+                    finishes.append(self.offset_date(start, days, duration_type, self.get_calendar(task)))
+                    finishes.append(self.offset_date(start, days, duration_type, self.get_calendar(predecessor)))
+                else:
+                    finishes.append(start)
+            else:
+                return
+
+        if starts and finishes:
+            start = max(starts)
+            finish = max(finishes)
+            potential_finish = ifcopenshell.util.sequence.get_finish_date(
+                start,
+                duration,
+                task.TaskTime.DurationType,
+                self.get_calendar(task),
+            )
+            if potential_finish > finish:
+                start_ifc = ifcopenshell.util.date.datetime2ifc(start, "IfcDateTime")
+                if task.TaskTime.ScheduleStart == start_ifc:
+                    return
+                task.TaskTime.ScheduleStart = start_ifc
+                task.TaskTime.ScheduleFinish = ifcopenshell.util.date.datetime2ifc(potential_finish, "IfcDateTime")
+            else:
+                finish_ifc = ifcopenshell.util.date.datetime2ifc(finish, "IfcDateTime")
+                if task.TaskTime.ScheduleFinish == finish_ifc:
+                    return
+                task.TaskTime.ScheduleFinish = finish_ifc
+                task.TaskTime.ScheduleStart = ifcopenshell.util.date.datetime2ifc(
+                    ifcopenshell.util.sequence.get_finish_date(
+                        finish,
+                        -duration,
+                        task.TaskTime.DurationType,
+                        self.get_calendar(task),
+                    ),
+                    "IfcDateTime",
+                )
+        elif finishes:
+            finish = max(finishes)
+            finish_ifc = ifcopenshell.util.date.datetime2ifc(finish, "IfcDateTime")
+            if task.TaskTime.ScheduleFinish == finish_ifc:
+                return
+            task.TaskTime.ScheduleFinish = finish_ifc
+            task.TaskTime.ScheduleStart = ifcopenshell.util.date.datetime2ifc(
+                ifcopenshell.util.sequence.get_finish_date(
+                    finish,
+                    -duration,
+                    task.TaskTime.DurationType,
+                    self.get_calendar(task),
+                ),
+                "IfcDateTime",
+            )
+        elif starts:
+            start = max(starts)
+            start_ifc = ifcopenshell.util.date.datetime2ifc(start, "IfcDateTime")
+            if task.TaskTime.ScheduleStart == start_ifc:
+                return
+            task.TaskTime.ScheduleStart = start_ifc
+            task.TaskTime.ScheduleFinish = ifcopenshell.util.date.datetime2ifc(
+                ifcopenshell.util.sequence.get_finish_date(
+                    start,
+                    duration,
+                    task.TaskTime.DurationType,
+                    self.get_calendar(task),
+                ),
+                "IfcDateTime",
+            )
+        else:
+            return
+
+        for rel in task.IsPredecessorTo:
+            self.cascade_task(rel.RelatedProcess)
+
+    def get_lag_time_days(self, lag_time):
+        return ifcopenshell.util.date.ifc2datetime(lag_time.LagValue.wrappedValue).days
+
+    def get_calendar(self, task):
+        if task.id() not in self.calendar_cache:
+            self.calendar_cache[task.id()] = ifcopenshell.util.sequence.derive_calendar(task)
+        return self.calendar_cache[task.id()]
+
+    def offset_date(self, date, days, duration_type, calendar):
+        return datetime.datetime.combine(
+            ifcopenshell.util.sequence.get_finish_date(date, datetime.timedelta(days=days), duration_type, calendar),
+            datetime.datetime.min.time(),
+        )
+
+    def get_task_time_attribute(self, task, attribute):
+        if task.TaskTime:
+            value = getattr(task.TaskTime, attribute)
+            if value:
+                return ifcopenshell.util.date.ifc2datetime(value)