mirror of
https://github.com/IfcOpenShell/IfcOpenShell.git
synced 2026-08-09 17:31:45 +00:00
typing
This commit is contained in:
@@ -29,11 +29,8 @@ def add_task_time(
|
||||
(especially for maintenance tasks).
|
||||
|
||||
:param task: The task to add time data to.
|
||||
:type task: ifcopenshell.entity_instance
|
||||
:param is_recurring: Whether or not the time should recur.
|
||||
:type is_recurring: bool
|
||||
:return: The newly created IfcTaskTime.
|
||||
:rtype: ifcopenshell.entity_instance
|
||||
|
||||
Example:
|
||||
|
||||
@@ -61,11 +58,9 @@ def add_task_time(
|
||||
ifcopenshell.api.sequence.edit_task_time(model,
|
||||
task_time=time, attributes={"ScheduleStart": "2000-01-01", "ScheduleDuration": "P2D"})
|
||||
"""
|
||||
settings = {"task": task, "is_recurring": is_recurring}
|
||||
|
||||
if settings["is_recurring"]:
|
||||
if is_recurring:
|
||||
task_time = file.create_entity("IfcTaskTimeRecurring")
|
||||
else:
|
||||
task_time = file.create_entity("IfcTaskTime")
|
||||
settings["task"].TaskTime = task_time
|
||||
task.TaskTime = task_time
|
||||
return task_time
|
||||
|
||||
@@ -34,17 +34,13 @@ def assign_lag_time(
|
||||
are allowed.
|
||||
|
||||
:param rel_sequence: The IfcRelSequence to assign the lag time to.
|
||||
:type rel_sequence: ifcopenshell.entity_instance
|
||||
:param lag_value: An ISO standardised duration string.
|
||||
:type lag_value: str
|
||||
:param duration_type: Choose from WORKTIME for the associated
|
||||
calendar-based lag times (this is the most common scenario and is
|
||||
recommended as a default), or ELAPSEDTIME to not follow the
|
||||
calendar. You may also choose NOTDEFINED but the behaviour of this
|
||||
is unclear.
|
||||
:type duration_type: str
|
||||
:return: The newly created IfcLagTime
|
||||
:rtype: ifcopenshell.entity_instance
|
||||
|
||||
Example:
|
||||
|
||||
@@ -84,16 +80,10 @@ def assign_lag_time(
|
||||
# for whatever reason.
|
||||
ifcopenshell.api.sequence.assign_lag_time(model, rel_sequence=sequence, lag_value="P1D")
|
||||
"""
|
||||
settings = {
|
||||
"rel_sequence": rel_sequence,
|
||||
"lag_value": lag_value,
|
||||
"duration_type": duration_type,
|
||||
}
|
||||
|
||||
lag_value = file.createIfcDuration(ifcopenshell.util.date.datetime2ifc(settings["lag_value"], "IfcDuration"))
|
||||
lag_time = file.create_entity("IfcLagTime", DurationType=settings["duration_type"], LagValue=lag_value)
|
||||
if settings["rel_sequence"].is_a("IfcRelSequence"):
|
||||
if settings["rel_sequence"].TimeLag and len(file.get_inverse(settings["rel_sequence"].TimeLag)) == 1:
|
||||
file.remove(settings["rel_sequence"].TimeLag)
|
||||
settings["rel_sequence"].TimeLag = lag_time
|
||||
lag_value = file.create_entity("IfcDuration", ifcopenshell.util.date.datetime2ifc(lag_value, "IfcDuration"))
|
||||
lag_time = file.create_entity("IfcLagTime", DurationType=duration_type, LagValue=lag_value)
|
||||
if rel_sequence.is_a("IfcRelSequence"):
|
||||
if rel_sequence.TimeLag and len(file.get_inverse(rel_sequence.TimeLag)) == 1:
|
||||
file.remove(rel_sequence.TimeLag)
|
||||
rel_sequence.TimeLag = lag_time
|
||||
return lag_time
|
||||
|
||||
@@ -65,11 +65,8 @@ def assign_recurrence_pattern(
|
||||
:param parent: Either an IfcTaskTimeRecurring if you are defining a
|
||||
recurring schedule for a task, or IfcWorkTime if you are defining a
|
||||
recurring pattern for a workdays or holidays in a calendar.
|
||||
:type parent: ifcopenshell.entity_instance
|
||||
:param recurrence_type: One of the types of recurrences.
|
||||
:type recurrence_type: str
|
||||
:return: The newly created IfcRecurrencePattern
|
||||
:rtype: ifcopenshell.entity_instance
|
||||
|
||||
Example:
|
||||
|
||||
@@ -108,16 +105,14 @@ def assign_recurrence_pattern(
|
||||
ifcopenshell.api.sequence.edit_recurrence_pattern(model,
|
||||
recurrence_pattern=pattern, attributes={"DayComponent": [1], "Interval": 6})
|
||||
"""
|
||||
settings = {"parent": parent, "recurrence_type": recurrence_type}
|
||||
recurrence = file.create_entity("IfcRecurrencePattern", recurrence_type)
|
||||
|
||||
recurrence = file.createIfcRecurrencePattern(settings["recurrence_type"])
|
||||
|
||||
if settings["parent"].is_a("IfcWorkTime"):
|
||||
if settings["parent"].RecurrencePattern and len(file.get_inverse(settings["parent"].RecurrencePattern)) == 1:
|
||||
file.remove(settings["parent"].RecurrencePattern)
|
||||
settings["parent"].RecurrencePattern = recurrence
|
||||
elif settings["parent"].is_a("IfcTaskTimeRecurring"):
|
||||
if recurrence_old := settings["parent"].Recurrence and len(file.get_inverse(recurrence_old)) == 1:
|
||||
if parent.is_a("IfcWorkTime"):
|
||||
if parent.RecurrencePattern and len(file.get_inverse(parent.RecurrencePattern)) == 1:
|
||||
file.remove(parent.RecurrencePattern)
|
||||
parent.RecurrencePattern = recurrence
|
||||
elif parent.is_a("IfcTaskTimeRecurring"):
|
||||
if (recurrence_old := parent.Recurrence) and len(file.get_inverse(recurrence_old)) == 1:
|
||||
file.remove(recurrence_old)
|
||||
settings["parent"].Recurrence = recurrence
|
||||
parent.Recurrence = recurrence
|
||||
return recurrence
|
||||
|
||||
@@ -51,13 +51,10 @@ def assign_sequence(
|
||||
predecessor and successor tasks in the planning profession.
|
||||
|
||||
:param relating_process: The previous / predecessor task.
|
||||
:type relating_process: ifcopenshell.entity_instance
|
||||
:param related_process: The next / successor task.
|
||||
:type related_process: ifcopenshell.entity_instance
|
||||
:param sequence_type: Choose from FINISH_START, FINISH_FINISH,
|
||||
START_START, or START_FINISH.
|
||||
:return: The newly created IfcRelSequence
|
||||
:rtype: ifcopenshell.entity_instance
|
||||
|
||||
Example:
|
||||
|
||||
@@ -109,24 +106,18 @@ def assign_sequence(
|
||||
# to be 2000-01-05.
|
||||
ifcopenshell.api.sequence.cascade_schedule(model, task=formwork)
|
||||
"""
|
||||
settings = {
|
||||
"relating_process": relating_process,
|
||||
"related_process": related_process,
|
||||
"sequence_type": sequence_type,
|
||||
}
|
||||
|
||||
for rel in settings["related_process"].IsSuccessorFrom or []:
|
||||
if rel.RelatingProcess == settings["relating_process"]:
|
||||
for rel in related_process.IsSuccessorFrom or []:
|
||||
if rel.RelatingProcess == relating_process:
|
||||
return rel
|
||||
rel = file.create_entity(
|
||||
"IfcRelSequence",
|
||||
**{
|
||||
"GlobalId": ifcopenshell.guid.new(),
|
||||
"OwnerHistory": ifcopenshell.api.owner.create_owner_history(file),
|
||||
"RelatingProcess": settings["relating_process"],
|
||||
"RelatedProcess": settings["related_process"],
|
||||
"SequenceType": settings["sequence_type"],
|
||||
"RelatingProcess": relating_process,
|
||||
"RelatedProcess": related_process,
|
||||
"SequenceType": sequence_type,
|
||||
}
|
||||
)
|
||||
ifcopenshell.api.sequence.cascade_schedule(file, task=settings["relating_process"])
|
||||
ifcopenshell.api.sequence.cascade_schedule(file, task=relating_process)
|
||||
return rel
|
||||
|
||||
@@ -20,6 +20,7 @@ import math
|
||||
import ifcopenshell.api.sequence
|
||||
import ifcopenshell.util.date
|
||||
import ifcopenshell.util.element
|
||||
from typing import Union
|
||||
|
||||
|
||||
def calculate_task_duration(file: ifcopenshell.file, task: ifcopenshell.entity_instance) -> None:
|
||||
@@ -35,9 +36,7 @@ def calculate_task_duration(file: ifcopenshell.file, task: ifcopenshell.entity_i
|
||||
then nothing happens.
|
||||
|
||||
:param task: The IfcTask to calculate the duration for.
|
||||
:type task: ifcopenshell.entity_instance
|
||||
:return: None
|
||||
:rtype: None
|
||||
|
||||
Example:
|
||||
|
||||
@@ -82,18 +81,20 @@ def calculate_task_duration(file: ifcopenshell.file, task: ifcopenshell.entity_i
|
||||
"""
|
||||
usecase = Usecase()
|
||||
usecase.file = file
|
||||
usecase.settings = {"task": task}
|
||||
return usecase.execute()
|
||||
return usecase.execute(task)
|
||||
|
||||
|
||||
class Usecase:
|
||||
def execute(self):
|
||||
file: ifcopenshell.file
|
||||
|
||||
def execute(self, task: ifcopenshell.entity_instance) -> None:
|
||||
self.task = task
|
||||
self.seconds_per_workday = self.calculate_seconds_per_workday()
|
||||
duration = self.calculate_max_resource_usage_duration()
|
||||
if duration:
|
||||
self.set_task_duration(duration)
|
||||
|
||||
def calculate_seconds_per_workday(self):
|
||||
def calculate_seconds_per_workday(self) -> float:
|
||||
def get_work_schedule(task):
|
||||
for rel in task.HasAssignments or []:
|
||||
if rel.is_a("IfcRelAssignsToControl") and rel.RelatingControl.is_a("IfcWorkSchedule"):
|
||||
@@ -102,7 +103,7 @@ class Usecase:
|
||||
return get_work_schedule(rel.RelatingObject)
|
||||
|
||||
default_seconds_per_workday = 8 * 60 * 60
|
||||
work_schedule = get_work_schedule(self.settings["task"])
|
||||
work_schedule = get_work_schedule(self.task)
|
||||
if not work_schedule:
|
||||
return default_seconds_per_workday
|
||||
psets = ifcopenshell.util.element.get_psets(work_schedule)
|
||||
@@ -115,9 +116,9 @@ class Usecase:
|
||||
work_day_duration = ifcopenshell.util.date.ifc2datetime(psets["Pset_WorkControlCommon"]["WorkDayDuration"])
|
||||
return work_day_duration.seconds
|
||||
|
||||
def calculate_max_resource_usage_duration(self):
|
||||
def calculate_max_resource_usage_duration(self) -> float:
|
||||
max_duration = 0
|
||||
for rel in self.settings["task"].OperatesOn or []:
|
||||
for rel in self.task.OperatesOn or []:
|
||||
for related_object in rel.RelatedObjects:
|
||||
if related_object.is_a("IfcConstructionResource"):
|
||||
duration = self.calculate_duration_in_days(related_object)
|
||||
@@ -125,7 +126,7 @@ class Usecase:
|
||||
max_duration = duration
|
||||
return max_duration
|
||||
|
||||
def calculate_duration_in_days(self, resource):
|
||||
def calculate_duration_in_days(self, resource: ifcopenshell.entity_instance) -> Union[float, None]:
|
||||
def is_hourly_work(schedule_work):
|
||||
return "T" in schedule_work
|
||||
|
||||
@@ -140,7 +141,7 @@ class Usecase:
|
||||
schedule_seconds = (schedule_duration.days + partial_days) * self.seconds_per_workday
|
||||
return math.ceil((schedule_seconds / self.seconds_per_workday) / schedule_usage)
|
||||
|
||||
def set_task_duration(self, duration):
|
||||
if not self.settings["task"].TaskTime:
|
||||
ifcopenshell.api.sequence.add_task_time(self.file, task=self.settings["task"])
|
||||
self.settings["task"].TaskTime.ScheduleDuration = f"P{duration}D"
|
||||
def set_task_duration(self, duration: float) -> None:
|
||||
if not (task_time := self.task.TaskTime):
|
||||
ifcopenshell.api.sequence.add_task_time(self.file, task=self.task)
|
||||
task_time.ScheduleDuration = f"P{duration}D"
|
||||
|
||||
@@ -24,7 +24,7 @@ import ifcopenshell.guid
|
||||
import ifcopenshell.util.element
|
||||
import ifcopenshell.util.sequence
|
||||
import ifcopenshell.util.system
|
||||
from typing import Optional
|
||||
from typing import Optional, Union
|
||||
|
||||
|
||||
def create_baseline(
|
||||
@@ -44,11 +44,8 @@ def create_baseline(
|
||||
* Same Resource Relationships
|
||||
|
||||
:param work_schedule: The planned work_schedule to baseline
|
||||
:type work_schedule: ifcopenshell.entity_instance
|
||||
:param name: baseline work schedule name
|
||||
:type name: str, optional
|
||||
:return: The baseline work_schedule
|
||||
:rtype: ifcopenshell.entity_instance
|
||||
|
||||
Example:
|
||||
|
||||
@@ -62,23 +59,20 @@ def create_baseline(
|
||||
"""
|
||||
usecase = Usecase()
|
||||
usecase.file = file
|
||||
usecase.settings = {"work_schedule": work_schedule, "name": name}
|
||||
return usecase.execute()
|
||||
return usecase.execute(work_schedule, name)
|
||||
|
||||
|
||||
class Usecase:
|
||||
def execute(self):
|
||||
result = self.create_baseline_work_schedule(self.settings["work_schedule"])
|
||||
return result
|
||||
file: ifcopenshell.file
|
||||
|
||||
def create_baseline_work_schedule(self, work_schedule):
|
||||
def execute(self, work_schedule: ifcopenshell.entity_instance, name: Union[str, None]) -> None:
|
||||
# create work schedule
|
||||
if not work_schedule.PredefinedType == "PLANNED":
|
||||
return
|
||||
baseline_work_schedule = ifcopenshell.api.sequence.add_work_schedule(
|
||||
self.file, name=work_schedule.Name, predefined_type="BASELINE"
|
||||
)
|
||||
baseline_work_schedule.Name = self.settings["name"]
|
||||
baseline_work_schedule.Name = name
|
||||
self.create_baseline_reference(work_schedule, baseline_work_schedule)
|
||||
for summary_task in ifcopenshell.util.sequence.get_root_tasks(work_schedule):
|
||||
current, duplicate = ifcopenshell.api.sequence.duplicate_task(self.file, task=summary_task)
|
||||
@@ -88,7 +82,9 @@ class Usecase:
|
||||
for i, task in enumerate(current):
|
||||
self.create_baseline_reference(task, duplicate[i])
|
||||
|
||||
def create_baseline_reference(self, relating_object, related_object):
|
||||
def create_baseline_reference(
|
||||
self, relating_object: ifcopenshell.entity_instance, related_object: ifcopenshell.entity_instance
|
||||
) -> ifcopenshell.entity_instance:
|
||||
referenced_by = None
|
||||
if relating_object.Declares:
|
||||
referenced_by = relating_object.Declares[0]
|
||||
|
||||
@@ -23,9 +23,12 @@ import ifcopenshell.api.owner
|
||||
import ifcopenshell.api.sequence
|
||||
import ifcopenshell.util.element
|
||||
import ifcopenshell.util.sequence
|
||||
from typing import Union, Any
|
||||
|
||||
|
||||
def duplicate_task(file: ifcopenshell.file, task: ifcopenshell.entity_instance) -> ifcopenshell.entity_instance:
|
||||
def duplicate_task(
|
||||
file: ifcopenshell.file, task: ifcopenshell.entity_instance
|
||||
) -> Union[ifcopenshell.entity_instance, list[ifcopenshell.entity_instance]]:
|
||||
"""Duplicates a task in the project
|
||||
|
||||
The following relationships are also duplicated:
|
||||
@@ -35,9 +38,7 @@ def duplicate_task(file: ifcopenshell.file, task: ifcopenshell.entity_instance)
|
||||
* The copy will have duplicated nested tasks
|
||||
|
||||
:param task: The task to be duplicated
|
||||
:type task: ifcopenshell.entity_instance
|
||||
:return: The duplicated task or the list of duplicated tasks if the latter has children
|
||||
:rtype: ifcopenshell.entity_instance or list of ifcopenshell.entity_instance
|
||||
|
||||
Example:
|
||||
.. code:: python
|
||||
@@ -55,6 +56,9 @@ def duplicate_task(file: ifcopenshell.file, task: ifcopenshell.entity_instance)
|
||||
|
||||
|
||||
class Usecase:
|
||||
file: ifcopenshell.file
|
||||
settings: dict[str, Any]
|
||||
|
||||
def execute(self):
|
||||
self.tracker = {"current": [], "duplicate": []}
|
||||
self.duplicate_task(self.settings["task"])
|
||||
|
||||
@@ -28,11 +28,8 @@ def edit_lag_time(file: ifcopenshell.file, lag_time: ifcopenshell.entity_instanc
|
||||
IfcLagTime, consult the IFC documentation.
|
||||
|
||||
:param lag_time: The IfcLagTime entity you want to edit
|
||||
:type lag_time: ifcopenshell.entity_instance
|
||||
:param attributes: a dictionary of attribute names and values.
|
||||
:type attributes: dict
|
||||
:return: None
|
||||
:rtype: None
|
||||
|
||||
Example:
|
||||
|
||||
@@ -75,14 +72,12 @@ def edit_lag_time(file: ifcopenshell.file, lag_time: ifcopenshell.entity_instanc
|
||||
# Or, let's make it 2 days instead.
|
||||
ifcopenshell.api.sequence.edit_lag_time(model, lag_time=lag, attributes={"LagValue": "P2D"})
|
||||
"""
|
||||
settings = {"lag_time": lag_time, "attributes": attributes}
|
||||
|
||||
for name, value in settings["attributes"].items():
|
||||
for name, value in attributes.items():
|
||||
if name == "LagValue" and value is not None:
|
||||
if isinstance(value, float):
|
||||
value = file.createIfcRatioMeasure(value)
|
||||
else:
|
||||
value = file.createIfcDuration(ifcopenshell.util.date.datetime2ifc(value, "IfcDuration"))
|
||||
setattr(settings["lag_time"], name, value)
|
||||
for rel in [r for r in file.get_inverse(settings["lag_time"]) if r.is_a("IfcRelSequence")]:
|
||||
setattr(lag_time, name, value)
|
||||
for rel in [r for r in file.get_inverse(lag_time) if r.is_a("IfcRelSequence")]:
|
||||
ifcopenshell.api.sequence.cascade_schedule(file, task=rel.RelatedProcess)
|
||||
|
||||
@@ -30,11 +30,8 @@ def edit_recurrence_pattern(
|
||||
IfcRecurrencePattern, consult the IFC documentation.
|
||||
|
||||
:param recurrence_pattern: The IfcRecurrencePattern entity you want to edit
|
||||
:type recurrence_pattern: ifcopenshell.entity_instance
|
||||
:param attributes: a dictionary of attribute names and values.
|
||||
:type attributes: dict
|
||||
:return: None
|
||||
:rtype: None
|
||||
|
||||
Example:
|
||||
|
||||
@@ -55,13 +52,8 @@ def edit_recurrence_pattern(
|
||||
ifcopenshell.api.sequence.edit_recurrence_pattern(model,
|
||||
recurrence_pattern=pattern, attributes={"WeekdayComponent": [1, 2, 3, 4, 5]})
|
||||
"""
|
||||
settings = {
|
||||
"recurrence_pattern": recurrence_pattern,
|
||||
"attributes": attributes,
|
||||
}
|
||||
|
||||
for name, value in settings["attributes"].items():
|
||||
setattr(settings["recurrence_pattern"], name, value)
|
||||
for name, value in attributes.items():
|
||||
setattr(recurrence_pattern, name, value)
|
||||
|
||||
ifcopenshell.util.sequence.is_working_day.cache_clear()
|
||||
ifcopenshell.util.sequence.is_calendar_applicable.cache_clear()
|
||||
|
||||
@@ -30,11 +30,8 @@ def edit_sequence(
|
||||
IfcRelSequence, consult the IFC documentation.
|
||||
|
||||
:param rel_sequence: The IfcRelSequence entity you want to edit
|
||||
:type rel_sequence: ifcopenshell.entity_instance
|
||||
:param attributes: a dictionary of attribute names and values.
|
||||
:type attributes: dict
|
||||
:return: None
|
||||
:rtype: None
|
||||
|
||||
Example:
|
||||
|
||||
@@ -62,9 +59,7 @@ def edit_sequence(
|
||||
ifcopenshell.api.sequence.edit_sequence(model,
|
||||
rel_sequence=sequence, attributes={"SequenceType": "START_START"})
|
||||
"""
|
||||
settings = {"rel_sequence": rel_sequence, "attributes": attributes}
|
||||
|
||||
for name, value in settings["attributes"].items():
|
||||
setattr(settings["rel_sequence"], name, value)
|
||||
if "SequenceType" in settings["attributes"].keys():
|
||||
ifcopenshell.api.sequence.cascade_schedule(file, task=settings["rel_sequence"].RelatedProcess)
|
||||
for name, value in attributes.items():
|
||||
setattr(rel_sequence, name, value)
|
||||
if "SequenceType" in attributes.keys():
|
||||
ifcopenshell.api.sequence.cascade_schedule(file, task=rel_sequence.RelatedProcess)
|
||||
|
||||
@@ -26,11 +26,8 @@ def edit_task(file: ifcopenshell.file, task: ifcopenshell.entity_instance, attri
|
||||
IfcTask, consult the IFC documentation.
|
||||
|
||||
:param task: The IfcTask entity you want to edit
|
||||
:type task: ifcopenshell.entity_instance
|
||||
:param attributes: a dictionary of attribute names and values.
|
||||
:type attributes: dict
|
||||
:return: None
|
||||
:rtype: None
|
||||
|
||||
Example:
|
||||
|
||||
@@ -48,7 +45,5 @@ def edit_task(file: ifcopenshell.file, task: ifcopenshell.entity_instance, attri
|
||||
# Change the identification
|
||||
ifcopenshell.api.sequence.edit_task(model, task=task, attributes={"Identification": "M"})
|
||||
"""
|
||||
settings = {"task": task, "attributes": attributes or {}}
|
||||
|
||||
for name, value in settings["attributes"].items():
|
||||
setattr(settings["task"], name, value)
|
||||
for name, value in attributes.items():
|
||||
setattr(task, name, value)
|
||||
|
||||
@@ -36,11 +36,8 @@ def edit_task_time(
|
||||
IfcTaskTime, consult the IFC documentation.
|
||||
|
||||
:param task_time: The IfcTaskTime entity you want to edit
|
||||
:type task_time: ifcopenshell.entity_instance
|
||||
:param attributes: a dictionary of attribute names and values.
|
||||
:type attributes: dict
|
||||
:return: None
|
||||
:rtype: None
|
||||
|
||||
Example:
|
||||
|
||||
@@ -61,94 +58,89 @@ def edit_task_time(
|
||||
"""
|
||||
usecase = Usecase()
|
||||
usecase.file = file
|
||||
usecase.settings = {"task_time": task_time, "attributes": attributes}
|
||||
return usecase.execute()
|
||||
return usecase.execute(task_time, attributes)
|
||||
|
||||
|
||||
class Usecase:
|
||||
def execute(self):
|
||||
file: ifcopenshell.file
|
||||
|
||||
def execute(self, task_time: ifcopenshell.entity_instance, attributes: dict[str, Any]) -> None:
|
||||
self.task_time = task_time
|
||||
self.task = self.get_task()
|
||||
self.calendar = ifcopenshell.util.sequence.derive_calendar(self.task)
|
||||
|
||||
# If the user specifies both an end date and a duration, the duration takes priority
|
||||
if (
|
||||
self.settings["attributes"].get("ScheduleDuration", None)
|
||||
and "ScheduleFinish" in self.settings["attributes"].keys()
|
||||
):
|
||||
del self.settings["attributes"]["ScheduleFinish"]
|
||||
if attributes.get("ScheduleDuration", None) and "ScheduleFinish" in attributes.keys():
|
||||
del attributes["ScheduleFinish"]
|
||||
|
||||
duration_type = self.settings["attributes"].get("DurationType", self.settings["task_time"].DurationType)
|
||||
finish = self.settings["attributes"].get("ScheduleFinish", None)
|
||||
duration_type = attributes.get("DurationType", self.task_time.DurationType)
|
||||
finish = attributes.get("ScheduleFinish", None)
|
||||
if finish:
|
||||
if isinstance(finish, str):
|
||||
finish = datetime.datetime.fromisoformat(finish)
|
||||
self.settings["attributes"]["ScheduleFinish"] = datetime.datetime.combine(
|
||||
attributes["ScheduleFinish"] = datetime.datetime.combine(
|
||||
ifcopenshell.util.sequence.get_soonest_working_day(finish, duration_type, self.calendar),
|
||||
datetime.time(17),
|
||||
)
|
||||
start = self.settings["attributes"].get("ScheduleStart", None)
|
||||
start = attributes.get("ScheduleStart", None)
|
||||
if start:
|
||||
if isinstance(start, str):
|
||||
start = datetime.datetime.fromisoformat(start)
|
||||
self.settings["attributes"]["ScheduleStart"] = datetime.datetime.combine(
|
||||
attributes["ScheduleStart"] = datetime.datetime.combine(
|
||||
ifcopenshell.util.sequence.get_soonest_working_day(start, duration_type, self.calendar),
|
||||
datetime.time(9),
|
||||
)
|
||||
|
||||
for name, value in self.settings["attributes"].items():
|
||||
for name, value in attributes.items():
|
||||
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":
|
||||
value = ifcopenshell.util.date.datetime2ifc(value, "IfcDuration")
|
||||
setattr(self.settings["task_time"], name, value)
|
||||
setattr(self.task_time, name, value)
|
||||
|
||||
if (
|
||||
"ScheduleDuration" in self.settings["attributes"].keys()
|
||||
and self.settings["task_time"].ScheduleDuration
|
||||
and self.settings["task_time"].ScheduleStart
|
||||
):
|
||||
if "ScheduleDuration" in attributes.keys() and task_time.ScheduleDuration and task_time.ScheduleStart:
|
||||
self.calculate_finish()
|
||||
elif self.settings["attributes"].get("ScheduleStart", None) and self.settings["task_time"].ScheduleDuration:
|
||||
elif attributes.get("ScheduleStart", None) and task_time.ScheduleDuration:
|
||||
self.calculate_finish()
|
||||
elif self.settings["attributes"].get("ScheduleFinish", None) and self.settings["task_time"].ScheduleStart:
|
||||
elif attributes.get("ScheduleFinish", None) and task_time.ScheduleStart:
|
||||
self.calculate_duration()
|
||||
|
||||
if self.settings["task_time"].ScheduleDuration and (
|
||||
"ScheduleStart" in self.settings["attributes"].keys()
|
||||
or "ScheduleFinish" in self.settings["attributes"].keys()
|
||||
or "ScheduleDuration" in self.settings["attributes"].keys()
|
||||
if task_time.ScheduleDuration and (
|
||||
"ScheduleStart" in attributes.keys()
|
||||
or "ScheduleFinish" in attributes.keys()
|
||||
or "ScheduleDuration" in attributes.keys()
|
||||
):
|
||||
ifcopenshell.api.sequence.cascade_schedule(self.file, task=self.task)
|
||||
if self.settings["task_time"].ScheduleDuration:
|
||||
if task_time.ScheduleDuration:
|
||||
self.handle_resource_calculation()
|
||||
|
||||
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),
|
||||
self.settings["task_time"].DurationType,
|
||||
ifcopenshell.util.date.ifc2datetime(self.task_time.ScheduleStart),
|
||||
ifcopenshell.util.date.ifc2datetime(self.task_time.ScheduleDuration),
|
||||
self.task_time.DurationType,
|
||||
self.calendar,
|
||||
date_type="FINISH",
|
||||
)
|
||||
self.settings["task_time"].ScheduleFinish = ifcopenshell.util.date.datetime2ifc(finish, "IfcDateTime")
|
||||
self.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.task_time.ScheduleStart)
|
||||
finish = ifcopenshell.util.date.ifc2datetime(self.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.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.task_time.ScheduleDuration = ifcopenshell.util.date.datetime2ifc(duration, "IfcDuration")
|
||||
|
||||
def get_task(self) -> ifcopenshell.entity_instance:
|
||||
return next(e for e in self.file.get_inverse(self.settings["task_time"]) if e.is_a("IfcTask"))
|
||||
return next(e for e in self.file.get_inverse(self.task_time) if e.is_a("IfcTask"))
|
||||
|
||||
def handle_resource_calculation(self):
|
||||
resources = ifcopenshell.util.sequence.get_task_resources(self.task, is_deep=False)
|
||||
|
||||
@@ -28,11 +28,8 @@ def edit_work_calendar(
|
||||
IfcWorkCalendar, consult the IFC documentation.
|
||||
|
||||
:param work_calendar: The IfcWorkCalendar entity you want to edit
|
||||
:type work_calendar: ifcopenshell.entity_instance
|
||||
:param attributes: a dictionary of attribute names and values.
|
||||
:type attributes: dict
|
||||
:return: None
|
||||
:rtype: None
|
||||
|
||||
Example:
|
||||
|
||||
@@ -45,7 +42,5 @@ def edit_work_calendar(
|
||||
ifcopenshell.api.sequence.edit_work_calendar(model,
|
||||
work_calendar=calendar, attributes={"Description": "Monday to Friday 8 hour days"})
|
||||
"""
|
||||
settings = {"work_calendar": work_calendar, "attributes": attributes}
|
||||
|
||||
for name, value in settings["attributes"].items():
|
||||
setattr(settings["work_calendar"], name, value)
|
||||
for name, value in attributes.items():
|
||||
setattr(work_calendar, name, value)
|
||||
|
||||
@@ -29,11 +29,8 @@ def edit_work_plan(
|
||||
IfcWorkPlan, consult the IFC documentation.
|
||||
|
||||
:param work_plan: The IfcWorkPlan entity you want to edit
|
||||
:type work_plan: ifcopenshell.entity_instance
|
||||
:param attributes: a dictionary of attribute names and values.
|
||||
:type attributes: dict
|
||||
:return: None
|
||||
:rtype: None
|
||||
|
||||
Example:
|
||||
|
||||
@@ -46,12 +43,10 @@ def edit_work_plan(
|
||||
ifcopenshell.api.sequence.edit_work_plan(model,
|
||||
work_plan=work_plan, attributes={"Description": "Construction of phase 1"})
|
||||
"""
|
||||
settings = {"work_plan": work_plan, "attributes": attributes}
|
||||
|
||||
for name, value in settings["attributes"].items():
|
||||
for name, value in attributes.items():
|
||||
if value:
|
||||
if "Date" in name or "Time" in name:
|
||||
value = ifcopenshell.util.date.datetime2ifc(value, "IfcDateTime")
|
||||
elif name == "Duration" or name == "TotalFloat":
|
||||
value = ifcopenshell.util.date.datetime2ifc(value, "IfcDuration")
|
||||
setattr(settings["work_plan"], name, value)
|
||||
setattr(work_plan, name, value)
|
||||
|
||||
@@ -29,11 +29,8 @@ def edit_work_schedule(
|
||||
IfcWorkSchedule, consult the IFC documentation.
|
||||
|
||||
:param work_schedule: The IfcWorkSchedule entity you want to edit
|
||||
:type work_schedule: ifcopenshell.entity_instance
|
||||
:param attributes: a dictionary of attribute names and values.
|
||||
:type attributes: dict
|
||||
:return: None
|
||||
:rtype: None
|
||||
|
||||
Example:
|
||||
|
||||
@@ -50,12 +47,10 @@ def edit_work_schedule(
|
||||
ifcopenshell.api.sequence.edit_work_schedule(model,
|
||||
work_schedule=work_schedule, attributes={"Description": "3 crane design option"})
|
||||
"""
|
||||
settings = {"work_schedule": work_schedule, "attributes": attributes}
|
||||
|
||||
for name, value in settings["attributes"].items():
|
||||
for name, value in attributes.items():
|
||||
if value:
|
||||
if "Date" in name or "Time" in name:
|
||||
value = ifcopenshell.util.date.datetime2ifc(value, "IfcDateTime")
|
||||
elif name == "Duration" or name == "TotalFloat":
|
||||
value = ifcopenshell.util.date.datetime2ifc(value, "IfcDuration")
|
||||
setattr(settings["work_schedule"], name, value)
|
||||
setattr(work_schedule, name, value)
|
||||
|
||||
@@ -31,11 +31,8 @@ def edit_work_time(
|
||||
IfcWorkTime, consult the IFC documentation.
|
||||
|
||||
:param work_time: The IfcWorkTime entity you want to edit
|
||||
:type work_time: ifcopenshell.entity_instance
|
||||
:param attributes: a dictionary of attribute names and values.
|
||||
:type attributes: dict
|
||||
:return: None
|
||||
:rtype: None
|
||||
|
||||
Example:
|
||||
|
||||
@@ -54,16 +51,14 @@ def edit_work_time(
|
||||
ifcopenshell.api.sequence.edit_work_time(model,
|
||||
work_time=work_time, attributes={"StartDate": "2000-01-01", "FinishDate": "2000-01-02"})
|
||||
"""
|
||||
settings = {"work_time": work_time, "attributes": attributes}
|
||||
|
||||
for name, value in settings["attributes"].items():
|
||||
for name, value in attributes.items():
|
||||
if name in ("Start", "StartDate"):
|
||||
value = ifcopenshell.util.date.datetime2ifc(value, "IfcDate")
|
||||
# 4 IfcWorktime Start
|
||||
settings["work_time"][4] = value
|
||||
work_time[4] = value
|
||||
elif name in ("Finish", "FinishDate"):
|
||||
value = ifcopenshell.util.date.datetime2ifc(value, "IfcDate")
|
||||
# 5 IfcWorktime Finish
|
||||
settings["work_time"][5] = value
|
||||
work_time[5] = value
|
||||
else:
|
||||
setattr(settings["work_time"], name, value)
|
||||
setattr(work_time, name, value)
|
||||
|
||||
@@ -35,9 +35,7 @@ def recalculate_schedule(file: ifcopenshell.file, work_schedule: ifcopenshell.en
|
||||
error.
|
||||
|
||||
:param work_schedule: The IfcWorkSchedule to perform the calculation on.
|
||||
:type work_schedule: ifcopenshell.entity_instance
|
||||
:return: None
|
||||
:rtype: None
|
||||
|
||||
Example:
|
||||
|
||||
@@ -50,12 +48,14 @@ def recalculate_schedule(file: ifcopenshell.file, work_schedule: ifcopenshell.en
|
||||
"""
|
||||
usecase = Usecase()
|
||||
usecase.file = file
|
||||
usecase.settings = {"work_schedule": work_schedule}
|
||||
return usecase.execute()
|
||||
return usecase.execute(work_schedule)
|
||||
|
||||
|
||||
class Usecase:
|
||||
def execute(self):
|
||||
file: ifcopenshell.file
|
||||
|
||||
def execute(self, work_schedule: ifcopenshell.entity_instance) -> None:
|
||||
self.work_schedule = work_schedule
|
||||
# The method implemented is the same as shown here:
|
||||
# https://www.youtube.com/watch?v=qTErIV6OqLg
|
||||
self.start_dates = []
|
||||
@@ -88,7 +88,6 @@ class Usecase:
|
||||
|
||||
if is_cyclic:
|
||||
raise RecursionError("Task graph is cyclic and so critical path method cannot be performed.")
|
||||
return
|
||||
|
||||
self.pending_nodes = set(self.g.nodes)
|
||||
while self.pending_nodes:
|
||||
@@ -100,7 +99,7 @@ class Usecase:
|
||||
|
||||
self.update_task_times()
|
||||
|
||||
def build_network_graph(self):
|
||||
def build_network_graph(self) -> None:
|
||||
self.sequence_type_map = {
|
||||
None: "FS",
|
||||
"START_START": "SS",
|
||||
@@ -114,14 +113,14 @@ class Usecase:
|
||||
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)
|
||||
for rel in self.settings["work_schedule"].Controls:
|
||||
for rel in self.work_schedule.Controls:
|
||||
for related_object in rel.RelatedObjects:
|
||||
if not related_object.is_a("IfcTask"):
|
||||
continue
|
||||
self.add_node(related_object)
|
||||
self.g.add_edges_from(self.edges)
|
||||
|
||||
def add_node(self, task):
|
||||
def add_node(self, task: ifcopenshell.entity_instance) -> None:
|
||||
if task.IsNestedBy:
|
||||
for rel in task.IsNestedBy:
|
||||
[self.add_node(o) for o in rel.RelatedObjects]
|
||||
@@ -176,7 +175,7 @@ class Usecase:
|
||||
if not successor_types:
|
||||
self.edges.append((task.id(), "finish", {"lag_time": 0, "type": "FF"}))
|
||||
|
||||
def update_task_times(self):
|
||||
def update_task_times(self) -> None:
|
||||
for ifc_definition_id in self.g.nodes:
|
||||
if ifc_definition_id in ("start", "finish"):
|
||||
continue
|
||||
@@ -198,12 +197,12 @@ class Usecase:
|
||||
},
|
||||
)
|
||||
|
||||
def offset_date(self, date, days, node):
|
||||
def offset_date(self, date: datetime.datetime, days: int, node: dict) -> datetime.datetime:
|
||||
return ifcopenshell.util.sequence.offset_date(
|
||||
date, datetime.timedelta(days=days), node["duration_type"], node["calendar"]
|
||||
)
|
||||
|
||||
def forward_pass(self, node):
|
||||
def forward_pass(self, node) -> bool:
|
||||
successors = self.g.successors(node)
|
||||
predecessors = list(self.g.predecessors(node))
|
||||
data = self.g.nodes[node]
|
||||
@@ -326,7 +325,7 @@ class Usecase:
|
||||
|
||||
return True
|
||||
|
||||
def backward_pass(self, node):
|
||||
def backward_pass(self, node) -> bool:
|
||||
successors = list(self.g.successors(node))
|
||||
predecessors = self.g.predecessors(node)
|
||||
data = self.g.nodes[node]
|
||||
@@ -496,12 +495,12 @@ class Usecase:
|
||||
|
||||
def calculate_free_float(
|
||||
self,
|
||||
predecessor_date,
|
||||
successor_date,
|
||||
lag_time,
|
||||
predecessor_data,
|
||||
successor_data,
|
||||
):
|
||||
predecessor_date: datetime.datetime,
|
||||
successor_date: datetime.datetime,
|
||||
lag_time: int,
|
||||
predecessor_data: dict,
|
||||
successor_data: dict,
|
||||
) -> datetime.timedelta:
|
||||
if not lag_time:
|
||||
min_successor_date = successor_date
|
||||
else:
|
||||
|
||||
Reference in New Issue
Block a user