Glassvision/IfcOpenShell
1
0
Fork 0
mirror of https://github.com/IfcOpenShell/IfcOpenShell.git synced 2026-09-15 13:48:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix astapowerproject import to ifc which seemed mostly broken for some reason

Browse Source
This commit is contained in:
Dion Moult
2022-11-11 11:37:56 +11:00
parent ff984f62af
commit a7febaa9b0
2 changed files with 125 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ifc4d/ifc4d/pp2ifc.py
+99 -116
View File
@@ -1,6 +1,7 @@
print("In module products __package__, __name__ ==", __package__) print("In module products __package__, __name__ ==", __package__)
print(__name__) print(__name__)
import sys import sys
print("In module products sys.path[0], __package__ ==", sys.path[0], __package__) print("In module products sys.path[0], __package__ ==", sys.path[0], __package__)
sys.path.append(sys.path[0]) sys.path.append(sys.path[0])
@@ -16,20 +17,20 @@ from .common import ScheduleIfcGenerator
import time import time
list_of_tables = [ list_of_tables = [
'BAR', "BAR",
'EXPANDED_TASK', "EXPANDED_TASK",
'PERMANENT_RESOURCE', "PERMANENT_RESOURCE",
'CONSUMABLE_RESOURCE', "CONSUMABLE_RESOURCE",
'TASK', "TASK",
'PROJECT_SUMMARY', "PROJECT_SUMMARY",
'WBS_ENTRY' "WBS_ENTRY",
] ]
class PP2Ifc: class PP2Ifc:
def __init__(self): def __init__(self):
self.pp = None self.pp = None
self.file = None self.file = None
self.work_plan = None self.work_plan = None
@@ -41,30 +42,14 @@ class PP2Ifc:
self.relationships = {} self.relationships = {}
self.resources = {} self.resources = {}
self.output = None self.output = None
self.day_map = {
"Monday": 1, self.relationship_map = {0: "FINISH_START", 1: "FINISH_FINISH", 2: "START_START", 3: "START_FINISH"}
"Tuesday": 2,
"Wednesday": 3,
"Thursday": 4,
"Friday": 5,
"Saturday": 6,
"Sunday": 7,
}
self.relationship_map = {
0: "FINISH_START",
1: "FINISH_FINISH",
2: "START_START",
3: "START_FINISH"
}
def get_json(self, table_name): def get_json(self, table_name):
self.cur.execute("select * from " + table_name) self.cur.execute("select * from " + table_name)
r = [dict((self.cur.description[i][0], value) for i, value in enumerate(row)) for row in self.cur.fetchall()] r = [dict((self.cur.description[i][0], value) for i, value in enumerate(row)) for row in self.cur.fetchall()]
return r return r
def get_json_with_filter(self, table_name, attr_name, attr_value): def get_json_with_filter(self, table_name, attr_name, attr_value):
self.cur.execute("select * from " + table_name + " where " + attr_name + " = " + str(attr_value)) self.cur.execute("select * from " + table_name + " where " + attr_name + " = " + str(attr_value))
r = [dict((self.cur.description[i][0], value) for i, value in enumerate(row)) for row in self.cur.fetchall()] r = [dict((self.cur.description[i][0], value) for i, value in enumerate(row)) for row in self.cur.fetchall()]
@@ -75,19 +60,19 @@ class PP2Ifc:
self.cur = self.con.cursor() self.cur = self.con.cursor()
self.parse_pp() self.parse_pp()
settings = { settings = {
"work_plan":self.work_plan, "work_plan": self.work_plan,
"project": self.project, "project": self.project,
"calendars": self.calendars, "calendars": self.calendars,
"wbs": self.wbs, "wbs": self.wbs,
"root_activities": self.root_activites, "root_activities": self.root_activites,
"activities": self.activities, "activities": self.activities,
"relationships": self.relationships, "relationships": self.relationships,
"resources": self.resources "resources": self.resources,
} }
start = time.time() start = time.time()
ifcCreator = ScheduleIfcGenerator(self.file, self.output, settings) ifcCreator = ScheduleIfcGenerator(self.file, self.output, settings)
end = time.time() end = time.time()
ifcCreator.create_ifc() ifcCreator.create_ifc()
end2 = time.time() end2 = time.time()
print("Parsing time is", end - start) print("Parsing time is", end - start)
@@ -96,34 +81,35 @@ class PP2Ifc:
# self.create_ifc() # self.create_ifc()
def parse_pp(self): def parse_pp(self):
project = self.get_json("PROJECT_SUMMARY")[0] project = self.get_json("PROJECT_SUMMARY")[0]
self.project["Name"] = project["SHORT_NAME"] self.project["Name"] = project["SHORT_NAME"]
self.parse_calendar_pp() self.parse_calendar_pp()
self.parse_wbs_pp() self.parse_bar()
self.parse_activity_pp()
self.parse_relationship_pp(project) self.parse_relationship_pp(project)
def parse_calendar_pp(self): def parse_calendar_pp(self):
calendars = self.get_json("CALENDAR") calendars = self.get_json("CALENDAR")
wp_data = self.get_json("WORK_PATTERN") wp_data = self.get_json("WORK_PATTERN")
work_types = self.get_json("EXCEPTIONN") work_types = self.get_json("EXCEPTIONN")
work_type_ids = [wt['ID'] for wt in work_types if wt['EXCEPTION_TYPE'] == 0] work_type_ids = [wt["ID"] for wt in work_types if wt["EXCEPTION_TYPE"] == 0]
for calendar in calendars: for calendar in calendars:
calendar_id = calendar["ID"] calendar_id = calendar["ID"]
calendar_wp = calendar["DOMINANT_WORK_PATTERN"] calendar_wp = calendar["DOMINANT_WORK_PATTERN"]
wp_data = self.get_json_with_filter("WORK_PATTERN", "ID", calendar_wp) wp_data = self.get_json_with_filter("WORK_PATTERN", "ID", calendar_wp)
wp = AstaCalendarWorkPattern(wp_data[0]['SHIFTS'], work_type_ids) wp = AstaCalendarWorkPattern(wp_data[0]["SHIFTS"], work_type_ids)
exceptions = {} exceptions = {}
timex = [] timex = []
for times in wp.dict_wp: for times in wp.dict_wp:
work_times = timedelta(hours=0) work_times = timedelta(hours=0)
for daily in times['WorkTimes']: for daily in times["WorkTimes"]:
fin = daily['Finish'] fin = daily["Finish"]
strt = daily['Start'] strt = daily["Start"]
work_times += timedelta(hours=fin.hour, minutes=fin.minute) - timedelta(hours=strt.hour, minutes=strt.minute) work_times += timedelta(hours=fin.hour, minutes=fin.minute) - timedelta(
timex.append(work_times.total_seconds() /(60*60)) hours=strt.hour, minutes=strt.minute
)
timex.append(work_times.total_seconds() / (60 * 60))
self.calendars[calendar_id] = { self.calendars[calendar_id] = {
"Name": calendar["NAME"], "Name": calendar["NAME"],
@@ -132,87 +118,84 @@ class PP2Ifc:
"StandardWorkWeek": wp.dict_wp, "StandardWorkWeek": wp.dict_wp,
"HolidayOrExceptions": exceptions, "HolidayOrExceptions": exceptions,
} }
#print(self.calendars[calendar_id]) # print(self.calendars[calendar_id])
def parse_wbs_pp(self): def parse_bar(self):
bars = self.get_json("BAR") bars = self.get_json("BAR")
extended_tasks = self.get_json("EXPANDED_TASK") expanded_tasks = {t["BAR"]: t for t in self.get_json("EXPANDED_TASK")}
tasks = {t["BAR"]: t for t in self.get_json("TASK")}
milestones = {m["BAR"]: m for m in self.get_json("MILESTONE")}
schedule_task = None
bar_tasks = {}
wbs_activities = {}
for bar in bars: for bar in bars:
self.wbs[bar["ID"]] = { extra_type = None
"Name": bar["NAME"], extra_data = None
"Code": bar["ID"], bar_id = None
"ParentObjectId": bar["EXPANDED_TASK"] if bar["EXPANDED_TASK"]> 0 else None,
"ifc": None,
"rel": None,
"activities": [],
}
for bar in extended_tasks:
self.wbs[bar["ID"]] = {
"Name": bar["NAME"],
"Code": bar["ID"],
"ParentObjectId": bar["BAR"] if bar["BAR"]> 0 else None,
"ifc": None,
"rel": None,
"activities": [],
}
#print(self.wbs)
def parse_activity_pp(self): if bar["ID"] in expanded_tasks:
activities = self.get_json("TASK") extra_type = "EXPANDED_TASK"
milestones = self.get_json("MILESTONE") extra_data = expanded_tasks[bar["ID"]]
for activity in activities: elif bar["ID"] in tasks:
activity_type = "TASK" extra_type = "TASK"
activity_id = activity["ID"] extra_data = tasks[bar["ID"]]
if 'PLANNED_DURATION' in activity.keys(): elif bar["ID"] in milestones:
activity_duration = float(activity["PLANNED_DURATION"].split(",")[-2].replace("<","").replace(">","")) extra_type = "MILESTONE"
elif 'GIVEN_DURATION' in activity.keys(): extra_data = milestones[bar["ID"]]
activity_duration = float(activity["GIVEN_DURATION"].split(",")[-2].replace("<","").replace(">",""))
else: else:
activity_duration = 0.0 # Not sure where this might be the case
continue
wbs_id = activity["BAR"] name = bar["NAME"]
if wbs_id: if extra_data["NAME"]:
self.wbs[wbs_id]["activities"].append(activity_id) name += f" - {extra_data['NAME']}"
bar_tasks[extra_data["ID"]] = {"bar": bar, "extra_type": extra_type, "extra_data": extra_data}
if extra_type == "EXPANDED_TASK":
self.wbs[extra_data["ID"]] = {
"Name": name,
"Code": extra_data["ID"],
"ParentObjectId": bar["EXPANDED_TASK"] if bar["EXPANDED_TASK"] > 0 else None,
"ifc": None,
"rel": None,
"activities": [],
}
else: else:
self.root_activites.append(activity_id) if "PLANNED_DURATION" in extra_data.keys():
self.activities[activity_id] = { activity_duration = float(
"Name": activity["NAME"], extra_data["PLANNED_DURATION"].split(",")[-2].replace("<", "").replace(">", "")
"Identification": activity["ID"], )
"StartDate": datetime.datetime.fromisoformat(activity["LINKABLE_START"]), elif "GIVEN_DURATION" in extra_data.keys():
"FinishDate": datetime.datetime.fromisoformat(activity["LINKABLE_FINISH"]), activity_duration = float(
"PlannedDuration": activity_duration, extra_data["GIVEN_DURATION"].split(",")[-2].replace("<", "").replace(">", "")
"Status": "PLANNED", )
"CalendarObjectId": activity["CALENDAR"], else:
"ifc": None, activity_duration = 0.0
}
self.activities[extra_data["ID"]] = {
for activity in milestones: "Name": name,
activity_type = "MILESTONE" "Identification": extra_data["ID"],
activity_id = activity["ID"] "StartDate": datetime.datetime.fromisoformat(extra_data["LINKABLE_START"]),
wbs_id = activity["BAR"] "FinishDate": datetime.datetime.fromisoformat(extra_data["LINKABLE_FINISH"]),
if wbs_id: "PlannedDuration": activity_duration,
self.wbs[wbs_id]["activities"].append(activity_id) "Status": "PLANNED",
else: "CalendarObjectId": extra_data["CALENDAR"],
self.root_activites.append(activity_id) "ifc": None,
self.activities[activity_id] = { }
"Name": activity["NAME"], wbs_activities.setdefault(bar["EXPANDED_TASK"], []).append(extra_data["ID"])
"Identification": activity["ID"],
"StartDate": datetime.datetime.fromisoformat(activity["LINKABLE_START"]), for wbs, activities in wbs_activities.items():
"FinishDate": datetime.datetime.fromisoformat(activity["LINKABLE_FINISH"]), self.wbs[wbs]["activities"] = activities
"PlannedDuration": 0.0,
"Status": "PLANNED",
"CalendarObjectId": activity["CALENDAR"],
"ifc": None,
}
#print("***ACTIVITIES", self.activities[activity_id])
def parse_relationship_pp(self, project): def parse_relationship_pp(self, project):
relations = self.get_json('LINK') relations = self.get_json("LINK")
print(relations) # print(relations)
for relationship in relations: for relationship in relations:
predecessor = relationship['START_TASK'] predecessor = relationship["START_TASK"]
successor = relationship['END_TASK'] successor = relationship["END_TASK"]
if predecessor not in self.activities: if predecessor not in self.activities:
print(f"Linked predecessor task {predecessor} cannot be found.") print(f"Linked predecessor task {predecessor} cannot be found.")
continue continue
@@ -222,6 +205,6 @@ class PP2Ifc:
self.relationships[relationship["ID"]] = { self.relationships[relationship["ID"]] = {
"PredecessorActivity": predecessor, "PredecessorActivity": predecessor,
"SuccessorActivity": successor, "SuccessorActivity": successor,
"Type": self.relationship_map[relationship['LINK_KIND']], "Type": self.relationship_map[relationship["LINK_KIND"]],
"Lag": float(relationship['END_LAG_TIME'].split(",")[-2].replace("<","").replace(">","")), "Lag": float(relationship["END_LAG_TIME"].split(",")[-2].replace("<", "").replace(">", "")),
} }
src/ifc4d/ifc4d/wpattern.py
+26 -25
View File
@@ -10,28 +10,27 @@ import re
from datetime import time, datetime from datetime import time, datetime
class AstaCalendarWorkPattern: class AstaCalendarWorkPattern:
def get_keys(self, s): def get_keys(self, s):
regex = r"\<(\".+?\")\>(\w|\d|)+?" regex = r"\<(\".+?\")\>(\w|\d|)+?"
matches=re.finditer(regex, s) matches = re.finditer(regex, s)
matcs = [] matcs = []
for matchNum, match in enumerate(matches, start=1): for matchNum, match in enumerate(matches, start=1):
matcs.append(match.group(1)) matcs.append(match.group(1))
return matcs return matcs
def get_values(self, s): def get_values(self, s):
rx2 = r"<\"[^<>]+\">" rx2 = r"<\"[^<>]+\">"
data = re.split('<[^<>]+>', s) data = re.split("<[^<>]+>", s)
return data return data
def __init__(self, string, work_type_ids): def __init__(self, string, work_type_ids):
self.string = string self.string = string
self.Days = { self.Days = {
'en': ['Sunday', 'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday'], "en": ["Sunday", "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday"],
'sv': ['Söndag', 'Måndag', 'Tisdag', 'Onsdag', 'Torsdag', 'Fredag', 'Lördag'], "sv": ["Söndag", "Måndag", "Tisdag", "Onsdag", "Torsdag", "Fredag", "Lördag"],
"de": ["Sonntag", "Montag", "Dienstag", "Mittwoch", "Donnerstag", "Freitag", "Samstag"],
} }
self.keys = self.get_keys(string) self.keys = self.get_keys(string)
self.values = self.get_values(string) self.values = self.get_values(string)
@@ -40,29 +39,31 @@ class AstaCalendarWorkPattern:
for d, m in zip(self.values[1:], self.keys): for d, m in zip(self.values[1:], self.keys):
splt_data = d.strip().split(",") splt_data = d.strip().split(",")
workhours = [] workhours = []
if len(splt_data)>=7: if len(splt_data) >= 7:
number_of_work_slots = splt_data[1] number_of_work_slots = splt_data[1]
for i in range(int(number_of_work_slots)): for i in range(int(number_of_work_slots)):
if int(splt_data[2+i*3]) in work_type_ids: if int(splt_data[2 + i * 3]) in work_type_ids:
st1_1 = datetime.strptime(splt_data[2+i*3+1].ljust(5,'0'), "%H%M%S") st1_1 = datetime.strptime(splt_data[2 + i * 3 + 1].ljust(5, "0"), "%H%M%S")
st1_2 = datetime.strptime(splt_data[2+i*3+2].ljust(5,'0'), "%H%M%S") st1_2 = datetime.strptime(splt_data[2 + i * 3 + 2].ljust(5, "0"), "%H%M%S")
st = {"Start":time(st1_1.hour, st1_1.minute), "Finish": time(st1_2.hour, st1_2.minute), "ifc": None} st = {
"Start": time(st1_1.hour, st1_1.minute),
"Finish": time(st1_2.hour, st1_2.minute),
"ifc": None,
}
workhours.append(st) workhours.append(st)
self.dict_wp.append({'DayOfWeek': m.replace("\"",""), self.dict_wp.append({"DayOfWeek": m.replace('"', ""), "WorkTimes": workhours, "ifc": None})
'WorkTimes': workhours, "ifc": None})
# Translate day names to english # Translate day names to english
for index, day in enumerate(self.Days['en']): for index, day in enumerate(self.Days["en"]):
d = self.dict_wp[-1]['DayOfWeek'] d = self.dict_wp[-1]["DayOfWeek"]
for lang in self.Days.keys(): for lang in self.Days.keys():
if lang == 'en': if lang == "en":
continue continue
self.dict_wp[-1]['DayOfWeek'] = d.replace(self.Days[lang][index], self.Days['en'][index]) self.dict_wp[-1]["DayOfWeek"] = d.replace(self.Days[lang][index], self.Days["en"][index])
for day in self.Days['en']: for day in self.Days["en"]:
if not len(list(filter(lambda d: d['DayOfWeek']== day , self.dict_wp))) > 0: if not len(list(filter(lambda d: d["DayOfWeek"] == day, self.dict_wp))) > 0:
print("MISSING", day) print("MISSING", day)
self.dict_wp.append({'DayOfWeek': day, self.dict_wp.append({"DayOfWeek": day, "WorkTimes": [], "ifc": None})
'WorkTimes': [], "ifc": None})
print("final", self.dict_wp) print("final", self.dict_wp)