Glassvision/IfcOpenShell
1
0
Fork 0
mirror of https://github.com/IfcOpenShell/IfcOpenShell.git synced 2026-08-10 01:41:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Quality - run black on api/sequence directory

Browse Source
This commit is contained in:
bosonprojets
2022-09-09 02:25:20 +01:00
parent 4689cc858a
commit 87f0475b99
25 changed files with 477 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ifcopenshell-python/ifcopenshell/api/sequence/add_task.py
+12 -3
View File
@@ -44,15 +44,24 @@ class Usecase:
"IfcRelAssignsToControl",
**{
"GlobalId": ifcopenshell.guid.new(),
"OwnerHistory": ifcopenshell.api.run("owner.create_owner_history", self.file),
"OwnerHistory": ifcopenshell.api.run(
"owner.create_owner_history", self.file
),
"RelatedObjects": [task],
"RelatingControl": self.settings["work_schedule"],
}
)
elif self.settings["parent_task"]:
rel = ifcopenshell.api.run(
"nest.assign_object", self.file, related_object=task, relating_object=self.settings["parent_task"]
"nest.assign_object",
self.file,
related_object=task,
relating_object=self.settings["parent_task"],
)
if self.settings["parent_task"].Identification:
task.Identification = self.settings["parent_task"].Identification + "." + str(len(rel.RelatedObjects))
task.Identification = (
self.settings["parent_task"].Identification
+ "."
+ str(len(rel.RelatedObjects))
)
return task
src/ifcopenshell-python/ifcopenshell/api/sequence/add_time_period.py
+6 -2
View File
@@ -35,8 +35,12 @@ class Usecase:
def execute(self):
time_period = self.file.create_entity("IfcTimePeriod")
time_period.StartTime = ifcopenshell.util.date.datetime2ifc(self.settings["start_time"], "IfcTime")
time_period.EndTime = ifcopenshell.util.date.datetime2ifc(self.settings["end_time"], "IfcTime")
time_period.StartTime = ifcopenshell.util.date.datetime2ifc(
self.settings["start_time"], "IfcTime"
)
time_period.EndTime = ifcopenshell.util.date.datetime2ifc(
self.settings["end_time"], "IfcTime"
)
time_periods = list(self.settings["recurrence_pattern"].TimePeriods or [])
time_periods.append(time_period)
self.settings["recurrence_pattern"].TimePeriods = time_periods
src/ifcopenshell-python/ifcopenshell/api/sequence/add_work_calendar.py
+4 -1
View File
@@ -36,6 +36,9 @@ class Usecase:
)
context = self.file.by_type("IfcContext")[0]
ifcopenshell.api.run(
"project.assign_declaration", self.file, definition=work_calendar, relating_context=context
"project.assign_declaration",
self.file,
definition=work_calendar,
relating_context=context,
)
return work_calendar
src/ifcopenshell-python/ifcopenshell/api/sequence/add_work_plan.py
+17 -4
View File
@@ -24,7 +24,11 @@ from datetime import datetime
class Usecase:
def __init__(self, file, **settings):
self.file = file
self.settings = {"name": None, "predefined_type": "NOTDEFINED", "start_time": datetime.now()}
self.settings = {
"name": None,
"predefined_type": "NOTDEFINED",
"start_time": datetime.now(),
}
for key, value in settings.items():
self.settings[key] = value
@@ -36,12 +40,21 @@ class Usecase:
predefined_type=self.settings["predefined_type"],
name=self.settings["name"],
)
work_plan.CreationDate = ifcopenshell.util.date.datetime2ifc(datetime.now(), "IfcDateTime")
work_plan.CreationDate = ifcopenshell.util.date.datetime2ifc(
datetime.now(), "IfcDateTime"
)
user = ifcopenshell.api.owner.settings.get_user(self.file)
if user:
work_plan.Creators = [user.ThePerson]
work_plan.StartTime = ifcopenshell.util.date.datetime2ifc(self.settings["start_time"], "IfcDateTime")
work_plan.StartTime = ifcopenshell.util.date.datetime2ifc(
self.settings["start_time"], "IfcDateTime"
)
context = self.file.by_type("IfcContext")[0]
ifcopenshell.api.run("project.assign_declaration", self.file, definition=work_plan, relating_context=context)
ifcopenshell.api.run(
"project.assign_declaration",
self.file,
definition=work_plan,
relating_context=context,
)
return work_plan
src/ifcopenshell-python/ifcopenshell/api/sequence/add_work_schedule.py
+14 -4
View File
@@ -41,23 +41,33 @@ class Usecase:
predefined_type=self.settings["predefined_type"],
name=self.settings["name"],
)
work_schedule.CreationDate = ifcopenshell.util.date.datetime2ifc(datetime.now(), "IfcDateTime")
work_schedule.CreationDate = ifcopenshell.util.date.datetime2ifc(
datetime.now(), "IfcDateTime"
)
user = ifcopenshell.api.owner.settings.get_user(self.file)
if user:
work_schedule.Creators = [user.ThePerson]
work_schedule.StartTime = ifcopenshell.util.date.datetime2ifc(self.settings["start_time"], "IfcDateTime")
work_schedule.StartTime = ifcopenshell.util.date.datetime2ifc(
self.settings["start_time"], "IfcDateTime"
)
if self.settings["work_plan"]:
ifcopenshell.api.run(
"aggregate.assign_object",
self.file,
**{"product": work_schedule, "relating_object": self.settings["work_plan"]}
**{
"product": work_schedule,
"relating_object": self.settings["work_plan"],
}
)
else:
# TODO: this is an ambiguity by buildingSMART
# See https://forums.buildingsmart.org/t/is-the-ifcworkschedule-project-declaration-mutually-exclusive-to-aggregation-within-a-relating-ifcworkplan/3510
context = self.file.by_type("IfcContext")[0]
ifcopenshell.api.run(
"project.assign_declaration", self.file, definition=work_schedule, relating_context=context
"project.assign_declaration",
self.file,
definition=work_schedule,
relating_context=context,
)
return work_schedule
src/ifcopenshell-python/ifcopenshell/api/sequence/assign_lag_time.py
+10 -3
View File
@@ -22,13 +22,19 @@ import ifcopenshell.util.date
class Usecase:
def __init__(self, file, **settings):
self.file = file
self.settings = {"rel_sequence": None, "lag_value": None, "duration_type": "NOTDEFINED"}
self.settings = {
"rel_sequence": None,
"lag_value": None,
"duration_type": "NOTDEFINED",
}
for key, value in settings.items():
self.settings[key] = value
def execute(self):
lag_value = self.file.createIfcDuration(
ifcopenshell.util.date.datetime2ifc(self.settings["lag_value"], "IfcDuration")
ifcopenshell.util.date.datetime2ifc(
self.settings["lag_value"], "IfcDuration"
)
)
lag_time = self.file.create_entity(
"IfcLagTime",
@@ -40,7 +46,8 @@ class Usecase:
if self.settings["rel_sequence"].is_a("IfcRelSequence"):
if (
self.settings["rel_sequence"].TimeLag
and len(self.file.get_inverse(self.settings["rel_sequence"].TimeLag)) == 1
and len(self.file.get_inverse(self.settings["rel_sequence"].TimeLag))
== 1
):
self.file.remove(self.settings["rel_sequence"].TimeLag)
self.settings["rel_sequence"].TimeLag = lag_time
src/ifcopenshell-python/ifcopenshell/api/sequence/assign_process.py
+6 -2
View File
@@ -48,13 +48,17 @@ class Usecase:
related_objects = list(operates_on.RelatedObjects)
related_objects.append(self.settings["related_object"])
operates_on.RelatedObjects = related_objects
ifcopenshell.api.run("owner.update_owner_history", self.file, **{"element": operates_on})
ifcopenshell.api.run(
"owner.update_owner_history", self.file, **{"element": operates_on}
)
else:
operates_on = self.file.create_entity(
"IfcRelAssignsToProcess",
**{
"GlobalId": ifcopenshell.guid.new(),
"OwnerHistory": ifcopenshell.api.run("owner.create_owner_history", self.file),
"OwnerHistory": ifcopenshell.api.run(
"owner.create_owner_history", self.file
),
"RelatedObjects": [self.settings["related_object"]],
"RelatingProcess": self.settings["relating_process"],
}
src/ifcopenshell-python/ifcopenshell/api/sequence/assign_product.py
+6 -2
View File
@@ -47,13 +47,17 @@ class Usecase:
related_objects = list(referenced_by.RelatedObjects)
related_objects.append(self.settings["related_object"])
referenced_by.RelatedObjects = related_objects
ifcopenshell.api.run("owner.update_owner_history", self.file, **{"element": referenced_by})
ifcopenshell.api.run(
"owner.update_owner_history", self.file, **{"element": referenced_by}
)
else:
referenced_by = self.file.create_entity(
"IfcRelAssignsToProduct",
**{
"GlobalId": ifcopenshell.guid.new(),
"OwnerHistory": ifcopenshell.api.run("owner.create_owner_history", self.file),
"OwnerHistory": ifcopenshell.api.run(
"owner.create_owner_history", self.file
),
"RelatedObjects": [self.settings["related_object"]],
"RelatingProduct": self.settings["relating_product"],
}
src/ifcopenshell-python/ifcopenshell/api/sequence/assign_recurrence_pattern.py
+7 -2
View File
@@ -25,12 +25,17 @@ class Usecase:
self.settings[key] = value
def execute(self):
recurrence = self.file.createIfcRecurrencePattern(self.settings["recurrence_type"])
recurrence = self.file.createIfcRecurrencePattern(
self.settings["recurrence_type"]
)
if self.settings["parent"].is_a("IfcWorkTime"):
if (
self.settings["parent"].RecurrencePattern
and len(self.file.get_inverse(self.settings["parent"].RecurrencePattern)) == 1
and len(
self.file.get_inverse(self.settings["parent"].RecurrencePattern)
)
== 1
):
self.file.remove(self.settings["parent"].RecurrencePattern)
self.settings["parent"].RecurrencePattern = recurrence
src/ifcopenshell-python/ifcopenshell/api/sequence/assign_sequence.py
+3 -1
View File
@@ -38,7 +38,9 @@ class Usecase:
"IfcRelSequence",
**{
"GlobalId": ifcopenshell.guid.new(),
"OwnerHistory": ifcopenshell.api.run("owner.create_owner_history", self.file),
"OwnerHistory": ifcopenshell.api.run(
"owner.create_owner_history", self.file
),
"RelatingProcess": self.settings["relating_process"],
"RelatedProcess": self.settings["related_process"],
"SequenceType": "FINISH_START",
src/ifcopenshell-python/ifcopenshell/api/sequence/assign_workplan.py
+4 -1
View File
@@ -39,6 +39,9 @@ class Usecase:
rel_aggregates = ifcopenshell.api.run(
"aggregate.assign_object",
self.file,
**{"product": self.settings["work_schedule"], "relating_object": self.settings["work_plan"]}
**{
"product": self.settings["work_schedule"],
"relating_object": self.settings["work_plan"],
}
)
return rel_aggregates
src/ifcopenshell-python/ifcopenshell/api/sequence/calculate_task_duration.py
+45 -33
View File
@@ -35,6 +35,32 @@ class Usecase:
if duration:
self.set_task_duration(duration)
def calculate_seconds_per_workday(self):
def get_work_schedule(task):
for rel in task.HasAssignments or []:
if rel.is_a("IfcRelAssignsToControl") and rel.RelatingControl.is_a(
"IfcWorkSchedule"
):
return rel.RelatingControl
for rel in task.Nests or []:
return get_work_schedule(rel.RelatingObject)
default_seconds_per_workday = 8 * 60 * 60
work_schedule = get_work_schedule(self.settings["task"])
if not work_schedule:
return default_seconds_per_workday
psets = ifcopenshell.util.element.get_psets(work_schedule)
if (
not psets
or "Pset_WorkControlCommon" not in psets
or "WorkDayDuration" not in psets["Pset_WorkControlCommon"]
):
return default_seconds_per_workday
work_day_duration = ifcopenshell.util.date.ifc2datetime(
psets["Pset_WorkControlCommon"]["WorkDayDuration"]
)
return work_day_duration.seconds
def calculate_max_resource_usage_duration(self):
max_duration = 0
for rel in self.settings["task"].OperatesOn or []:
@@ -45,44 +71,30 @@ class Usecase:
max_duration = duration
return max_duration
def set_task_duration(self, duration):
if not self.settings["task"].TaskTime:
ifcopenshell.api.run("sequence.add_task_time", self.file, task=self.settings["task"])
self.settings["task"].TaskTime.ScheduleDuration = f"P{duration}D"
def calculate_seconds_per_workday(self):
default_seconds_per_workday = 8 * 60 * 60
work_schedule = self.get_work_schedule(self.settings["task"])
if not work_schedule:
return default_seconds_per_workday
psets = ifcopenshell.util.element.get_psets(work_schedule)
if (
not psets
or "Pset_WorkControlCommon" not in psets
or "WorkDayDuration" not in psets["Pset_WorkControlCommon"]
):
return default_seconds_per_workday
work_day_duration = ifcopenshell.util.date.ifc2datetime(psets["Pset_WorkControlCommon"]["WorkDayDuration"])
return work_day_duration.seconds
def get_work_schedule(self, task):
for rel in task.HasAssignments or []:
if rel.is_a("IfcRelAssignsToControl") and rel.RelatingControl.is_a("IfcWorkSchedule"):
return rel.RelatingControl
for rel in task.Nests or []:
return self.get_work_schedule(rel.RelatingObject)
def calculate_duration_in_days(self, resource):
def is_hourly_work(schedule_work):
return "T" in schedule_work
if not resource.Usage or not resource.Usage.ScheduleWork:
return
schedule_usage = resource.Usage.ScheduleUsage or 1
schedule_duration = ifcopenshell.util.date.ifc2datetime(resource.Usage.ScheduleWork)
if self.is_hourly_work(resource.Usage.ScheduleWork):
schedule_seconds = (schedule_duration.days * 24 * 60 * 60) + schedule_duration.seconds
schedule_duration = ifcopenshell.util.date.ifc2datetime(
resource.Usage.ScheduleWork
)
if is_hourly_work(resource.Usage.ScheduleWork):
schedule_seconds = (
schedule_duration.days * 24 * 60 * 60
) + schedule_duration.seconds
else:
partial_days = schedule_duration.seconds / (24 * 60 * 60)
schedule_seconds = (schedule_duration.days + partial_days) * self.seconds_per_workday
schedule_seconds = (
schedule_duration.days + partial_days
) * self.seconds_per_workday
return math.ceil((schedule_seconds / self.seconds_per_workday) / schedule_usage)
def is_hourly_work(self, schedule_work):
return "T" in schedule_work
def set_task_duration(self, duration):
if not self.settings["task"].TaskTime:
ifcopenshell.api.run(
"sequence.add_task_time", self.file, task=self.settings["task"]
)
self.settings["task"].TaskTime.ScheduleDuration = f"P{duration}D"
src/ifcopenshell-python/ifcopenshell/api/sequence/cascade_schedule.py
+72 -16
View File
@@ -61,7 +61,9 @@ class Usecase:
for rel in task.IsSuccessorFrom:
predecessor = rel.RelatingProcess
predecessor_duration = (
ifcopenshell.util.date.ifc2datetime(predecessor.TaskTime.ScheduleDuration)
ifcopenshell.util.date.ifc2datetime(
predecessor.TaskTime.ScheduleDuration
)
if predecessor.TaskTime and predecessor.TaskTime.ScheduleDuration
else datetime.timedelta()
)
@@ -77,12 +79,20 @@ class Usecase:
if days:
starts.append(
datetime.datetime.combine(
self.offset_date(finish, days, duration_type, self.get_calendar(task)), datetime.time(9)
self.offset_date(
finish, days, duration_type, self.get_calendar(task)
),
datetime.time(9),
)
)
starts.append(
datetime.datetime.combine(
self.offset_date(finish, days, duration_type, self.get_calendar(predecessor)),
self.offset_date(
finish,
days,
duration_type,
self.get_calendar(predecessor),
),
datetime.time(9),
)
)
@@ -95,8 +105,16 @@ class Usecase:
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)))
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":
@@ -106,8 +124,16 @@ class Usecase:
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)))
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":
@@ -122,12 +148,20 @@ class Usecase:
if days or rel.TimeLag:
finishes.append(
datetime.datetime.combine(
self.offset_date(start, days, duration_type, self.get_calendar(task)), datetime.time(17)
self.offset_date(
start, days, duration_type, self.get_calendar(task)
),
datetime.time(17),
)
)
finishes.append(
datetime.datetime.combine(
self.offset_date(start, days, duration_type, self.get_calendar(predecessor)),
self.offset_date(
start,
days,
duration_type,
self.get_calendar(predecessor),
),
datetime.time(17),
)
)
@@ -138,14 +172,20 @@ class Usecase:
start = max(starts)
finish = max(finishes)
potential_finish = ifcopenshell.util.sequence.get_start_or_finish_date(
start, duration, task.TaskTime.DurationType, self.get_calendar(task), date_type="FINISH"
start,
duration,
task.TaskTime.DurationType,
self.get_calendar(task),
date_type="FINISH",
)
if potential_finish > finish:
start_ifc = ifcopenshell.util.date.datetime2ifc(start, "IfcDateTime")
if task.TaskTime.ScheduleStart == start_ifc and not is_first_task:
return
task.TaskTime.ScheduleStart = start_ifc
task.TaskTime.ScheduleFinish = ifcopenshell.util.date.datetime2ifc(potential_finish, "IfcDateTime")
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 and not is_first_task:
@@ -153,7 +193,11 @@ class Usecase:
task.TaskTime.ScheduleFinish = finish_ifc
task.TaskTime.ScheduleStart = ifcopenshell.util.date.datetime2ifc(
ifcopenshell.util.sequence.get_start_or_finish_date(
finish, duration, task.TaskTime.DurationType, self.get_calendar(task), date_type="START"
finish,
duration,
task.TaskTime.DurationType,
self.get_calendar(task),
date_type="START",
),
"IfcDateTime",
)
@@ -165,7 +209,11 @@ class Usecase:
task.TaskTime.ScheduleFinish = finish_ifc
task.TaskTime.ScheduleStart = ifcopenshell.util.date.datetime2ifc(
ifcopenshell.util.sequence.get_start_or_finish_date(
finish, duration, task.TaskTime.DurationType, self.get_calendar(task), date_type="START"
finish,
duration,
task.TaskTime.DurationType,
self.get_calendar(task),
date_type="START",
),
"IfcDateTime",
)
@@ -177,7 +225,11 @@ class Usecase:
task.TaskTime.ScheduleStart = start_ifc
task.TaskTime.ScheduleFinish = ifcopenshell.util.date.datetime2ifc(
ifcopenshell.util.sequence.get_start_or_finish_date(
start, duration, task.TaskTime.DurationType, self.get_calendar(task), date_type="FINISH"
start,
duration,
task.TaskTime.DurationType,
self.get_calendar(task),
date_type="FINISH",
),
"IfcDateTime",
)
@@ -190,11 +242,15 @@ class Usecase:
def get_calendar(self, task):
if task.id() not in self.calendar_cache:
self.calendar_cache[task.id()] = ifcopenshell.util.sequence.derive_calendar(task)
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 ifcopenshell.util.sequence.offset_date(date, datetime.timedelta(days=days), duration_type, calendar)
return ifcopenshell.util.sequence.offset_date(
date, datetime.timedelta(days=days), duration_type, calendar
)
def get_task_time_attribute(self, task, attribute):
if task.TaskTime:
src/ifcopenshell-python/ifcopenshell/api/sequence/data.py
+54 -14
View File
@@ -68,10 +68,14 @@ class Data:
del data["OwnerHistory"]
if data["Creators"]:
data["Creators"] = [p.id() for p in data["Creators"]]
data["CreationDate"] = ifcopenshell.util.date.ifc2datetime(data["CreationDate"])
data["CreationDate"] = ifcopenshell.util.date.ifc2datetime(
data["CreationDate"]
)
data["StartTime"] = ifcopenshell.util.date.ifc2datetime(data["StartTime"])
if data["FinishTime"]:
data["FinishTime"] = ifcopenshell.util.date.ifc2datetime(data["FinishTime"])
data["FinishTime"] = ifcopenshell.util.date.ifc2datetime(
data["FinishTime"]
)
data["RelatedObjects"] = []
for rel in work_schedule.Controls:
for obj in rel.RelatedObjects:
@@ -86,7 +90,9 @@ class Data:
data = work_calendar.get_info()
del data["OwnerHistory"]
data["WorkingTimes"] = [t.id() for t in work_calendar.WorkingTimes or []]
data["ExceptionTimes"] = [t.id() for t in work_calendar.ExceptionTimes or []]
data["ExceptionTimes"] = [
t.id() for t in work_calendar.ExceptionTimes or []
]
cls.work_calendars[work_calendar.id()] = data
@classmethod
@@ -94,9 +100,21 @@ class Data:
cls.work_times = {}
for work_time in cls._file.by_type("IfcWorkTime"):
data = work_time.get_info()
data["Start"] = ifcopenshell.util.date.ifc2datetime(data["Start"]) if data["Start"] else None
data["Finish"] = ifcopenshell.util.date.ifc2datetime(data["Finish"]) if data["Finish"] else None
data["RecurrencePattern"] = work_time.RecurrencePattern.id() if work_time.RecurrencePattern else None
data["Start"] = (
ifcopenshell.util.date.ifc2datetime(data["Start"])
if data["Start"]
else None
)
data["Finish"] = (
ifcopenshell.util.date.ifc2datetime(data["Finish"])
if data["Finish"]
else None
)
data["RecurrencePattern"] = (
work_time.RecurrencePattern.id()
if work_time.RecurrencePattern
else None
)
cls.work_times[work_time.id()] = data
@classmethod
@@ -133,7 +151,11 @@ class Data:
if task.TaskTime:
data["TaskTime"] = data["TaskTime"].id()
for rel in task.IsNestedBy:
[data["RelatedObjects"].append(o.id()) for o in rel.RelatedObjects if o.is_a("IfcTask")]
[
data["RelatedObjects"].append(o.id())
for o in rel.RelatedObjects
if o.is_a("IfcTask")
]
data["Nests"] = [r.RelatingObject.id() for r in task.Nests or []]
[
data["Outputs"].append(r.RelatingProduct.id())
@@ -141,20 +163,36 @@ class Data:
if r.is_a("IfcRelAssignsToProduct")
]
[
data["Resources"].extend([o.id() for o in r.RelatedObjects if o.is_a("IfcResource")])
data["Resources"].extend(
[o.id() for o in r.RelatedObjects if o.is_a("IfcResource")]
)
for r in task.OperatesOn
]
[
data["Controls"].extend([o.id() for o in r.RelatedObjects if o.is_a("IfcControl")])
data["Controls"].extend(
[o.id() for o in r.RelatedObjects if o.is_a("IfcControl")]
)
for r in task.OperatesOn
]
[data["Inputs"].extend([o.id() for o in r.RelatedObjects if o.is_a("IfcProduct")]) for r in task.OperatesOn]
[data["IsPredecessorTo"].append(rel.id()) for rel in task.IsPredecessorTo or []]
[data["IsSuccessorFrom"].append(rel.id()) for rel in task.IsSuccessorFrom or []]
[
data["Inputs"].extend(
[o.id() for o in r.RelatedObjects if o.is_a("IfcProduct")]
)
for r in task.OperatesOn
]
[
data["IsPredecessorTo"].append(rel.id())
for rel in task.IsPredecessorTo or []
]
[
data["IsSuccessorFrom"].append(rel.id())
for rel in task.IsSuccessorFrom or []
]
[
data["HasAssignmentsWorkCalendar"].append(rel.RelatingControl.id())
for rel in task.HasAssignments or []
if rel.is_a("IfcRelAssignsToControl") and rel.RelatingControl.is_a("IfcWorkCalendar")
if rel.is_a("IfcRelAssignsToControl")
and rel.RelatingControl.is_a("IfcWorkCalendar")
]
cls.tasks[task.id()] = data
@@ -179,7 +217,9 @@ class Data:
data = lag_time.get_info()
if data["LagValue"]:
if data["LagValue"].is_a("IfcDuration"):
data["LagValue"] = ifcopenshell.util.date.ifc2datetime(data["LagValue"].wrappedValue)
data["LagValue"] = ifcopenshell.util.date.ifc2datetime(
data["LagValue"].wrappedValue
)
else:
data["LagValue"] = float(data["LagValue"].wrappedValue)
cls.lag_times[lag_time.id()] = data
src/ifcopenshell-python/ifcopenshell/api/sequence/edit_lag_time.py
+11 -3
View File
@@ -33,7 +33,15 @@ class Usecase:
if isinstance(value, float):
value = self.file.createIfcRatioMeasure(value)
else:
value = self.file.createIfcDuration(ifcopenshell.util.date.datetime2ifc(value, "IfcDuration"))
value = self.file.createIfcDuration(
ifcopenshell.util.date.datetime2ifc(value, "IfcDuration")
)
setattr(self.settings["lag_time"], name, value)
for rel in [r for r in self.file.get_inverse(self.settings["lag_time"]) if r.is_a("IfcRelSequence")]:
ifcopenshell.api.run("sequence.cascade_schedule", self.file, task=rel.RelatedProcess)
for rel in [
r
for r in self.file.get_inverse(self.settings["lag_time"])
if r.is_a("IfcRelSequence")
]:
ifcopenshell.api.run(
"sequence.cascade_schedule", self.file, task=rel.RelatedProcess
)
src/ifcopenshell-python/ifcopenshell/api/sequence/edit_sequence.py
+3 -1
View File
@@ -32,5 +32,7 @@ class Usecase:
setattr(self.settings["rel_sequence"], name, value)
if "SequenceType" in self.settings["attributes"].keys():
ifcopenshell.api.run(
"sequence.cascade_schedule", self.file, task=self.settings["rel_sequence"].RelatedProcess
"sequence.cascade_schedule",
self.file,
task=self.settings["rel_sequence"].RelatedProcess,
)
src/ifcopenshell-python/ifcopenshell/api/sequence/edit_task_time.py
+51 -14
View File
@@ -39,13 +39,17 @@ class Usecase:
):
del self.settings["attributes"]["ScheduleFinish"]
duration_type = self.settings["attributes"].get("DurationType", self.settings["task_time"].DurationType)
duration_type = self.settings["attributes"].get(
"DurationType", self.settings["task_time"].DurationType
)
finish = self.settings["attributes"].get("ScheduleFinish", None)
if finish:
if isinstance(finish, str):
finish = datetime.datetime.fromisoformat(finish)
self.settings["attributes"]["ScheduleFinish"] = datetime.datetime.combine(
ifcopenshell.util.sequence.get_soonest_working_day(finish, duration_type, self.calendar),
ifcopenshell.util.sequence.get_soonest_working_day(
finish, duration_type, self.calendar
),
datetime.time(17),
)
start = self.settings["attributes"].get("ScheduleStart", None)
@@ -53,7 +57,9 @@ class Usecase:
if isinstance(start, str):
start = datetime.datetime.fromisoformat(start)
self.settings["attributes"]["ScheduleStart"] = datetime.datetime.combine(
ifcopenshell.util.sequence.get_soonest_working_day(start, duration_type, self.calendar),
ifcopenshell.util.sequence.get_soonest_working_day(
start, duration_type, self.calendar
),
datetime.time(9),
)
@@ -61,7 +67,11 @@ class Usecase:
if value is not None:
if "Start" in name or "Finish" in name or name == "StatusTime":
value = ifcopenshell.util.date.datetime2ifc(value, "IfcDateTime")
elif name == "ScheduleDuration" or name == "ActualDuration" or name == "RemainingTime":
elif (
name == "ScheduleDuration"
or name == "ActualDuration"
or name == "RemainingTime"
):
value = ifcopenshell.util.date.datetime2ifc(value, "IfcDuration")
setattr(self.settings["task_time"], name, value)
@@ -71,9 +81,15 @@ class Usecase:
and self.settings["task_time"].ScheduleStart
):
self.calculate_finish()
elif self.settings["attributes"].get("ScheduleStart", None) and self.settings["task_time"].ScheduleDuration:
elif (
self.settings["attributes"].get("ScheduleStart", None)
and self.settings["task_time"].ScheduleDuration
):
self.calculate_finish()
elif self.settings["attributes"].get("ScheduleFinish", None) and self.settings["task_time"].ScheduleStart:
elif (
self.settings["attributes"].get("ScheduleFinish", None)
and self.settings["task_time"].ScheduleStart
):
self.calculate_duration()
if self.settings["task_time"].ScheduleDuration and (
@@ -85,27 +101,48 @@ class Usecase:
def calculate_finish(self):
finish = ifcopenshell.util.sequence.get_start_or_finish_date(
ifcopenshell.util.date.ifc2datetime(self.settings["task_time"].ScheduleStart),
ifcopenshell.util.date.ifc2datetime(self.settings["task_time"].ScheduleDuration),
ifcopenshell.util.date.ifc2datetime(
self.settings["task_time"].ScheduleStart
),
ifcopenshell.util.date.ifc2datetime(
self.settings["task_time"].ScheduleDuration
),
self.settings["task_time"].DurationType,
self.calendar,
date_type="FINISH",
)
self.settings["task_time"].ScheduleFinish = ifcopenshell.util.date.datetime2ifc(finish, "IfcDateTime")
self.settings["task_time"].ScheduleFinish = ifcopenshell.util.date.datetime2ifc(
finish, "IfcDateTime"
)
def calculate_duration(self):
start = ifcopenshell.util.date.ifc2datetime(self.settings["task_time"].ScheduleStart)
finish = ifcopenshell.util.date.ifc2datetime(self.settings["task_time"].ScheduleFinish)
start = ifcopenshell.util.date.ifc2datetime(
self.settings["task_time"].ScheduleStart
)
finish = ifcopenshell.util.date.ifc2datetime(
self.settings["task_time"].ScheduleFinish
)
current_date = datetime.date(start.year, start.month, start.day)
finish_date = datetime.date(finish.year, finish.month, finish.day)
duration = datetime.timedelta(days=1)
while current_date < finish_date:
if self.settings["task_time"].DurationType == "ELAPSEDTIME" or not self.calendar:
if (
self.settings["task_time"].DurationType == "ELAPSEDTIME"
or not self.calendar
):
duration += datetime.timedelta(days=1)
elif ifcopenshell.util.sequence.is_working_day(current_date, self.calendar):
duration += datetime.timedelta(days=1)
current_date += datetime.timedelta(days=1)
self.settings["task_time"].ScheduleDuration = ifcopenshell.util.date.datetime2ifc(duration, "IfcDuration")
self.settings[
"task_time"
].ScheduleDuration = ifcopenshell.util.date.datetime2ifc(
duration, "IfcDuration"
)
def get_task(self):
return [e for e in self.file.get_inverse(self.settings["task_time"]) if e.is_a("IfcTask")][0]
return [
e
for e in self.file.get_inverse(self.settings["task_time"])
if e.is_a("IfcTask")
][0]
src/ifcopenshell-python/ifcopenshell/api/sequence/recalculate_schedule.py
+117 -35
View File
@@ -59,10 +59,12 @@ class Usecase:
if attempts > max_worst_case_attempts:
is_cyclic = True
break # We have an infinite loop due to a cyclic graph
break # We have an infinite loop due to a cyclic graph
if is_cyclic:
raise RecursionError("Task graph is cyclic and so critical path method cannot be performed.")
raise RecursionError(
"Task graph is cyclic and so critical path method cannot be performed."
)
return
self.pending_nodes = set(self.g.nodes)
@@ -88,7 +90,9 @@ class Usecase:
self.g = nx.DiGraph()
self.edges = []
self.g.add_node("start", duration=0, duration_type="ELAPSEDTIME", calendar=None)
self.g.add_node("finish", duration=0, duration_type="ELAPSEDTIME", calendar=None)
self.g.add_node(
"finish", duration=0, duration_type="ELAPSEDTIME", calendar=None
)
for rel in self.settings["work_schedule"].Controls:
for related_object in rel.RelatedObjects:
if not related_object.is_a("IfcTask"):
@@ -103,7 +107,9 @@ class Usecase:
return
if task.TaskTime and task.TaskTime.ScheduleDuration:
duration = ifcopenshell.util.date.ifc2datetime(task.TaskTime.ScheduleDuration).days
duration = ifcopenshell.util.date.ifc2datetime(
task.TaskTime.ScheduleDuration
).days
duration_type = task.TaskTime.DurationType
else:
duration = 0
@@ -124,7 +130,9 @@ class Usecase:
{
"lag_time": 0
if not rel.TimeLag
else ifcopenshell.util.date.ifc2datetime(rel.TimeLag.LagValue.wrappedValue).days,
else ifcopenshell.util.date.ifc2datetime(
rel.TimeLag.LagValue.wrappedValue
).days,
"type": self.sequence_type_map[rel.SequenceType],
},
)
@@ -137,7 +145,9 @@ class Usecase:
if not predecessor_types:
self.edges.append(("start", task.id(), {"lag_time": 0, "type": "FS"}))
if task.TaskTime and task.TaskTime.ScheduleStart:
self.start_dates.append(ifcopenshell.util.date.ifc2datetime(task.TaskTime.ScheduleStart))
self.start_dates.append(
ifcopenshell.util.date.ifc2datetime(task.TaskTime.ScheduleStart)
)
if not successor_types:
self.edges.append((task.id(), "finish", {"lag_time": 0, "type": "FF"}))
@@ -154,13 +164,25 @@ class Usecase:
self.file,
task_time=task.TaskTime,
attributes={
"FreeFloat": ifcopenshell.util.date.datetime2ifc(data["free_float"], "IfcDuration"),
"TotalFloat": ifcopenshell.util.date.datetime2ifc(data["total_float"], "IfcDuration"),
"FreeFloat": ifcopenshell.util.date.datetime2ifc(
data["free_float"], "IfcDuration"
),
"TotalFloat": ifcopenshell.util.date.datetime2ifc(
data["total_float"], "IfcDuration"
),
"IsCritical": data["total_float"].days == 0,
"EarlyStart": ifcopenshell.util.date.datetime2ifc(data["early_start"], "IfcDateTime"),
"EarlyFinish": ifcopenshell.util.date.datetime2ifc(data["early_finish"], "IfcDateTime"),
"LateStart": ifcopenshell.util.date.datetime2ifc(data["late_start"], "IfcDateTime"),
"LateFinish": ifcopenshell.util.date.datetime2ifc(data["late_finish"], "IfcDateTime"),
"EarlyStart": ifcopenshell.util.date.datetime2ifc(
data["early_start"], "IfcDateTime"
),
"EarlyFinish": ifcopenshell.util.date.datetime2ifc(
data["early_finish"], "IfcDateTime"
),
"LateStart": ifcopenshell.util.date.datetime2ifc(
data["late_start"], "IfcDateTime"
),
"LateFinish": ifcopenshell.util.date.datetime2ifc(
data["late_finish"], "IfcDateTime"
),
},
)
@@ -190,10 +212,15 @@ class Usecase:
if edge["lag_time"]:
days += edge["lag_time"]
if days:
starts.append(datetime.datetime.combine(self.offset_date(finish, days, data), datetime.time(9)))
starts.append(
datetime.datetime.combine(
self.offset_date(finish, days, predecessor_data), datetime.time(9)
self.offset_date(finish, days, data), datetime.time(9)
)
)
starts.append(
datetime.datetime.combine(
self.offset_date(finish, days, predecessor_data),
datetime.time(9),
)
)
else:
@@ -204,7 +231,9 @@ class Usecase:
return
if edge["lag_time"]:
starts.append(self.offset_date(start, edge["lag_time"], data))
starts.append(self.offset_date(start, edge["lag_time"], predecessor_data))
starts.append(
self.offset_date(start, edge["lag_time"], predecessor_data)
)
else:
starts.append(start)
elif edge["type"] == "FF":
@@ -212,8 +241,12 @@ class Usecase:
if finish is None:
return
if edge["lag_time"]:
finishes.append(self.offset_date(finish, edge["lag_time"], data))
finishes.append(self.offset_date(finish, edge["lag_time"], predecessor_data))
finishes.append(
self.offset_date(finish, edge["lag_time"], data)
)
finishes.append(
self.offset_date(finish, edge["lag_time"], predecessor_data)
)
else:
finishes.append(finish)
elif edge["type"] == "SF":
@@ -225,11 +258,14 @@ class Usecase:
days += edge["lag_time"]
if days or edge["lag_time"]:
finishes.append(
datetime.datetime.combine(self.offset_date(start, days, data), datetime.time(17))
datetime.datetime.combine(
self.offset_date(start, days, data), datetime.time(17)
)
)
finishes.append(
datetime.datetime.combine(
self.offset_date(start, days, predecessor_data), datetime.time(17)
self.offset_date(start, days, predecessor_data),
datetime.time(17),
)
)
else:
@@ -247,7 +283,9 @@ class Usecase:
if potential_finish > data["early_finish"]:
data["early_finish"] = potential_finish
else:
data["early_start"] = ifcopenshell.util.sequence.get_start_or_finish_date(
data[
"early_start"
] = ifcopenshell.util.sequence.get_start_or_finish_date(
data["early_finish"],
datetime.timedelta(days=data["duration"]),
data["duration_type"],
@@ -303,10 +341,15 @@ class Usecase:
days += edge["lag_time"]
if days or edge["lag_time"]:
finishes.append(
datetime.datetime.combine(self.offset_date(start, -days, data), datetime.time(17))
datetime.datetime.combine(
self.offset_date(start, -days, data), datetime.time(17)
)
)
finishes.append(
datetime.datetime.combine(self.offset_date(start, -days, successor_data), datetime.time(17))
datetime.datetime.combine(
self.offset_date(start, -days, successor_data),
datetime.time(17),
)
)
else:
finishes.append(start)
@@ -325,12 +368,18 @@ class Usecase:
return
if edge["lag_time"]:
starts.append(self.offset_date(start, -edge["lag_time"], data))
starts.append(self.offset_date(start, -edge["lag_time"], successor_data))
starts.append(
self.offset_date(start, -edge["lag_time"], successor_data)
)
else:
starts.append(start)
free_floats.append(
self.calculate_free_float(
data["early_start"], successor_data["early_start"], edge["lag_time"], data, successor_data
data["early_start"],
successor_data["early_start"],
edge["lag_time"],
data,
successor_data,
)
)
elif edge["type"] == "FF":
@@ -338,13 +387,21 @@ class Usecase:
if finish is None:
return
if edge["lag_time"]:
finishes.append(self.offset_date(finish, -edge["lag_time"], data))
finishes.append(self.offset_date(finish, -edge["lag_time"], successor_data))
finishes.append(
self.offset_date(finish, -edge["lag_time"], data)
)
finishes.append(
self.offset_date(finish, -edge["lag_time"], successor_data)
)
else:
finishes.append(finish)
free_floats.append(
self.calculate_free_float(
data["early_finish"], successor_data["early_finish"], edge["lag_time"], data, successor_data
data["early_finish"],
successor_data["early_finish"],
edge["lag_time"],
data,
successor_data,
)
)
elif edge["type"] == "SF":
@@ -356,23 +413,37 @@ class Usecase:
days += edge["lag_time"]
if days:
starts.append(
datetime.datetime.combine(self.offset_date(finish, -days, data), datetime.time(9))
datetime.datetime.combine(
self.offset_date(finish, -days, data), datetime.time(9)
)
)
starts.append(
datetime.datetime.combine(self.offset_date(finish, -days, successor_data), datetime.time(9))
datetime.datetime.combine(
self.offset_date(finish, -days, successor_data),
datetime.time(9),
)
)
else:
starts.append(finish)
free_floats.append(
self.calculate_free_float(
data["early_start"], successor_data["early_finish"], edge["lag_time"], data, successor_data
data["early_start"],
successor_data["early_finish"],
edge["lag_time"],
data,
successor_data,
)
)
if starts and finishes:
data["late_start"] = min(starts)
data["late_finish"] = min(finishes)
if self.offset_date(data["late_start"], data["duration"], data) < data["late_finish"]:
data["late_finish"] = ifcopenshell.util.sequence.get_start_or_finish_date(
if (
self.offset_date(data["late_start"], data["duration"], data)
< data["late_finish"]
):
data[
"late_finish"
] = ifcopenshell.util.sequence.get_start_or_finish_date(
data["late_start"],
datetime.timedelta(days=data["duration"]),
data["duration_type"],
@@ -380,7 +451,9 @@ class Usecase:
date_type="FINISH",
)
else:
data["late_start"] = ifcopenshell.util.sequence.get_start_or_finish_date(
data[
"late_start"
] = ifcopenshell.util.sequence.get_start_or_finish_date(
data["late_finish"],
datetime.timedelta(days=data["duration"]),
data["duration_type"],
@@ -421,7 +494,9 @@ class Usecase:
data["total_float"] = data["late_finish"] - data["early_finish"]
# If the float is within the span of a single day, it may show as a 8 hours
if data["total_float"].seconds == 60 * 60 * 8:
data["total_float"] = datetime.timedelta(days=data["total_float"].days + 1)
data["total_float"] = datetime.timedelta(
days=data["total_float"].days + 1
)
data["free_float"] = min(free_floats) if free_floats else None
# If the float is within the span of a single day, it may show as a 8 hours
@@ -430,7 +505,14 @@ class Usecase:
return True
def calculate_free_float(self, predecessor_date, successor_date, lag_time, predecessor_data, successor_data):
def calculate_free_float(
self,
predecessor_date,
successor_date,
lag_time,
predecessor_data,
successor_data,
):
if not lag_time:
min_successor_date = successor_date
else:
src/ifcopenshell-python/ifcopenshell/api/sequence/remove_task.py
+3 -1
View File
@@ -40,7 +40,9 @@ class Usecase:
elif inverse.is_a("IfcRelNests"):
if inverse.RelatingObject == self.settings["task"]:
for related_object in inverse.RelatedObjects:
ifcopenshell.api.run("sequence.remove_task", self.file, task=related_object)
ifcopenshell.api.run(
"sequence.remove_task", self.file, task=related_object
)
elif inverse.RelatedObjects == tuple(self.settings["task"]):
self.file.remove(inverse)
elif inverse.is_a("IfcRelAssignsToControl"):
src/ifcopenshell-python/ifcopenshell/api/sequence/remove_work_calendar.py
+5 -1
View File
@@ -40,5 +40,9 @@ class Usecase:
ifcopenshell.api.run(
"control.unassign_control",
self.file,
**{"relating_control": self.settings["work_calendar"], "related_object": object})
**{
"relating_control": self.settings["work_calendar"],
"related_object": object,
}
)
self.file.remove(self.settings["work_calendar"])
src/ifcopenshell-python/ifcopenshell/api/sequence/remove_work_schedule.py
+3 -1
View File
@@ -37,5 +37,7 @@ class Usecase:
for rel in self.settings["work_schedule"].Controls:
for related_object in rel.RelatedObjects:
if related_object.is_a("IfcTask"):
ifcopenshell.api.run("sequence.remove_task", self.file, task=related_object)
ifcopenshell.api.run(
"sequence.remove_task", self.file, task=related_object
)
self.file.remove(self.settings["work_schedule"])
src/ifcopenshell-python/ifcopenshell/api/sequence/unassign_lag_time.py
+5 -1
View File
@@ -33,4 +33,8 @@ class Usecase:
self.file.remove(self.settings["rel_sequence"].TimeLag)
else:
self.settings["rel_sequence"].TimeLag = None
ifcopenshell.api.run("sequence.cascade_schedule", self.file, task=self.settings["rel_sequence"].RelatedProcess)
ifcopenshell.api.run(
"sequence.cascade_schedule",
self.file,
task=self.settings["rel_sequence"].RelatedProcess,
)
src/ifcopenshell-python/ifcopenshell/api/sequence/unassign_process.py
+7 -2
View File
@@ -32,12 +32,17 @@ class Usecase:
def execute(self):
for rel in self.settings["related_object"].HasAssignments or []:
if not rel.is_a("IfcRelAssignsToProcess") or rel.RelatingProcess != self.settings["relating_process"]:
if (
not rel.is_a("IfcRelAssignsToProcess")
or rel.RelatingProcess != self.settings["relating_process"]
):
continue
if len(rel.RelatedObjects) == 1:
return self.file.remove(rel)
related_objects = list(rel.RelatedObjects)
related_objects.remove(self.settings["related_object"])
rel.RelatedObjects = related_objects
ifcopenshell.api.run("owner.update_owner_history", self.file, **{"element": rel})
ifcopenshell.api.run(
"owner.update_owner_history", self.file, **{"element": rel}
)
return rel
src/ifcopenshell-python/ifcopenshell/api/sequence/unassign_product.py
+7 -2
View File
@@ -32,12 +32,17 @@ class Usecase:
def execute(self):
for rel in self.settings["related_object"].HasAssignments or []:
if not rel.is_a("IfcRelAssignsToProduct") or rel.RelatingProduct != self.settings["relating_product"]:
if (
not rel.is_a("IfcRelAssignsToProduct")
or rel.RelatingProduct != self.settings["relating_product"]
):
continue
if len(rel.RelatedObjects) == 1:
return self.file.remove(rel)
related_objects = list(rel.RelatedObjects)
related_objects.remove(self.settings["related_object"])
rel.RelatedObjects = related_objects
ifcopenshell.api.run("owner.update_owner_history", self.file, **{"element": rel})
ifcopenshell.api.run(
"owner.update_owner_history", self.file, **{"element": rel}
)
return rel
src/ifcopenshell-python/ifcopenshell/api/sequence/unassign_sequence.py
+5 -1
View File
@@ -34,4 +34,8 @@ class Usecase:
for rel in self.settings["related_process"].IsSuccessorFrom or []:
if rel.RelatingProcess == self.settings["relating_process"]:
self.file.remove(rel)
ifcopenshell.api.run("sequence.cascade_schedule", self.file, task=self.settings["related_process"])
ifcopenshell.api.run(
"sequence.cascade_schedule",
self.file,
task=self.settings["related_process"],
)