mirror of
https://github.com/IfcOpenShell/IfcOpenShell.git
synced 2026-08-09 09:21:46 +00:00
IFC4D code refactoring (#1834)
* P6XML Parser failing on calendar type * add work plan references a get_person which is not an attribute. modified to get_user to work * Initial support for Primavera XER files * IFC4D impl XER calendars exceptions and work time * merged updats * refactoring code for the 4D model to avoid duplicate code * Assign process in the api module was missing an atrribute QuantityIn Process * Merged changes from master * refactoring * resolved conflicts * refactoring code for IFC4D * refactoring code for IFC4D * refactoring code for IFC4D * Reverted the changes to IfcRelAssignsToProcess usecase * Changed the class to be more discriptive from Utils to ScheduleIfcGenerator * reverted changes to add_work_plan and Makefile Co-authored-by: Dion Moult <dion@thinkmoult.com>
This commit is contained in:
HassanEmam
@@ -0,0 +1,287 @@
|
||||
from xerparser.reader import Reader
|
||||
import ifcopenshell
|
||||
import ifcopenshell.api
|
||||
import ifcopenshell.util.date
|
||||
from datetime import datetime, timedelta, date
|
||||
|
||||
|
||||
class ScheduleIfcGenerator:
|
||||
|
||||
def __init__(self, file, work_plan, project, calendars, wbs, root_activites, activities, relationships):
|
||||
self.file = file
|
||||
self.work_plan = work_plan
|
||||
self.project = project
|
||||
self.calendars = calendars
|
||||
self.wbs = wbs
|
||||
self.root_activites = root_activites
|
||||
self.activities = activities
|
||||
self.relationships = relationships
|
||||
self.day_map = {
|
||||
"Monday": 1,
|
||||
"Tuesday": 2,
|
||||
"Wednesday": 3,
|
||||
"Thursday": 4,
|
||||
"Friday": 5,
|
||||
"Saturday": 6,
|
||||
"Sunday": 7,
|
||||
}
|
||||
|
||||
def create_ifc(self):
|
||||
if not self.file:
|
||||
self.file = self.create_boilerplate_ifc()
|
||||
if not self.work_plan:
|
||||
self.work_plan = ifcopenshell.api.run("sequence.add_work_plan", self.file)
|
||||
work_schedule = self.create_work_schedule()
|
||||
self.create_calendars()
|
||||
self.create_tasks(work_schedule)
|
||||
self.create_rel_sequences()
|
||||
|
||||
def create_work_schedule(self):
|
||||
return ifcopenshell.api.run(
|
||||
"sequence.add_work_schedule", self.file, name=self.project["Name"], work_plan=self.work_plan
|
||||
)
|
||||
|
||||
def create_calendars(self):
|
||||
for calendar in self.calendars.values():
|
||||
calendar["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_work_calendar", self.file, name=calendar["Name"]
|
||||
)
|
||||
self.process_working_week(calendar["StandardWorkWeek"], calendar["ifc"])
|
||||
self.process_exceptions(calendar["HolidayOrExceptions"], calendar["ifc"])
|
||||
|
||||
def process_working_week(self, week, calendar):
|
||||
for day in week:
|
||||
if day["ifc"] or not day["WorkTimes"]:
|
||||
continue
|
||||
|
||||
day["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_work_time", self.file, work_calendar=calendar, time_type="WorkingTimes"
|
||||
)
|
||||
weekday_component = [self.day_map[day["DayOfWeek"]]]
|
||||
for day2 in week:
|
||||
if day["DayOfWeek"] == day2["DayOfWeek"]:
|
||||
continue
|
||||
if day["WorkTimes"] == day2["WorkTimes"]:
|
||||
weekday_component.append(self.day_map[day2["DayOfWeek"]])
|
||||
# Don't process the next day, as we can group it
|
||||
day2["ifc"] = day["ifc"]
|
||||
|
||||
work_time_name = "Weekdays: {}".format(", ".join([str(c) for c in sorted(weekday_component)]))
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_work_time",
|
||||
self.file,
|
||||
work_time=day["ifc"],
|
||||
attributes={"Name": work_time_name},
|
||||
)
|
||||
|
||||
recurrence = ifcopenshell.api.run(
|
||||
"sequence.assign_recurrence_pattern", self.file, parent=day["ifc"], recurrence_type="WEEKLY"
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_recurrence_pattern",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
attributes={"WeekdayComponent": weekday_component},
|
||||
)
|
||||
for work_time in day["WorkTimes"]:
|
||||
ifcopenshell.api.run(
|
||||
"sequence.add_time_period",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
start_time=work_time["Start"],
|
||||
end_time=work_time["Finish"],
|
||||
)
|
||||
|
||||
def process_exceptions(self, exceptions, calendar):
|
||||
for year, year_data in exceptions.items():
|
||||
for month, month_data in year_data.items():
|
||||
if month_data["FullDay"]:
|
||||
self.process_full_day_exceptions(year, month, month_data, calendar)
|
||||
if month_data["WorkTime"]:
|
||||
self.process_work_time_exceptions(year, month, month_data, calendar)
|
||||
|
||||
def process_full_day_exceptions(self, year, month, month_data, calendar):
|
||||
work_time = ifcopenshell.api.run(
|
||||
"sequence.add_work_time", self.file, work_calendar=calendar, time_type="ExceptionTimes"
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_work_time",
|
||||
self.file,
|
||||
work_time=work_time,
|
||||
attributes={
|
||||
"Name": f"{year}-{month}",
|
||||
"Start": date(year, 1, 1),
|
||||
"Finish": date(year, 12, 31),
|
||||
},
|
||||
)
|
||||
recurrence = ifcopenshell.api.run(
|
||||
"sequence.assign_recurrence_pattern",
|
||||
self.file,
|
||||
parent=work_time,
|
||||
recurrence_type="YEARLY_BY_DAY_OF_MONTH",
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_recurrence_pattern",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
attributes={"DayComponent": month_data["FullDay"], "MonthComponent": [month]},
|
||||
)
|
||||
|
||||
def process_work_time_exceptions(self, year, month, month_data, calendar):
|
||||
for day in month_data["WorkTime"]:
|
||||
if day["ifc"]:
|
||||
continue
|
||||
|
||||
day["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_work_time", self.file, work_calendar=calendar, time_type="ExceptionTimes"
|
||||
)
|
||||
|
||||
day_component = [day["Day"]]
|
||||
for day2 in month_data["WorkTime"]:
|
||||
if day["Day"] == day2["Day"]:
|
||||
continue
|
||||
if day["WorkTimes"] == day2["WorkTimes"]:
|
||||
day_component.append(day2["Day"])
|
||||
# Don't process the next day, as we can group it
|
||||
day2["ifc"] = day["ifc"]
|
||||
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_work_time",
|
||||
self.file,
|
||||
work_time=day["ifc"],
|
||||
attributes={
|
||||
"Name": "{}-{}-{}".format(year, month, ", ".join([str(d) for d in day_component])),
|
||||
"Start": date(year, 1, 1),
|
||||
"Finish": date(year, 12, 31),
|
||||
},
|
||||
)
|
||||
recurrence = ifcopenshell.api.run(
|
||||
"sequence.assign_recurrence_pattern",
|
||||
self.file,
|
||||
parent=day["ifc"],
|
||||
recurrence_type="YEARLY_BY_DAY_OF_MONTH",
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_recurrence_pattern",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
attributes={"DayComponent": day_component, "MonthComponent": [month]},
|
||||
)
|
||||
for work_time in day["WorkTimes"]:
|
||||
ifcopenshell.api.run(
|
||||
"sequence.add_time_period",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
start_time=work_time["Start"],
|
||||
end_time=work_time["Finish"],
|
||||
)
|
||||
|
||||
def create_tasks(self, work_schedule):
|
||||
for wbs in self.wbs.values():
|
||||
self.create_task_from_wbs(wbs, work_schedule)
|
||||
for activity_id in self.root_activites:
|
||||
self.create_task_from_activity(self.activities[activity_id], None, work_schedule)
|
||||
|
||||
def create_task_from_wbs(self, wbs, work_schedule):
|
||||
if not self.wbs.get(wbs["ParentObjectId"]):
|
||||
wbs["ParentObjectId"] = None
|
||||
wbs["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_task",
|
||||
self.file,
|
||||
work_schedule=None if wbs["ParentObjectId"] else work_schedule,
|
||||
parent_task=self.wbs[wbs["ParentObjectId"]]["ifc"] if wbs["ParentObjectId"] else None,
|
||||
)
|
||||
identification = wbs["Code"]
|
||||
if wbs["ParentObjectId"]:
|
||||
identification = self.wbs[wbs["ParentObjectId"]]["ifc"].Identification + "." + wbs["Code"]
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_task",
|
||||
self.file,
|
||||
task=wbs["ifc"],
|
||||
attributes={"Name": wbs["Name"], "Identification": identification},
|
||||
)
|
||||
for activity_id in wbs["activities"]:
|
||||
self.create_task_from_activity(self.activities[activity_id], wbs, None)
|
||||
|
||||
def create_task_from_activity(self, activity, wbs, work_schedule):
|
||||
activity["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_task",
|
||||
self.file,
|
||||
work_schedule=None if wbs else work_schedule,
|
||||
parent_task=wbs["ifc"] if wbs else None,
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_task",
|
||||
self.file,
|
||||
task=activity["ifc"],
|
||||
attributes={
|
||||
"Name": activity["Name"],
|
||||
"Identification": activity["Identification"],
|
||||
"Status": activity["Status"],
|
||||
"IsMilestone": activity["StartDate"] == activity["FinishDate"],
|
||||
"PredefinedType": "CONSTRUCTION"
|
||||
},
|
||||
)
|
||||
task_time = ifcopenshell.api.run("sequence.add_task_time", self.file, task=activity["ifc"])
|
||||
calendar = self.calendars[activity["CalendarObjectId"]]
|
||||
#print(calendar, self.calendars)
|
||||
# Seems intermittently crashy - can we investigate for larger files?
|
||||
ifcopenshell.api.run(
|
||||
"control.assign_control",
|
||||
self.file,
|
||||
**{
|
||||
"relating_control": calendar["ifc"],
|
||||
"related_object": activity["ifc"],
|
||||
},
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_task_time",
|
||||
self.file,
|
||||
task_time=task_time,
|
||||
attributes={
|
||||
"ScheduleStart": activity["StartDate"],
|
||||
"ScheduleFinish": activity["FinishDate"],
|
||||
"DurationType": "WORKTIME" if activity["PlannedDuration"] else None,
|
||||
"ScheduleDuration": timedelta(
|
||||
days=float(activity["PlannedDuration"]) / float(calendar["HoursPerDay"])
|
||||
)
|
||||
or None
|
||||
if activity["PlannedDuration"]
|
||||
else None,
|
||||
},
|
||||
)
|
||||
|
||||
def create_rel_sequences(self):
|
||||
self.sequence_type_map = {
|
||||
"Start to Start": "START_START",
|
||||
"Start to Finish": "START_FINISH",
|
||||
"Finish to Start": "FINISH_START",
|
||||
"Finish to Finish": "FINISH_FINISH",
|
||||
}
|
||||
for relationship in self.relationships.values():
|
||||
rel_sequence = ifcopenshell.api.run(
|
||||
"sequence.assign_sequence",
|
||||
self.file,
|
||||
relating_process=self.activities[relationship["PredecessorActivity"]]["ifc"],
|
||||
related_process=self.activities[relationship["SuccessorActivity"]]["ifc"],
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_sequence",
|
||||
self.file,
|
||||
rel_sequence=rel_sequence,
|
||||
attributes={"SequenceType": relationship["Type"]},
|
||||
)
|
||||
lag = float(relationship["Lag"])
|
||||
if lag:
|
||||
calendar = self.calendars[self.activities[relationship["PredecessorActivity"]]["CalendarObjectId"]]
|
||||
ifcopenshell.api.run(
|
||||
"sequence.assign_lag_time",
|
||||
self.file,
|
||||
rel_sequence=rel_sequence,
|
||||
lag_value=datetime.timedelta(days=lag / float(calendar["HoursPerDay"])),
|
||||
duration_type="WORKTIME",
|
||||
)
|
||||
|
||||
def create_boilerplate_ifc(self):
|
||||
self.file = ifcopenshell.file(schema="IFC4")
|
||||
self.work_plan = self.file.create_entity("IfcWorkPlan")
|
||||
+6
-134
@@ -23,6 +23,7 @@ import ifcopenshell
|
||||
import ifcopenshell.api
|
||||
import ifcopenshell.util.date
|
||||
import xml.etree.ElementTree as ET
|
||||
from common import ScheduleIfcGenerator
|
||||
|
||||
|
||||
class MSP2Ifc:
|
||||
@@ -40,7 +41,10 @@ class MSP2Ifc:
|
||||
|
||||
def execute(self):
|
||||
self.parse_xml()
|
||||
self.create_ifc()
|
||||
ifcCreator = ScheduleIfcGenerator(self.file, self.work_plan, self.project, self.calendars,
|
||||
self.wbs, self.root_activites, self.activities, self.relationships)
|
||||
ifcCreator.create_ifc()
|
||||
#self.create_ifc()
|
||||
|
||||
def parse_xml(self):
|
||||
tree = ET.parse(self.xml)
|
||||
@@ -119,136 +123,4 @@ class MSP2Ifc:
|
||||
self.calendars[calendar_id] = {
|
||||
"Name": calendar.find("pr:Name", self.ns).text,
|
||||
"StandardWorkWeek": week_days,
|
||||
}
|
||||
|
||||
def create_ifc(self):
|
||||
if not self.file:
|
||||
self.create_boilerplate_ifc()
|
||||
if not self.work_plan:
|
||||
self.work_plan = ifcopenshell.api.run("sequence.add_work_plan", self.file)
|
||||
work_schedule = self.create_work_schedule()
|
||||
self.create_tasks(work_schedule)
|
||||
self.create_calendars()
|
||||
self.create_rel_sequences()
|
||||
|
||||
def create_boilerplate_ifc(self):
|
||||
self.file = ifcopenshell.file(schema="IFC4")
|
||||
self.work_plan = self.file.create_entity("IfcWorkPlan")
|
||||
|
||||
def create_tasks(self, work_schedule):
|
||||
for task_id in self.tasks:
|
||||
task = self.tasks[task_id]
|
||||
if task["OutlineLevel"] == 0:
|
||||
self.create_task(task, work_schedule=work_schedule)
|
||||
|
||||
def create_work_schedule(self):
|
||||
return ifcopenshell.api.run(
|
||||
"sequence.add_work_schedule", self.file, name=self.project["Name"], work_plan=self.work_plan
|
||||
)
|
||||
|
||||
def create_calendars(self):
|
||||
for calendar in self.calendars.values():
|
||||
calendar["ifc"] = ifcopenshell.api.run("sequence.add_work_calendar", self.file, name=calendar["Name"])
|
||||
self.process_working_week(calendar["StandardWorkWeek"], calendar["ifc"])
|
||||
|
||||
def create_task(self, task, work_schedule=None, parent_task=None):
|
||||
task["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_task",
|
||||
self.file,
|
||||
work_schedule=work_schedule if work_schedule else None,
|
||||
parent_task=parent_task["ifc"] if parent_task else None,
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_task",
|
||||
self.file,
|
||||
task=task["ifc"],
|
||||
attributes={
|
||||
"Name": task["Name"],
|
||||
"Identification": task["OutlineNumber"],
|
||||
"IsMilestone": task["Start"] == task["Finish"],
|
||||
},
|
||||
)
|
||||
task_time = ifcopenshell.api.run("sequence.add_task_time", self.file, task=task["ifc"])
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_task_time",
|
||||
self.file,
|
||||
task_time=task_time,
|
||||
attributes={
|
||||
"ScheduleStart": task["Start"],
|
||||
"ScheduleFinish": task["Finish"],
|
||||
"DurationType": "WORKTIME" if task["Duration"] else None,
|
||||
"ScheduleDuration": task["Duration"] if task["Duration"] else None,
|
||||
},
|
||||
)
|
||||
for subtask_id in task["subtasks"]:
|
||||
self.create_task(self.tasks[subtask_id], parent_task=task)
|
||||
|
||||
def process_working_week(self, week, calendar):
|
||||
for day in week:
|
||||
if day["ifc"]:
|
||||
continue
|
||||
|
||||
day["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_work_time", self.file, work_calendar=calendar, time_type="WorkingTimes"
|
||||
)
|
||||
|
||||
weekday_component = [int(day["DayType"])]
|
||||
for day2 in week:
|
||||
if day["DayType"] == day2["DayType"]:
|
||||
continue
|
||||
if day["WorkingTimes"] == day2["WorkingTimes"]:
|
||||
weekday_component.append(int(day2["DayType"]))
|
||||
# Don't process the next day, as we can group it
|
||||
day2["ifc"] = day["ifc"]
|
||||
|
||||
work_time_name = "Weekdays: {}".format(", ".join([str(c) for c in sorted(weekday_component)]))
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_work_time",
|
||||
self.file,
|
||||
work_time=day["ifc"],
|
||||
attributes={"Name": work_time_name},
|
||||
)
|
||||
|
||||
recurrence = ifcopenshell.api.run(
|
||||
"sequence.assign_recurrence_pattern", self.file, parent=day["ifc"], recurrence_type="WEEKLY"
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_recurrence_pattern",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
attributes={"WeekdayComponent": weekday_component},
|
||||
)
|
||||
for work_time in day["WorkingTimes"]:
|
||||
ifcopenshell.api.run(
|
||||
"sequence.add_time_period",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
start_time=work_time["Start"],
|
||||
end_time=work_time["Finish"],
|
||||
)
|
||||
|
||||
def create_rel_sequences(self):
|
||||
self.sequence_type_map = {
|
||||
"1": "START_START",
|
||||
"2": "START_FINISH",
|
||||
"3": "FINISH_START",
|
||||
"4": "FINISH_FINISH",
|
||||
"0": "NOTDEFINED",
|
||||
}
|
||||
for task in self.tasks.values():
|
||||
if not task["PredecessorTasks"]:
|
||||
continue
|
||||
for predecessor in task["PredecessorTasks"].values():
|
||||
rel_sequence = ifcopenshell.api.run(
|
||||
"sequence.assign_sequence",
|
||||
self.file,
|
||||
related_process=task["ifc"],
|
||||
relating_process=self.tasks[predecessor["PredecessorTask"]]["ifc"],
|
||||
)
|
||||
if predecessor["Type"]:
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_sequence",
|
||||
self.file,
|
||||
rel_sequence=rel_sequence,
|
||||
attributes={"SequenceType": self.sequence_type_map[predecessor["Type"]]},
|
||||
)
|
||||
}
|
||||
+4
-259
@@ -23,7 +23,7 @@ import ifcopenshell
|
||||
import ifcopenshell.api
|
||||
import ifcopenshell.util.date
|
||||
import xml.etree.ElementTree as ET
|
||||
|
||||
from common import ScheduleIfcGenerator
|
||||
|
||||
class P62Ifc:
|
||||
def __init__(self):
|
||||
@@ -48,7 +48,9 @@ class P62Ifc:
|
||||
|
||||
def execute(self):
|
||||
self.parse_xml()
|
||||
self.create_ifc()
|
||||
ifcCreator = ScheduleIfcGenerator(self.file, self.work_plan, self.project, self.calendars,
|
||||
self.wbs, self.root_activites, self.activities, self.relationships)
|
||||
ifcCreator.create_ifc()
|
||||
|
||||
def parse_xml(self):
|
||||
tree = ET.parse(self.xml)
|
||||
@@ -166,260 +168,3 @@ class P62Ifc:
|
||||
|
||||
def get_wbs(self, wbs):
|
||||
return {"Name": wbs.find("pr:Name", self.ns).text, "subtasks": []}
|
||||
|
||||
def create_ifc(self):
|
||||
if not self.file:
|
||||
self.file = self.create_boilerplate_ifc()
|
||||
if not self.work_plan:
|
||||
self.work_plan = ifcopenshell.api.run("sequence.add_work_plan", self.file)
|
||||
work_schedule = self.create_work_schedule()
|
||||
self.create_calendars()
|
||||
self.create_tasks(work_schedule)
|
||||
self.create_rel_sequences()
|
||||
|
||||
def create_work_schedule(self):
|
||||
return ifcopenshell.api.run(
|
||||
"sequence.add_work_schedule", self.file, name=self.project["Name"], work_plan=self.work_plan
|
||||
)
|
||||
|
||||
def create_calendars(self):
|
||||
for calendar in self.calendars.values():
|
||||
calendar["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_work_calendar", self.file, name=calendar["Name"])
|
||||
self.process_working_week(calendar["StandardWorkWeek"], calendar["ifc"])
|
||||
self.process_exceptions(calendar["HolidayOrExceptions"], calendar["ifc"])
|
||||
|
||||
def process_working_week(self, week, calendar):
|
||||
for day in week:
|
||||
if day["ifc"] or not day["WorkTimes"]:
|
||||
continue
|
||||
|
||||
day["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_work_time", self.file, work_calendar=calendar, time_type="WorkingTimes"
|
||||
)
|
||||
|
||||
weekday_component = [self.day_map[day["DayOfWeek"]]]
|
||||
for day2 in week:
|
||||
if day["DayOfWeek"] == day2["DayOfWeek"]:
|
||||
continue
|
||||
if day["WorkTimes"] == day2["WorkTimes"]:
|
||||
weekday_component.append(self.day_map[day2["DayOfWeek"]])
|
||||
# Don't process the next day, as we can group it
|
||||
day2["ifc"] = day["ifc"]
|
||||
|
||||
work_time_name = "Weekdays: {}".format(", ".join([str(c) for c in sorted(weekday_component)]))
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_work_time",
|
||||
self.file,
|
||||
work_time=day["ifc"],
|
||||
attributes={"Name": work_time_name},
|
||||
)
|
||||
|
||||
recurrence = ifcopenshell.api.run(
|
||||
"sequence.assign_recurrence_pattern", self.file, parent=day["ifc"], recurrence_type="WEEKLY"
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_recurrence_pattern",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
attributes={"WeekdayComponent": weekday_component},
|
||||
)
|
||||
for work_time in day["WorkTimes"]:
|
||||
ifcopenshell.api.run(
|
||||
"sequence.add_time_period",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
start_time=work_time["Start"],
|
||||
end_time=work_time["Finish"],
|
||||
)
|
||||
|
||||
def process_exceptions(self, exceptions, calendar):
|
||||
for year, year_data in exceptions.items():
|
||||
for month, month_data in year_data.items():
|
||||
if month_data["FullDay"]:
|
||||
self.process_full_day_exceptions(year, month, month_data, calendar)
|
||||
if month_data["WorkTime"]:
|
||||
self.process_work_time_exceptions(year, month, month_data, calendar)
|
||||
|
||||
def process_full_day_exceptions(self, year, month, month_data, calendar):
|
||||
work_time = ifcopenshell.api.run(
|
||||
"sequence.add_work_time", self.file, work_calendar=calendar, time_type="ExceptionTimes"
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_work_time",
|
||||
self.file,
|
||||
work_time=work_time,
|
||||
attributes={
|
||||
"Name": f"{year}-{month}",
|
||||
"Start": datetime.date(year, 1, 1),
|
||||
"Finish": datetime.date(year, 12, 31),
|
||||
},
|
||||
)
|
||||
recurrence = ifcopenshell.api.run(
|
||||
"sequence.assign_recurrence_pattern",
|
||||
self.file,
|
||||
parent=work_time,
|
||||
recurrence_type="YEARLY_BY_DAY_OF_MONTH",
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_recurrence_pattern",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
attributes={"DayComponent": month_data["FullDay"], "MonthComponent": [month]},
|
||||
)
|
||||
|
||||
def process_work_time_exceptions(self, year, month, month_data, calendar):
|
||||
for day in month_data["WorkTime"]:
|
||||
if day["ifc"]:
|
||||
continue
|
||||
|
||||
day["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_work_time", self.file, work_calendar=calendar, time_type="ExceptionTimes"
|
||||
)
|
||||
|
||||
day_component = [day["Day"]]
|
||||
for day2 in month_data["WorkTime"]:
|
||||
if day["Day"] == day2["Day"]:
|
||||
continue
|
||||
if day["WorkTimes"] == day2["WorkTimes"]:
|
||||
day_component.append(day2["Day"])
|
||||
# Don't process the next day, as we can group it
|
||||
day2["ifc"] = day["ifc"]
|
||||
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_work_time",
|
||||
self.file,
|
||||
work_time=day["ifc"],
|
||||
attributes={
|
||||
"Name": "{}-{}-{}".format(year, month, ", ".join([str(d) for d in day_component])),
|
||||
"Start": datetime.date(year, 1, 1),
|
||||
"Finish": datetime.date(year, 12, 31),
|
||||
},
|
||||
)
|
||||
recurrence = ifcopenshell.api.run(
|
||||
"sequence.assign_recurrence_pattern",
|
||||
self.file,
|
||||
parent=day["ifc"],
|
||||
recurrence_type="YEARLY_BY_DAY_OF_MONTH",
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_recurrence_pattern",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
attributes={"DayComponent": day_component, "MonthComponent": [month]},
|
||||
)
|
||||
for work_time in day["WorkTimes"]:
|
||||
ifcopenshell.api.run(
|
||||
"sequence.add_time_period",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
start_time=work_time["Start"],
|
||||
end_time=work_time["Finish"],
|
||||
)
|
||||
|
||||
def create_tasks(self, work_schedule):
|
||||
for wbs in self.wbs.values():
|
||||
self.create_task_from_wbs(wbs, work_schedule)
|
||||
for activity_id in self.root_activites:
|
||||
self.create_task_from_activity(self.activities[activity_id], None, work_schedule)
|
||||
|
||||
def create_task_from_wbs(self, wbs, work_schedule):
|
||||
wbs["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_task",
|
||||
self.file,
|
||||
work_schedule=None if wbs["ParentObjectId"] else work_schedule,
|
||||
parent_task=self.wbs[wbs["ParentObjectId"]]["ifc"] if wbs["ParentObjectId"] else None,
|
||||
)
|
||||
identification = wbs["Code"]
|
||||
if wbs["ParentObjectId"]:
|
||||
identification = self.wbs[wbs["ParentObjectId"]]["ifc"].Identification + "." + wbs["Code"]
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_task",
|
||||
self.file,
|
||||
task=wbs["ifc"],
|
||||
attributes={"Name": wbs["Name"], "Identification": identification},
|
||||
)
|
||||
for activity_id in wbs["activities"]:
|
||||
self.create_task_from_activity(self.activities[activity_id], wbs, None)
|
||||
|
||||
def create_task_from_activity(self, activity, wbs, work_schedule):
|
||||
activity["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_task",
|
||||
self.file,
|
||||
work_schedule=None if wbs else work_schedule,
|
||||
parent_task=wbs["ifc"] if wbs else None,
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_task",
|
||||
self.file,
|
||||
task=activity["ifc"],
|
||||
attributes={
|
||||
"Name": activity["Name"],
|
||||
"Identification": activity["Identification"],
|
||||
"Status": activity["Status"],
|
||||
"IsMilestone": activity["StartDate"] == activity["FinishDate"],
|
||||
"PredefinedType": "CONSTRUCTION"
|
||||
},
|
||||
)
|
||||
task_time = ifcopenshell.api.run("sequence.add_task_time", self.file, task=activity["ifc"])
|
||||
calendar = self.calendars[activity["CalendarObjectId"]]
|
||||
# Seems intermittently crashy - can we investigate for larger files?
|
||||
ifcopenshell.api.run(
|
||||
"control.assign_control",
|
||||
self.file,
|
||||
**{
|
||||
"relating_control": calendar["ifc"],
|
||||
"related_object": activity["ifc"],
|
||||
},
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_task_time",
|
||||
self.file,
|
||||
task_time=task_time,
|
||||
attributes={
|
||||
"ScheduleStart": activity["StartDate"],
|
||||
"ScheduleFinish": activity["FinishDate"],
|
||||
"DurationType": "WORKTIME" if activity["PlannedDuration"] else None,
|
||||
"ScheduleDuration": datetime.timedelta(
|
||||
days=float(activity["PlannedDuration"]) / float(calendar["HoursPerDay"])
|
||||
)
|
||||
or None
|
||||
if activity["PlannedDuration"]
|
||||
else None,
|
||||
},
|
||||
)
|
||||
|
||||
def create_rel_sequences(self):
|
||||
self.sequence_type_map = {
|
||||
"Start to Start": "START_START",
|
||||
"Start to Finish": "START_FINISH",
|
||||
"Finish to Start": "FINISH_START",
|
||||
"Finish to Finish": "FINISH_FINISH",
|
||||
}
|
||||
for relationship in self.relationships.values():
|
||||
rel_sequence = ifcopenshell.api.run(
|
||||
"sequence.assign_sequence",
|
||||
self.file,
|
||||
relating_process=self.activities[relationship["PredecessorActivity"]]["ifc"],
|
||||
related_process=self.activities[relationship["SuccessorActivity"]]["ifc"],
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_sequence",
|
||||
self.file,
|
||||
rel_sequence=rel_sequence,
|
||||
attributes={"SequenceType": self.sequence_type_map[relationship["Type"]]},
|
||||
)
|
||||
lag = float(relationship["Lag"])
|
||||
if lag:
|
||||
calendar = self.calendars[self.activities[relationship["PredecessorActivity"]]["CalendarObjectId"]]
|
||||
ifcopenshell.api.run(
|
||||
"sequence.assign_lag_time",
|
||||
self.file,
|
||||
rel_sequence=rel_sequence,
|
||||
lag_value=datetime.timedelta(days=lag / float(calendar["HoursPerDay"])),
|
||||
duration_type="WORKTIME",
|
||||
)
|
||||
|
||||
def create_boilerplate_ifc(self):
|
||||
self.file = ifcopenshell.file(schema="IFC4")
|
||||
self.work_plan = self.file.create_entity("IfcWorkPlan")
|
||||
|
||||
@@ -21,6 +21,7 @@ import ifcopenshell
|
||||
import ifcopenshell.api
|
||||
import ifcopenshell.util.date
|
||||
from datetime import datetime, timedelta, date
|
||||
from common import ScheduleIfcGenerator
|
||||
|
||||
|
||||
class P6XER2Ifc():
|
||||
@@ -57,15 +58,7 @@ class P6XER2Ifc():
|
||||
self.root_activites = []
|
||||
self.activities = {}
|
||||
self.relationships = {}
|
||||
self.day_map = {
|
||||
"Monday": 1,
|
||||
"Tuesday": 2,
|
||||
"Wednesday": 3,
|
||||
"Thursday": 4,
|
||||
"Friday": 5,
|
||||
"Saturday": 6,
|
||||
"Sunday": 7,
|
||||
}
|
||||
|
||||
|
||||
self.day_map2 = {
|
||||
'1': "Monday",
|
||||
@@ -80,7 +73,10 @@ class P6XER2Ifc():
|
||||
|
||||
def execute(self):
|
||||
self.parse_xer()
|
||||
self.create_ifc()
|
||||
ifcCreator = ScheduleIfcGenerator(self.file, self.work_plan, self.project, self.calendars,
|
||||
self.wbs, self.root_activites, self.activities, self.relationships)
|
||||
ifcCreator.create_ifc()
|
||||
# self.create_ifc()
|
||||
|
||||
|
||||
def parse_xer(self):
|
||||
@@ -164,266 +160,7 @@ class P6XER2Ifc():
|
||||
}
|
||||
|
||||
|
||||
def create_ifc(self):
|
||||
if not self.file:
|
||||
self.file = self.create_boilerplate_ifc()
|
||||
if not self.work_plan:
|
||||
self.work_plan = ifcopenshell.api.run("sequence.add_work_plan", self.file)
|
||||
work_schedule = self.create_work_schedule()
|
||||
self.create_calendars()
|
||||
self.create_tasks(work_schedule)
|
||||
self.create_rel_sequences()
|
||||
|
||||
def create_work_schedule(self):
|
||||
return ifcopenshell.api.run(
|
||||
"sequence.add_work_schedule", self.file, name=self.project["Name"], work_plan=self.work_plan
|
||||
)
|
||||
|
||||
def create_calendars(self):
|
||||
for calendar in self.calendars.values():
|
||||
calendar["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_work_calendar", self.file, name=calendar["Name"]
|
||||
)
|
||||
self.process_working_week(calendar["StandardWorkWeek"], calendar["ifc"])
|
||||
self.process_exceptions(calendar["HolidayOrExceptions"], calendar["ifc"])
|
||||
|
||||
def process_working_week(self, week, calendar):
|
||||
for day in week:
|
||||
if day["ifc"] or not day["WorkTimes"]:
|
||||
continue
|
||||
|
||||
day["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_work_time", self.file, work_calendar=calendar, time_type="WorkingTimes"
|
||||
)
|
||||
weekday_component = [self.day_map[day["DayOfWeek"]]]
|
||||
for day2 in week:
|
||||
if day["DayOfWeek"] == day2["DayOfWeek"]:
|
||||
continue
|
||||
if day["WorkTimes"] == day2["WorkTimes"]:
|
||||
weekday_component.append(self.day_map[day2["DayOfWeek"]])
|
||||
# Don't process the next day, as we can group it
|
||||
day2["ifc"] = day["ifc"]
|
||||
|
||||
work_time_name = "Weekdays: {}".format(", ".join([str(c) for c in sorted(weekday_component)]))
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_work_time",
|
||||
self.file,
|
||||
work_time=day["ifc"],
|
||||
attributes={"Name": work_time_name},
|
||||
)
|
||||
|
||||
recurrence = ifcopenshell.api.run(
|
||||
"sequence.assign_recurrence_pattern", self.file, parent=day["ifc"], recurrence_type="WEEKLY"
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_recurrence_pattern",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
attributes={"WeekdayComponent": weekday_component},
|
||||
)
|
||||
for work_time in day["WorkTimes"]:
|
||||
ifcopenshell.api.run(
|
||||
"sequence.add_time_period",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
start_time=work_time["Start"],
|
||||
end_time=work_time["Finish"],
|
||||
)
|
||||
|
||||
def process_exceptions(self, exceptions, calendar):
|
||||
for year, year_data in exceptions.items():
|
||||
for month, month_data in year_data.items():
|
||||
if month_data["FullDay"]:
|
||||
self.process_full_day_exceptions(year, month, month_data, calendar)
|
||||
if month_data["WorkTime"]:
|
||||
self.process_work_time_exceptions(year, month, month_data, calendar)
|
||||
|
||||
def process_full_day_exceptions(self, year, month, month_data, calendar):
|
||||
work_time = ifcopenshell.api.run(
|
||||
"sequence.add_work_time", self.file, work_calendar=calendar, time_type="ExceptionTimes"
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_work_time",
|
||||
self.file,
|
||||
work_time=work_time,
|
||||
attributes={
|
||||
"Name": f"{year}-{month}",
|
||||
"Start": date(year, 1, 1),
|
||||
"Finish": date(year, 12, 31),
|
||||
},
|
||||
)
|
||||
recurrence = ifcopenshell.api.run(
|
||||
"sequence.assign_recurrence_pattern",
|
||||
self.file,
|
||||
parent=work_time,
|
||||
recurrence_type="YEARLY_BY_DAY_OF_MONTH",
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_recurrence_pattern",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
attributes={"DayComponent": month_data["FullDay"], "MonthComponent": [month]},
|
||||
)
|
||||
|
||||
def process_work_time_exceptions(self, year, month, month_data, calendar):
|
||||
for day in month_data["WorkTime"]:
|
||||
if day["ifc"]:
|
||||
continue
|
||||
|
||||
day["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_work_time", self.file, work_calendar=calendar, time_type="ExceptionTimes"
|
||||
)
|
||||
|
||||
day_component = [day["Day"]]
|
||||
for day2 in month_data["WorkTime"]:
|
||||
if day["Day"] == day2["Day"]:
|
||||
continue
|
||||
if day["WorkTimes"] == day2["WorkTimes"]:
|
||||
day_component.append(day2["Day"])
|
||||
# Don't process the next day, as we can group it
|
||||
day2["ifc"] = day["ifc"]
|
||||
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_work_time",
|
||||
self.file,
|
||||
work_time=day["ifc"],
|
||||
attributes={
|
||||
"Name": "{}-{}-{}".format(year, month, ", ".join([str(d) for d in day_component])),
|
||||
"Start": date(year, 1, 1),
|
||||
"Finish": date(year, 12, 31),
|
||||
},
|
||||
)
|
||||
recurrence = ifcopenshell.api.run(
|
||||
"sequence.assign_recurrence_pattern",
|
||||
self.file,
|
||||
parent=day["ifc"],
|
||||
recurrence_type="YEARLY_BY_DAY_OF_MONTH",
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_recurrence_pattern",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
attributes={"DayComponent": day_component, "MonthComponent": [month]},
|
||||
)
|
||||
for work_time in day["WorkTimes"]:
|
||||
ifcopenshell.api.run(
|
||||
"sequence.add_time_period",
|
||||
self.file,
|
||||
recurrence_pattern=recurrence,
|
||||
start_time=work_time["Start"],
|
||||
end_time=work_time["Finish"],
|
||||
)
|
||||
|
||||
def create_tasks(self, work_schedule):
|
||||
for wbs in self.wbs.values():
|
||||
self.create_task_from_wbs(wbs, work_schedule)
|
||||
for activity_id in self.root_activites:
|
||||
self.create_task_from_activity(self.activities[activity_id], None, work_schedule)
|
||||
|
||||
def create_task_from_wbs(self, wbs, work_schedule):
|
||||
if not self.wbs.get(wbs["ParentObjectId"]):
|
||||
wbs["ParentObjectId"] = None
|
||||
wbs["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_task",
|
||||
self.file,
|
||||
work_schedule=None if wbs["ParentObjectId"] else work_schedule,
|
||||
parent_task=self.wbs[wbs["ParentObjectId"]]["ifc"] if wbs["ParentObjectId"] else None,
|
||||
)
|
||||
identification = wbs["Code"]
|
||||
if wbs["ParentObjectId"]:
|
||||
identification = self.wbs[wbs["ParentObjectId"]]["ifc"].Identification + "." + wbs["Code"]
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_task",
|
||||
self.file,
|
||||
task=wbs["ifc"],
|
||||
attributes={"Name": wbs["Name"], "Identification": identification},
|
||||
)
|
||||
for activity_id in wbs["activities"]:
|
||||
self.create_task_from_activity(self.activities[activity_id], wbs, None)
|
||||
|
||||
def create_task_from_activity(self, activity, wbs, work_schedule):
|
||||
activity["ifc"] = ifcopenshell.api.run(
|
||||
"sequence.add_task",
|
||||
self.file,
|
||||
work_schedule=None if wbs else work_schedule,
|
||||
parent_task=wbs["ifc"] if wbs else None,
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_task",
|
||||
self.file,
|
||||
task=activity["ifc"],
|
||||
attributes={
|
||||
"Name": activity["Name"],
|
||||
"Identification": activity["Identification"],
|
||||
"Status": activity["Status"],
|
||||
"IsMilestone": activity["StartDate"] == activity["FinishDate"],
|
||||
"PredefinedType": "CONSTRUCTION"
|
||||
},
|
||||
)
|
||||
task_time = ifcopenshell.api.run("sequence.add_task_time", self.file, task=activity["ifc"])
|
||||
calendar = self.calendars[activity["CalendarObjectId"]]
|
||||
#print(calendar, self.calendars)
|
||||
# Seems intermittently crashy - can we investigate for larger files?
|
||||
ifcopenshell.api.run(
|
||||
"control.assign_control",
|
||||
self.file,
|
||||
**{
|
||||
"relating_control": calendar["ifc"],
|
||||
"related_object": activity["ifc"],
|
||||
},
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_task_time",
|
||||
self.file,
|
||||
task_time=task_time,
|
||||
attributes={
|
||||
"ScheduleStart": activity["StartDate"],
|
||||
"ScheduleFinish": activity["FinishDate"],
|
||||
"DurationType": "WORKTIME" if activity["PlannedDuration"] else None,
|
||||
"ScheduleDuration": timedelta(
|
||||
days=float(activity["PlannedDuration"]) / float(calendar["HoursPerDay"])
|
||||
)
|
||||
or None
|
||||
if activity["PlannedDuration"]
|
||||
else None,
|
||||
},
|
||||
)
|
||||
|
||||
def create_rel_sequences(self):
|
||||
self.sequence_type_map = {
|
||||
"Start to Start": "START_START",
|
||||
"Start to Finish": "START_FINISH",
|
||||
"Finish to Start": "FINISH_START",
|
||||
"Finish to Finish": "FINISH_FINISH",
|
||||
}
|
||||
for relationship in self.relationships.values():
|
||||
rel_sequence = ifcopenshell.api.run(
|
||||
"sequence.assign_sequence",
|
||||
self.file,
|
||||
relating_process=self.activities[relationship["PredecessorActivity"]]["ifc"],
|
||||
related_process=self.activities[relationship["SuccessorActivity"]]["ifc"],
|
||||
)
|
||||
ifcopenshell.api.run(
|
||||
"sequence.edit_sequence",
|
||||
self.file,
|
||||
rel_sequence=rel_sequence,
|
||||
attributes={"SequenceType": relationship["Type"]},
|
||||
)
|
||||
lag = float(relationship["Lag"])
|
||||
if lag:
|
||||
calendar = self.calendars[self.activities[relationship["PredecessorActivity"]]["CalendarObjectId"]]
|
||||
ifcopenshell.api.run(
|
||||
"sequence.assign_lag_time",
|
||||
self.file,
|
||||
rel_sequence=rel_sequence,
|
||||
lag_value=datetime.timedelta(days=lag / float(calendar["HoursPerDay"])),
|
||||
duration_type="WORKTIME",
|
||||
)
|
||||
|
||||
def create_boilerplate_ifc(self):
|
||||
self.file = ifcopenshell.file(schema="IFC4")
|
||||
self.work_plan = self.file.create_entity("IfcWorkPlan")
|
||||
|
||||
|
||||
|
||||
# TODO: add support for resources
|
||||
# TODO: consider showing progress bar for better user experience
|
||||
|
||||
@@ -9,6 +9,7 @@ class Usecase:
|
||||
"relating_process": None,
|
||||
"related_object": None,
|
||||
}
|
||||
|
||||
for key, value in settings.items():
|
||||
self.settings[key] = value
|
||||
|
||||
|
||||
Reference in New Issue
Block a user