mirror of
https://github.com/IfcOpenShell/IfcOpenShell.git
synced 2026-08-09 09:21:46 +00:00
Implement support of importing calendars from P6
This commit is contained in:
Dion Moult
@@ -278,7 +278,8 @@ class BIM_PT_work_calendars(Panel):
|
||||
row = self.layout.row(align=True)
|
||||
row.label(text=work_calendar["Name"] or "Unnamed", icon="VIEW_ORTHO")
|
||||
if self.props.active_work_calendar_id == work_calendar_id:
|
||||
row.operator("bim.edit_work_calendar", text="", icon="CHECKMARK")
|
||||
if self.props.is_editing == "ATTRIBUTES":
|
||||
row.operator("bim.edit_work_calendar", text="", icon="CHECKMARK")
|
||||
row.operator("bim.disable_editing_work_calendar", text="", icon="CANCEL")
|
||||
elif self.props.active_work_calendar_id:
|
||||
row.operator("bim.remove_work_calendar", text="", icon="X").work_calendar = work_calendar_id
|
||||
@@ -313,7 +314,7 @@ class BIM_PT_work_calendars(Panel):
|
||||
def draw_work_time_ui(self, work_time, time_type):
|
||||
row = self.layout.row(align=True)
|
||||
row.label(
|
||||
text=work_time["Name"] or "Unnamed", icon="MESH_GRID" if time_type == "WorkingTimes" else "LIGHTPROBE_GRID"
|
||||
text=work_time["Name"] or "Unnamed", icon="AUTO" if time_type == "WorkingTimes" else "HOME"
|
||||
)
|
||||
if work_time["Start"] or work_time["Finish"]:
|
||||
row.label(text="{} - {}".format(work_time["Start"] or "*", work_time["Finish"] or "*"))
|
||||
|
||||
@@ -4,7 +4,7 @@ import ifcopenshell.api
|
||||
class Usecase:
|
||||
def __init__(self, file, **settings):
|
||||
self.file = file
|
||||
self.settings = {"name": "Unnamed", "predefined_type": "NOTDEFINED", "working_times": [], "exception_times": []}
|
||||
self.settings = {"name": "Unnamed", "predefined_type": "NOTDEFINED"}
|
||||
for key, value in settings.items():
|
||||
self.settings[key] = value
|
||||
|
||||
|
||||
@@ -19,3 +19,4 @@ class Usecase:
|
||||
if len(self.file.get_inverse(self.settings["parent"].Recurrence)) == 1:
|
||||
self.file.remove(self.settings["parent"].Recurrence)
|
||||
self.settings["parent"].Recurrence = recurrence
|
||||
return recurrence
|
||||
|
||||
@@ -1,6 +1,3 @@
|
||||
import ifcopenshell.util.date
|
||||
|
||||
|
||||
class Usecase:
|
||||
def __init__(self, file, **settings):
|
||||
self.file = file
|
||||
@@ -10,17 +7,4 @@ class Usecase:
|
||||
|
||||
def execute(self):
|
||||
for name, value in self.settings["attributes"].items():
|
||||
if name == "TimePeriods" and value:
|
||||
periods = []
|
||||
for period in value:
|
||||
periods.append(
|
||||
self.file.create_entity(
|
||||
"IfcTimePeriod",
|
||||
**{
|
||||
"StartTime": ifcopenshell.util.date.datetime2ifc(period[0]),
|
||||
"EndTime": ifcopenshell.util.date.datetime2ifc(period[1]),
|
||||
},
|
||||
)
|
||||
)
|
||||
value = periods
|
||||
setattr(self.settings["recurrence_pattern"], name, value)
|
||||
|
||||
+235
-24
@@ -1,8 +1,8 @@
|
||||
import datetime
|
||||
import ifcopenshell
|
||||
import ifcopenshell.api
|
||||
import ifcopenshell.util.date
|
||||
import xml.etree.ElementTree as ET
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class P62Ifc:
|
||||
@@ -11,10 +11,20 @@ class P62Ifc:
|
||||
self.file = None
|
||||
self.work_plan = None
|
||||
self.project = {}
|
||||
self.calendars = {}
|
||||
self.wbs = {}
|
||||
self.root_activites = []
|
||||
self.activities = {}
|
||||
self.relationships = {}
|
||||
self.day_map = {
|
||||
"Monday": 1,
|
||||
"Tuesday": 2,
|
||||
"Wednesday": 3,
|
||||
"Thursday": 4,
|
||||
"Friday": 5,
|
||||
"Saturday": 6,
|
||||
"Sunday": 7,
|
||||
}
|
||||
|
||||
def execute(self):
|
||||
self.parse_xml()
|
||||
@@ -22,58 +32,259 @@ class P62Ifc:
|
||||
|
||||
def parse_xml(self):
|
||||
tree = ET.parse(self.xml)
|
||||
ns = {"pr": "http://xmlns.oracle.com/Primavera/P6/V19.12/API/BusinessObjects"}
|
||||
self.ns = {"pr": "http://xmlns.oracle.com/Primavera/P6/V19.12/API/BusinessObjects"}
|
||||
root = tree.getroot()
|
||||
project = root.find("pr:Project", ns)
|
||||
self.project["Name"] = project.find("pr:Name", ns).text
|
||||
project = root.find("pr:Project", self.ns)
|
||||
self.project["Name"] = project.find("pr:Name", self.ns).text
|
||||
self.parse_calendar_xml(project)
|
||||
self.parse_wbs_xml(project)
|
||||
self.parse_activity_xml(project)
|
||||
self.parse_relationship_xml(project)
|
||||
|
||||
for wbs in project.findall("pr:WBS", ns):
|
||||
self.wbs[wbs.find("pr:ObjectId", ns).text] = {
|
||||
"Name": wbs.find("pr:Name", ns).text,
|
||||
"Code": wbs.find("pr:Code", ns).text,
|
||||
"ParentObjectId": wbs.find("pr:ParentObjectId", ns).text,
|
||||
def parse_calendar_xml(self, project):
|
||||
for calendar in project.findall("pr:Calendar", self.ns):
|
||||
calendar_id = calendar.find("pr:ObjectId", self.ns).text
|
||||
standard_work_week = []
|
||||
for standard_work_hour in calendar.find("pr:StandardWorkWeek", self.ns).findall(
|
||||
"pr:StandardWorkHours", self.ns
|
||||
):
|
||||
work_times = []
|
||||
for work_time in standard_work_hour.findall("pr:WorkTime", self.ns):
|
||||
if work_time.find("pr:Start", self.ns) is None:
|
||||
continue
|
||||
work_times.append(
|
||||
{
|
||||
"Start": datetime.time.fromisoformat(work_time.find("pr:Start", self.ns).text),
|
||||
"Finish": datetime.time.fromisoformat(work_time.find("pr:Finish", self.ns).text),
|
||||
}
|
||||
)
|
||||
standard_work_week.append(
|
||||
{
|
||||
"DayOfWeek": standard_work_hour.find("pr:DayOfWeek", self.ns).text,
|
||||
"WorkTimes": work_times,
|
||||
"ifc": None,
|
||||
}
|
||||
)
|
||||
exceptions = {}
|
||||
holiday_or_exceptions = calendar.find("pr:HolidayOrExceptions", self.ns)
|
||||
holiday_or_exception = []
|
||||
if holiday_or_exceptions:
|
||||
holiday_or_exception = holiday_or_exceptions.findall("pr:HolidayOrException", self.ns)
|
||||
for exception in holiday_or_exception:
|
||||
d = datetime.datetime.fromisoformat(exception.find("pr:Date", self.ns).text).date()
|
||||
month = exceptions.setdefault(d.year, {}).setdefault(d.month, {})
|
||||
month.setdefault("FullDay", [])
|
||||
month.setdefault("WorkTime", [])
|
||||
work_times = []
|
||||
for work_time in exception.findall("pr:WorkTime", self.ns):
|
||||
if work_time.find("pr:Start", self.ns) is None:
|
||||
continue
|
||||
work_times.append(
|
||||
{
|
||||
"Start": datetime.time.fromisoformat(work_time.find("pr:Start", self.ns).text),
|
||||
"Finish": datetime.time.fromisoformat(work_time.find("pr:Finish", self.ns).text),
|
||||
}
|
||||
)
|
||||
if work_times:
|
||||
exceptions[d.year][d.month]["WorkTime"].append({"Day": d.day, "WorkTimes": work_times, "ifc": None})
|
||||
else:
|
||||
exceptions[d.year][d.month]["FullDay"].append(d.day)
|
||||
self.calendars[calendar_id] = {
|
||||
"Name": calendar.find("pr:Name", self.ns).text,
|
||||
"Type": calendar.find("pr:Type", self.ns).text,
|
||||
"StandardWorkWeek": standard_work_week,
|
||||
"HolidayOrExceptions": exceptions,
|
||||
}
|
||||
|
||||
def parse_wbs_xml(self, project):
|
||||
for wbs in project.findall("pr:WBS", self.ns):
|
||||
self.wbs[wbs.find("pr:ObjectId", self.ns).text] = {
|
||||
"Name": wbs.find("pr:Name", self.ns).text,
|
||||
"Code": wbs.find("pr:Code", self.ns).text,
|
||||
"ParentObjectId": wbs.find("pr:ParentObjectId", self.ns).text,
|
||||
"ifc": None,
|
||||
"rel": None,
|
||||
"activities": [],
|
||||
}
|
||||
|
||||
for activity in project.findall("pr:Activity", ns):
|
||||
activity_id = activity.find("pr:ObjectId", ns).text
|
||||
wbs_id = activity.find("pr:WBSObjectId", ns).text
|
||||
def parse_activity_xml(self, project):
|
||||
for activity in project.findall("pr:Activity", self.ns):
|
||||
activity_id = activity.find("pr:ObjectId", self.ns).text
|
||||
wbs_id = activity.find("pr:WBSObjectId", self.ns).text
|
||||
if wbs_id:
|
||||
self.wbs[wbs_id]["activities"].append(activity_id)
|
||||
else:
|
||||
self.root_activites.append(activity_id)
|
||||
self.activities[activity_id] = {
|
||||
"Name": activity.find("pr:Name", ns).text,
|
||||
"Identification": activity.find("pr:Id", ns).text,
|
||||
"StartDate": datetime.fromisoformat(activity.find("pr:StartDate", ns).text),
|
||||
"FinishDate": datetime.fromisoformat(activity.find("pr:FinishDate", ns).text),
|
||||
"Status": activity.find("pr:Status", ns).text,
|
||||
"Name": activity.find("pr:Name", self.ns).text,
|
||||
"Identification": activity.find("pr:Id", self.ns).text,
|
||||
"StartDate": datetime.datetime.fromisoformat(activity.find("pr:StartDate", self.ns).text),
|
||||
"FinishDate": datetime.datetime.fromisoformat(activity.find("pr:FinishDate", self.ns).text),
|
||||
"Status": activity.find("pr:Status", self.ns).text,
|
||||
"ifc": None,
|
||||
}
|
||||
|
||||
for relationship in project.findall("pr:Relationship", ns):
|
||||
self.relationships[relationship.find("pr:ObjectId", ns).text] = {
|
||||
"PredecessorActivity": relationship.find("pr:PredecessorActivityObjectId", ns).text,
|
||||
"SuccessorActivity": relationship.find("pr:SuccessorActivityObjectId", ns).text,
|
||||
def parse_relationship_xml(self, project):
|
||||
for relationship in project.findall("pr:Relationship", self.ns):
|
||||
self.relationships[relationship.find("pr:ObjectId", self.ns).text] = {
|
||||
"PredecessorActivity": relationship.find("pr:PredecessorActivityObjectId", self.ns).text,
|
||||
"SuccessorActivity": relationship.find("pr:SuccessorActivityObjectId", self.ns).text,
|
||||
}
|
||||
|
||||
def get_wbs(self, wbs):
|
||||
return {"Name": wbs.find("pr:Name", ns).text, "subtasks": []}
|
||||
return {"Name": wbs.find("pr:Name", self.ns).text, "subtasks": []}
|
||||
|
||||
def create_ifc(self):
|
||||
if not self.file:
|
||||
self.file = self.create_boilerplate_ifc()
|
||||
work_schedule = self.create_work_schedule()
|
||||
self.create_tasks(work_schedule)
|
||||
self.create_rel_sequences()
|
||||
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"], predefined_type=calendar["Type"]
|
||||
)
|
||||
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"]:
|
||||
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], "Occurrences": 1},
|
||||
)
|
||||
|
||||
|
||||
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], "Occurrences": 1},
|
||||
)
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user