diff --git a/src/ifc4d/ifc4d/common.py b/src/ifc4d/ifc4d/common.py index 296b8f36cb..abd177c340 100644 --- a/src/ifc4d/ifc4d/common.py +++ b/src/ifc4d/ifc4d/common.py @@ -4,7 +4,11 @@ from datetime import date, datetime, timedelta from typing import Any, TypedDict, Union import ifcopenshell +import ifcopenshell.guid +import ifcopenshell.util.date import ifcopenshell.api.control +import ifcopenshell.api.owner +import ifcopenshell.api.pset import ifcopenshell.api.resource import ifcopenshell.api.sequence from typing_extensions import NotRequired @@ -41,6 +45,20 @@ class Activity(TypedDict): PlannedDuration: float Status: str CalendarObjectId: str + Type: NotRequired[str] + # {Title: (ifc_type, value)} — the IFC type comes from the P6 UDFType + # declaration, not from the value, so it is resolved at parse time. + UDFs: NotRequired[dict[str, tuple[str, str]]] + Codes: NotRequired[dict[str, str]] + ActualStartDate: NotRequired[Union[datetime, None]] + ActualFinishDate: NotRequired[Union[datetime, None]] + EarlyStartDate: NotRequired[Union[datetime, None]] + EarlyFinishDate: NotRequired[Union[datetime, None]] + LateStartDate: NotRequired[Union[datetime, None]] + LateFinishDate: NotRequired[Union[datetime, None]] + TotalFloat: NotRequired[Union[float, None]] + FreeFloat: NotRequired[Union[float, None]] + PercentComplete: NotRequired[Union[float, None]] ifc: Union[ifcopenshell.entity_instance, None] @@ -230,8 +248,21 @@ class ScheduleIfcGenerator: ) def create_tasks(self, work_schedule: ifcopenshell.entity_instance) -> None: - for wbs in self.wbs.values(): - self.create_task_from_wbs(wbs, work_schedule) + # Depth first, siblings in P6's SequenceNumber order rather than the + # order the export happens to list them in. Parents still come before + # children, which create_task_from_wbs depends on to find its parent's + # IFC task. + siblings: dict[Any, list[tuple[float, Any]]] = {} + for wbs_id, wbs in self.wbs.items(): + parent = wbs["ParentObjectId"] if self.wbs.get(wbs["ParentObjectId"]) else None + siblings.setdefault(parent, []).append((wbs.get("SequenceNumber") or 0, wbs_id)) + + def emit(parent) -> None: + for _, wbs_id in sorted(siblings.get(parent, []), key=lambda pair: pair[0]): + self.create_task_from_wbs(self.wbs[wbs_id], work_schedule) + emit(wbs_id) + + emit(None) for activity_id in self.root_activites: self.create_task_from_activity(self.activities[activity_id], None, work_schedule) @@ -275,9 +306,23 @@ class ScheduleIfcGenerator: "Identification": str(activity["Identification"]), "Status": activity["Status"], "IsMilestone": activity["StartDate"] == activity["FinishDate"], - "PredefinedType": "CONSTRUCTION", + # A P6 Level of Effort activity has no duration of its own: it + # stretches to span whatever it hangs off, and its dates are + # derived from its relationships rather than planned. IFC has no + # such concept, and ATTENDANCE is the closest thing in + # IfcTaskTypeEnum — support work that runs alongside the tasks + # it serves rather than driving them. The derivation itself is + # not lost: the SS/FS and FF predecessors P6 computes the span + # from are ordinary relationships, and they are written out as + # IfcRelSequence like any other, so a scheduler can recompute + # the span the same way P6 did. + "PredefinedType": ( + "ATTENDANCE" if activity.get("Type") == "Level of Effort" else "CONSTRUCTION" + ), }, ) + self.create_udf_pset(activity) + self.create_code_pset(activity) task_time = ifcopenshell.api.sequence.add_task_time(self.file, task=activity["ifc"]) calendar = self.calendars[activity["CalendarObjectId"]] # Seems intermittently crashy - can we investigate for larger files? @@ -286,21 +331,101 @@ class ScheduleIfcGenerator: relating_control=calendar["ifc"], related_objects=[activity["ifc"]], ) - ifcopenshell.api.sequence.edit_task_time( + self.transcribe_task_time(task_time, activity, calendar) + + def transcribe_task_time( + self, + task_time: ifcopenshell.entity_instance, + activity: Activity, + calendar: Calendar, + ) -> None: + """Write P6's times onto the IfcTaskTime verbatim. + + Intentionally do not recalculate durations or finish times. Match P6 + exactly. Users may recalculate later if needed. + """ + date = ifcopenshell.util.date.datetime2ifc + hours_per_day = float(calendar["HoursPerDay"] or 8) + + task_time.ScheduleStart = date(activity["StartDate"], "IfcDateTime") + task_time.ScheduleFinish = date(activity["FinishDate"], "IfcDateTime") + task_time.DurationType = "WORKTIME" + planned = activity["PlannedDuration"] + if planned is not None and ( + float(planned) or activity["StartDate"] == activity["FinishDate"] + ): + task_time.ScheduleDuration = date( + timedelta(days=float(planned) / hours_per_day), "IfcDuration" + ) + + for attribute, value in ( + ("ActualStart", activity.get("ActualStartDate")), + ("ActualFinish", activity.get("ActualFinishDate")), + ("EarlyStart", activity.get("EarlyStartDate")), + ("EarlyFinish", activity.get("EarlyFinishDate")), + ("LateStart", activity.get("LateStartDate")), + ("LateFinish", activity.get("LateFinishDate")), + ): + if value is not None: + setattr(task_time, attribute, date(value, "IfcDateTime")) + + # P6 states float in hours against the activity's own calendar, so the + # calendar is consulted for nothing more than that conversion. + for attribute, hours in ( + ("TotalFloat", activity.get("TotalFloat")), + ("FreeFloat", activity.get("FreeFloat")), + ): + if hours is not None: + setattr(task_time, attribute, date(timedelta(days=hours / hours_per_day), "IfcDuration")) + if activity.get("TotalFloat") is not None: + task_time.IsCritical = activity["TotalFloat"] <= 0 + + # Completion is an IfcPositiveRatioMeasure, so zero is not merely + # uninteresting, it is invalid. An activity that has not started says + # nothing rather than saying "0% done". + if activity.get("PercentComplete"): + task_time.Completion = activity["PercentComplete"] + + def create_udf_pset(self, activity: Activity) -> None: + """This activity's P6 user-defined fields, as a P6_UDF property set.""" + assert self.file + udfs = activity.get("UDFs") + if not udfs: + return + pset = ifcopenshell.api.pset.add_pset(self.file, product=activity["ifc"], name="P6_UDF") + ifcopenshell.api.pset.edit_pset( 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 8)) or None - if activity["PlannedDuration"] - else None - ), + pset=pset, + properties={ + title: self.file.create_entity(ifc_type, value) + for title, (ifc_type, value) in udfs.items() }, ) + def create_code_pset(self, activity: Activity) -> None: + """This activity's P6 activity codes, as a P6_ActivityCodes set.""" + assert self.file + codes = activity.get("Codes") + if not codes: + return + pset = ifcopenshell.api.pset.add_pset( + self.file, product=activity["ifc"], name="P6_ActivityCodes" + ) + ifcopenshell.api.pset.edit_pset( + self.file, + pset=pset, + properties={name: self.file.create_entity("IfcLabel", value) + for name, (value, _) in codes.items()}, + ) + # A P6 code carries a short value and a readable description — "SO" and + # "Start on Site Milestone". Both matter to a reader, and IfcProperty + # already has the second slot, so the description goes on Description + # rather than being mangled into the value or dropped. + for prop in pset.HasProperties: + description = codes.get(prop.Name, (None, ""))[1] + if description: + prop.Description = description + def create_rel_sequences(self) -> None: self.sequence_type_map = { "Start to Start": "START_START", @@ -309,16 +434,29 @@ class ScheduleIfcGenerator: "Finish to Finish": "FINISH_FINISH", } for relationship in self.relationships.values(): - rel_sequence = ifcopenshell.api.sequence.assign_sequence( - self.file, - relating_process=self.activities[relationship["PredecessorActivity"]]["ifc"], - related_process=self.activities[relationship["SuccessorActivity"]]["ifc"], - ) - ifcopenshell.api.sequence.edit_sequence( - self.file, - rel_sequence=rel_sequence, - attributes={"SequenceType": relationship["Type"]}, + predecessor = self.activities[relationship["PredecessorActivity"]]["ifc"] + successor = self.activities[relationship["SuccessorActivity"]]["ifc"] + + rel_sequence = next( + ( + rel + for rel in successor.IsSuccessorFrom or [] + if rel.RelatingProcess == predecessor + and rel.SequenceType == relationship["Type"] + ), + None, ) + if rel_sequence is None: + attributes = { + "GlobalId": ifcopenshell.guid.new(), + "RelatingProcess": predecessor, + "RelatedProcess": successor, + "SequenceType": relationship["Type"], + } + owner_history = ifcopenshell.api.owner.create_owner_history(self.file) + if owner_history is not None: + attributes["OwnerHistory"] = owner_history + rel_sequence = self.file.create_entity("IfcRelSequence", **attributes) lag = float(relationship["Lag"]) if lag: calendar = self.calendars[self.activities[relationship["PredecessorActivity"]]["CalendarObjectId"]] diff --git a/src/ifc4d/ifc4d/p62ifc.py b/src/ifc4d/ifc4d/p62ifc.py index aa6aafc8dd..625a05fb31 100644 --- a/src/ifc4d/ifc4d/p62ifc.py +++ b/src/ifc4d/ifc4d/p62ifc.py @@ -21,6 +21,25 @@ import xml.etree.ElementTree as ET from .common import ScheduleIfcGenerator +# P6 declares its user-defined fields once at the root and then references them +# by ObjectId from each activity, so the declaration is the only place the type +# is known. Each entry is the element P6 carries the value in, and the IFC type +# it becomes. Text maps to IfcLabel rather than IfcText: these are short tags +# and names — a planner, a responsible party, a work front — and IfcText means +# long-form prose. +# The third element coerces the XML text: IFC's numeric simple types reject a +# string outright ("Attribute not set"), and the value element carries text +# whatever the declared type says. +UDF_DATA_TYPES = { + "Text": ("TextValue", "IfcLabel", str), + "Double": ("DoubleValue", "IfcReal", float), + "Integer": ("IntegerValue", "IfcInteger", int), + "Cost": ("CostValue", "IfcMonetaryMeasure", float), + "Indicator": ("IndicatorValue", "IfcLabel", str), + "Start Date": ("StartDateValue", "IfcDateTime", str), + "Finish Date": ("FinishDateValue", "IfcDateTime", str), +} + class P62Ifc: def __init__(self): @@ -31,6 +50,9 @@ class P62Ifc: self.default_calendar_id = None self.calendars = {} self.wbs = {} + self.udf_types = {} + self.code_types = {} + self.code_values = {} self.root_activites = [] self.activities = {} self.relationships = {} @@ -89,10 +111,15 @@ class P62Ifc: root = tree.getroot() self.ns = {"pr": root.tag[1:].partition("}")[0]} project = root.find("pr:Project", self.ns) - self.project["Name"] = project.findtext("pr:Name") or "Unnamed" + # findtext needs the namespace map too — without it the "pr:" prefix + # resolves to nothing, every schedule came out named "Unnamed", and that + # is the string a viewer puts in its header. + self.project["Name"] = project.findtext("pr:Name", namespaces=self.ns) or "Unnamed" self.default_calendar_id = project.findtext("pr:ActivityDefaultCalendarObjectId", namespaces=self.ns) self.parse_calendar_xml(root) self.parse_calendar_xml(project) + self.parse_udf_type_xml(root) + self.parse_code_type_xml(root) self.parse_wbs_xml(project) self.parse_activity_xml(project) self.parse_relationship_xml(project) @@ -154,12 +181,85 @@ class P62Ifc: "HolidayOrExceptions": exceptions, } + def parse_udf_type_xml(self, root): + """The declarations for P6's user-defined fields. + + Only Activity fields are read. P6 also allows them on projects, + resources and WBS nodes, and those would need somewhere else to land. + A type this importer has no mapping for is skipped rather than guessed + at, because the value element it would be carried in is not knowable + from the value itself. + """ + for udf_type in root.findall("pr:UDFType", self.ns): + if udf_type.findtext("pr:SubjectArea", namespaces=self.ns) != "Activity": + continue + data_type = udf_type.findtext("pr:DataType", namespaces=self.ns) + if data_type not in UDF_DATA_TYPES: + continue + self.udf_types[udf_type.findtext("pr:ObjectId", namespaces=self.ns)] = { + "Title": udf_type.findtext("pr:Title", namespaces=self.ns), + "DataType": data_type, + } + + def parse_udfs(self, activity): + """This activity's user-defined fields, as {Title: (ifc_type, value)}.""" + udfs = {} + for udf in activity.findall("pr:UDF", self.ns): + declared = self.udf_types.get(udf.findtext("pr:TypeObjectId", namespaces=self.ns)) + if not declared: + continue + value_tag, ifc_type, coerce = UDF_DATA_TYPES[declared["DataType"]] + value = udf.findtext(f"pr:{value_tag}", namespaces=self.ns) + if value is None or value == "": + continue + try: + value = coerce(value) + except (TypeError, ValueError): + # A value that does not match its own declared type is one bad + # field, not a reason to lose the whole programme. + continue + udfs[declared["Title"]] = (ifc_type, value) + return udfs + + def parse_code_type_xml(self, root): + """Activity code types and their values, both keyed by ObjectId. + + Codes are a two-table reference like everything else in a P6 export: an + activity carries , and the readable + name and value live up here. + """ + for code_type in root.findall("pr:ActivityCodeType", self.ns): + self.code_types[code_type.findtext("pr:ObjectId", namespaces=self.ns)] = ( + code_type.findtext("pr:Name", namespaces=self.ns) + ) + for code in root.findall("pr:ActivityCode", self.ns): + self.code_values[code.findtext("pr:ObjectId", namespaces=self.ns)] = ( + code.findtext("pr:CodeValue", namespaces=self.ns), + code.findtext("pr:Description", namespaces=self.ns) or "", + ) + + def parse_codes(self, activity): + """This activity's assigned activity codes, as {type: (value, description)}.""" + codes = {} + for code in activity.findall("pr:Code", self.ns): + name = self.code_types.get(code.findtext("pr:TypeObjectId", namespaces=self.ns)) + assigned = self.code_values.get(code.findtext("pr:ValueObjectId", namespaces=self.ns)) + if name and assigned and assigned[0]: + codes[name] = assigned + return codes + 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, + # P6's own ordering of siblings, which is NOT the order the + # export lists them in. It is the only thing that reproduces the + # breakdown a planner recognises, and IfcRelNests preserves it + # for free because RelatedObjects is an ordered LIST — provided + # the tasks are created in this order in the first place. + "SequenceNumber": self.parse_float(wbs, "SequenceNumber") or 0, "ifc": None, "rel": None, "activities": [], @@ -168,8 +268,6 @@ class P62Ifc: def parse_activity_xml(self, project): for activity in project.findall("pr:Activity", self.ns): activity_type = activity.find("pr:Type", self.ns).text - if activity_type == "Level of Effort": - continue activity_id = activity.find("pr:ObjectId", self.ns).text wbs_id = activity.find("pr:WBSObjectId", self.ns).text if wbs_id: @@ -187,9 +285,47 @@ class P62Ifc: "PlannedDuration": activity.find("pr:PlannedDuration", self.ns).text, "Status": activity.find("pr:Status", self.ns).text, "CalendarObjectId": calendar_id or self.default_calendar_id, + "Type": activity_type, + "UDFs": self.parse_udfs(activity), + "Codes": self.parse_codes(activity), + # StartDate/FinishDate above are P6's CURRENT dates, which is + # the live plan — P6 re-plans as actuals land, so on a started + # activity the current start is the actual start. That is why + # they, and not PlannedStartDate, map to ScheduleStart/Finish. + # The actuals are carried separately so the fact that a date is + # recorded rather than forecast is not lost. + "ActualStartDate": self.parse_date(activity, "ActualStartDate"), + "ActualFinishDate": self.parse_date(activity, "ActualFinishDate"), + "EarlyStartDate": self.parse_date(activity, "EarlyStartDate"), + "EarlyFinishDate": self.parse_date(activity, "EarlyFinishDate"), + "LateStartDate": self.parse_date(activity, "LateStartDate"), + "LateFinishDate": self.parse_date(activity, "LateFinishDate"), + "TotalFloat": self.parse_float(activity, "TotalFloat"), + "FreeFloat": self.parse_float(activity, "FreeFloat"), + # P6 stores this as a 0..1 fraction, which is already what + # IfcPositiveRatioMeasure wants. + "PercentComplete": self.parse_float(activity, "PercentComplete"), "ifc": None, } + def parse_date(self, activity, tag): + text = activity.findtext(f"pr:{tag}", namespaces=self.ns) + if not text: + return None + try: + return datetime.datetime.fromisoformat(text) + except ValueError: + return None + + def parse_float(self, activity, tag): + text = activity.findtext(f"pr:{tag}", namespaces=self.ns) + if not text: + return None + try: + return float(text) + except ValueError: + return None + def parse_relationship_xml(self, project): for relationship in project.findall("pr:Relationship", self.ns): predecessor = relationship.find("pr:PredecessorActivityObjectId", self.ns).text