pp2ifc refactor and add support for more languages #5417

This commit is contained in:
Andrej730
2024-09-19 14:20:26 +05:00
parent a5b3c9beb4
commit 96756d63d4
3 changed files with 216 additions and 87 deletions
+75 -18
View File
@@ -1,12 +1,61 @@
from xerparser.reader import Reader
from __future__ import annotations
import ifcopenshell
import ifcopenshell.api
import ifcopenshell.util.date
from datetime import datetime, timedelta, date
from typing import Union, Any, TypedDict, NotRequired
class WorkSlot(TypedDict):
DayOfWeek: str
WorkTimes: list[dict[str, Any]]
ifc: Union[ifcopenshell.entity_instance, None]
class ExceptionDict(TypedDict):
WorkTime: list[int]
FullDay: list[int]
ExceptionsDict = dict[int, dict[int, ExceptionDict]]
class Calendar(TypedDict):
Name: str
Type: str
HoursPerDay: int
StandardWorkWeek: list[WorkSlot]
HolidayOrExceptions: NotRequired[ExceptionsDict]
ifc: NotRequired[ifcopenshell.entity_instance]
class Activity(TypedDict):
Name: str
Identification: int
StartDate: datetime
FinishDate: datetime
PlannedDuration: float
Status: str
CalendarObjectId: str
ifc: Union[ifcopenshell.entity_instance, None]
class WBSEntry(TypedDict):
"""Work Breakdown Strcture Entry"""
Name: str
Code: int
ParentObjectId: Union[int, None]
ifc: Union[ifcopenshell.entity_instance, None]
rel: Union[ifcopenshell.entity_instance, None]
activities: list[int]
class ScheduleIfcGenerator:
def __init__(self, file, output, settings):
file: Union[ifcopenshell.file, None]
calendars: dict[int, Calendar]
def __init__(self, file: Union[ifcopenshell.file, None], output, settings):
self.file = file
self.work_plan = settings["work_plan"]
self.project = settings["project"]
@@ -27,7 +76,7 @@ class ScheduleIfcGenerator:
"Sunday": 7,
}
def create_ifc(self):
def create_ifc(self) -> None:
if not self.file:
self.file = self.create_boilerplate_ifc()
if not self.work_plan:
@@ -40,18 +89,18 @@ class ScheduleIfcGenerator:
if self.output:
self.file.write(self.output)
def create_work_schedule(self):
def create_work_schedule(self) -> ifcopenshell.entity_instance:
return ifcopenshell.api.run(
"sequence.add_work_schedule", self.file, name=self.project["Name"], work_plan=self.work_plan
)
def create_calendars(self):
def create_calendars(self) -> None:
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.get("HolidayOrExceptions"), calendar["ifc"])
def process_working_week(self, week, calendar):
def process_working_week(self, week: list[WorkSlot], calendar: ifcopenshell.entity_instance) -> None:
for day in week:
if day["ifc"] or not day.get("WorkTimes"):
continue
@@ -94,7 +143,9 @@ class ScheduleIfcGenerator:
end_time=work_time["Finish"],
)
def process_exceptions(self, exceptions, calendar):
def process_exceptions(
self, exceptions: Union[ExceptionsDict, None], calendar: ifcopenshell.entity_instance
) -> None:
if exceptions:
for year, year_data in exceptions.items():
for month, month_data in year_data.items():
@@ -103,7 +154,9 @@ class ScheduleIfcGenerator:
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):
def process_full_day_exceptions(
self, year: int, month: int, month_data: dict[str, Any], calendar: ifcopenshell.entity_instance
):
work_time = ifcopenshell.api.run(
"sequence.add_work_time", self.file, work_calendar=calendar, time_type="ExceptionTimes"
)
@@ -130,7 +183,9 @@ class ScheduleIfcGenerator:
attributes={"DayComponent": month_data["FullDay"], "MonthComponent": [month]},
)
def process_work_time_exceptions(self, year, month, month_data, calendar):
def process_work_time_exceptions(
self, year: int, month: int, month_data: dict[str, Any], calendar: ifcopenshell.entity_instance
) -> None:
for day in month_data["WorkTime"]:
if day["ifc"]:
continue
@@ -179,13 +234,13 @@ class ScheduleIfcGenerator:
end_time=work_time["Finish"],
)
def create_tasks(self, work_schedule):
def create_tasks(self, work_schedule: ifcopenshell.entity_instance) -> None:
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):
def create_task_from_wbs(self, wbs: WBSEntry, work_schedule: ifcopenshell.entity_instance) -> None:
if not self.wbs.get(wbs["ParentObjectId"]):
wbs["ParentObjectId"] = None
wbs["ifc"] = ifcopenshell.api.run(
@@ -207,7 +262,12 @@ class ScheduleIfcGenerator:
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):
def create_task_from_activity(
self,
activity: Activity,
wbs: Union[WBSEntry, None],
work_schedule: Union[ifcopenshell.entity_instance, None],
) -> None:
activity["ifc"] = ifcopenshell.api.run(
"sequence.add_task",
self.file,
@@ -228,7 +288,6 @@ class ScheduleIfcGenerator:
)
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",
@@ -254,7 +313,7 @@ class ScheduleIfcGenerator:
},
)
def create_rel_sequences(self):
def create_rel_sequences(self) -> None:
self.sequence_type_map = {
"Start to Start": "START_START",
"Start to Finish": "START_FINISH",
@@ -285,8 +344,7 @@ class ScheduleIfcGenerator:
duration_type="WORKTIME",
)
def create_resources(self):
# print("Resources", self.resources)
def create_resources(self) -> None:
if self.resources:
for id, resource in self.resources.items():
@@ -312,8 +370,7 @@ class ScheduleIfcGenerator:
resource["ifc"] = ifcopenshell.api.run(
"resource.add_resource", self.file, **{"ifc_class": "IfcCrewResource", "name": resource["Name"]}
)
print(self.resources)
def create_boilerplate_ifc(self):
def create_boilerplate_ifc(self) -> None:
self.file = ifcopenshell.file(schema="IFC4")
self.work_plan = self.file.create_entity("IfcWorkPlan")
+31 -34
View File
@@ -6,15 +6,13 @@ print("In module products sys.path[0], __package__ ==", sys.path[0], __package__
sys.path.append(sys.path[0])
import sqlite3
import math
import datetime
from datetime import timedelta
import ifcopenshell
import ifcopenshell.api
import ifcopenshell.util.date
from .wpattern import AstaCalendarWorkPattern
from .common import ScheduleIfcGenerator
from .common import ScheduleIfcGenerator, Calendar, WBSEntry, Activity
import time
from typing import Any
list_of_tables = [
@@ -29,6 +27,9 @@ list_of_tables = [
class PP2Ifc:
wbs: dict[int, WBSEntry]
activities: dict[int, Activity]
def __init__(self):
self.pp = None
@@ -45,17 +46,17 @@ class PP2Ifc:
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: str) -> list[dict[str, Any]]:
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()]
return r
def get_json_with_filter(self, table_name, attr_name, attr_value):
def get_json_with_filter(self, table_name: str, attr_name: str, attr_value: Any) -> list[dict[str, Any]]:
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()]
return r
def execute(self):
def execute(self) -> None:
self.con = sqlite3.connect(self.pp)
self.cur = self.con.cursor()
self.parse_pp()
@@ -78,17 +79,15 @@ class PP2Ifc:
print("Parsing time is", end - start)
print("IFC Creation took", end2 - end)
print("Overall Time", end2 - start)
# self.create_ifc()
def parse_pp(self):
def parse_pp(self) -> None:
project = self.get_json("PROJECT_SUMMARY")[0]
self.project["Name"] = project["SHORT_NAME"]
self.parse_calendar_pp()
self.parse_bar()
self.parse_relationship_pp(project)
self.parse_relationship_pp()
def parse_calendar_pp(self):
def parse_calendar_pp(self) -> None:
calendars = self.get_json("CALENDAR")
wp_data = self.get_json("WORK_PATTERN")
work_types = self.get_json("EXCEPTIONN")
@@ -111,16 +110,15 @@ class PP2Ifc:
)
timex.append(work_times.total_seconds() / (60 * 60))
self.calendars[calendar_id] = {
"Name": calendar["NAME"],
"Type": "NOTDEFINED",
"HoursPerDay": max(timex),
"StandardWorkWeek": wp.dict_wp,
"HolidayOrExceptions": exceptions,
}
# print(self.calendars[calendar_id])
self.calendars[calendar_id] = Calendar(
Name=calendar["NAME"],
Type="NOTDEFINED",
HoursPerDay=max(timex),
StandardWorkWeek=wp.dict_wp,
HolidayOrExceptions=exceptions,
)
def parse_bar(self):
def parse_bar(self) -> None:
bars = self.get_json("BAR")
expanded_tasks = {t["BAR"]: t for t in self.get_json("EXPANDED_TASK")}
tasks = {t["BAR"]: t for t in self.get_json("TASK")}
@@ -129,7 +127,7 @@ class PP2Ifc:
schedule_task = None
bar_tasks = {}
wbs_activities = {}
wbs_activities: dict[int, list[int]] = {}
for bar in bars:
extra_type = None
@@ -175,24 +173,23 @@ class PP2Ifc:
else:
activity_duration = 0.0
self.activities[extra_data["ID"]] = {
"Name": name,
"Identification": extra_data["ID"],
"StartDate": datetime.datetime.fromisoformat(extra_data["LINKABLE_START"]),
"FinishDate": datetime.datetime.fromisoformat(extra_data["LINKABLE_FINISH"]),
"PlannedDuration": activity_duration,
"Status": "PLANNED",
"CalendarObjectId": extra_data["CALENDAR"],
"ifc": None,
}
self.activities[extra_data["ID"]] = Activity(
Name=name,
Identification=bar["ID"],
StartDate=datetime.datetime.fromisoformat(extra_data["LINKABLE_START"]),
FinishDate=datetime.datetime.fromisoformat(extra_data["LINKABLE_FINISH"]),
PlannedDuration=activity_duration,
Status="PLANNED",
CalendarObjectId=extra_data["CALENDAR"],
ifc=None,
)
wbs_activities.setdefault(bar["EXPANDED_TASK"], []).append(extra_data["ID"])
for wbs, activities in wbs_activities.items():
self.wbs[wbs]["activities"] = activities
def parse_relationship_pp(self, project):
def parse_relationship_pp(self) -> None:
relations = self.get_json("LINK")
# print(relations)
for relationship in relations:
predecessor = relationship["START_TASK"]
successor = relationship["END_TASK"]
+110 -35
View File
@@ -8,10 +8,12 @@ for each day as a key there is a list of working times with the format
import re
from datetime import time, datetime
from typing import Any
from ifc4d.common import WorkSlot
class AstaCalendarWorkPattern:
def get_keys(self, s):
def get_keys(self, s: str) -> list[str]:
regex = r"\<(\".+?\")\>(\w|\d|)+?"
matches = re.finditer(regex, s)
@@ -20,30 +22,115 @@ class AstaCalendarWorkPattern:
matcs.append(match.group(1))
return matcs
def get_values(self, s):
rx2 = r"<\"[^<>]+\">"
def get_values(self, s: str) -> list[str]:
rx2 = r"<\"[^<>]+\">" # TODO: unused?
data = re.split("<[^<>]+>", s)
return data
def __init__(self, string, work_type_ids):
self.string = string
self.Days = {
"en": ["Sunday", "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday"],
"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)
# Used the list of Blender supported languages as a reference.
# Method to generate names for other language:
# def get_day_names(lang):
# from babel.dates import get_day_names
# from babel import Locale
# locale = Locale(lang)
# return [i.capitalize() for i in get_day_names(locale=locale, width="wide").values()]
Days = {
"am": ("እሑድ", "ሰኞ", "ማክሰኞ", "ረቡዕ", "ሐሙስ", "ዓርብ", "ቅዳሜ"),
"ar": ("الأحد", "الاثنين", "الثلاثاء", "الأربعاء", "الخميس", "الجمعة", "السبت"),
"be": ("Нядзеля", "Панядзелак", "Аўторак", "Серада", "Чацвер", "Пятніца", "Субота"),
"bg": ("Неделя", "Понеделник", "Вторник", "Сряда", "Четвъртък", "Петък", "Събота"),
"ca": ("Diumenge", "Dilluns", "Dimarts", "Dimecres", "Dijous", "Divendres", "Dissabte"),
"cs": ("Neděle", "Pondělí", "Úterý", "Středa", "Čtvrtek", "Pátek", "Sobota"),
"da": ("Søndag", "Mandag", "Tirsdag", "Onsdag", "Torsdag", "Fredag", "Lørdag"),
"de": ("Sonntag", "Montag", "Dienstag", "Mittwoch", "Donnerstag", "Freitag", "Samstag"),
"el": ("Κυριακή", "Δευτέρα", "Τρίτη", "Τετάρτη", "Πέμπτη", "Παρασκευή", "Σάββατο"),
"en": ("Sunday", "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday"),
"eo": ("Dimanĉo", "Lundo", "Mardo", "Merkredo", "Ĵaŭdo", "Vendredo", "Sabato"),
"es": ("Domingo", "Lunes", "Martes", "Miércoles", "Jueves", "Viernes", "Sábado"),
"et": ("Pühapäev", "Esmaspäev", "Teisipäev", "Kolmapäev", "Neljapäev", "Reede", "Laupäev"),
"eu": ("Igandea", "Astelehena", "Asteartea", "Asteazkena", "Osteguna", "Ostirala", "Larunbata"),
"fa": ("یکشنبه", "دوشنبه", "سه\u200cشنبه", "چهارشنبه", "پنجشنبه", "جمعه", "شنبه"),
"fi": ("Sunnuntaina", "Maanantaina", "Tiistaina", "Keskiviikkona", "Torstaina", "Perjantaina", "Lauantaina"),
"fr": ("Dimanche", "Lundi", "Mardi", "Mercredi", "Jeudi", "Vendredi", "Samedi"),
"ha": ("Lahadi", "Litinin", "Talata", "Laraba", "Alhamis", "Jummaʼa", "Asabar"),
"he": ("יום ראשון", "יום שני", "יום שלישי", "יום רביעי", "יום חמישי", "יום שישי", "יום שבת"),
"hi": ("रविवार", "सोमवार", "मंगलवार", "बुधवार", "गुरुवार", "शुक्रवार", "शनिवार"),
"hr": ("Nedjelja", "Ponedjeljak", "Utorak", "Srijeda", "Četvrtak", "Petak", "Subota"),
"hu": ("Vasárnap", "Hétfő", "Kedd", "Szerda", "Csütörtök", "Péntek", "Szombat"),
"id": ("Minggu", "Senin", "Selasa", "Rabu", "Kamis", "Jumat", "Sabtu"),
"it": ("Domenica", "Lunedì", "Martedì", "Mercoledì", "Giovedì", "Venerdì", "Sabato"),
"ja": ("日曜日", "月曜日", "火曜日", "水曜日", "木曜日", "金曜日", "土曜日"),
"ka": ("კვირა", "ორშაბათი", "სამშაბათი", "ოთხშაბათი", "ხუთშაბათი", "პარასკევი", "შაბათი"),
"kk": ("Жексенбі", "Дүйсенбі", "Сейсенбі", "Сәрсенбі", "Бейсенбі", "Жұма", "Сенбі"),
"km": ("អាទិត្យ", "ច័ន្ទ", "អង្គារ", "ពុធ", "ព្រហស្បតិ៍", "សុក្រ", "សៅរ៍"),
"ko": ("일요일", "월요일", "화요일", "수요일", "목요일", "금요일", "토요일"),
"ky": ("Жекшемби", "Дүйшөмбү", "Шейшемби", "Шаршемби", "Бейшемби", "Жума", "Ишемби"),
"ne": ("आइतबार", "सोमबार", "मङ्गलबार", "बुधबार", "बिहिबार", "शुक्रबार", "शनिबार"),
"nl": ("Zondag", "Maandag", "Dinsdag", "Woensdag", "Donderdag", "Vrijdag", "Zaterdag"),
"pl": ("Niedziela", "Poniedziałek", "Wtorek", "Środa", "Czwartek", "Piątek", "Sobota"),
"pt": ("Domingo", "Segunda-feira", "Terça-feira", "Quarta-feira", "Quinta-feira", "Sexta-feira", "Sábado"),
"ro": ("Duminică", "Luni", "Marți", "Miercuri", "Joi", "Vineri", "Sâmbătă"),
"ru": ("Воскресенье", "Понедельник", "Вторник", "Среда", "Четверг", "Пятница", "Суббота"),
"sk": ("Nedeľa", "Pondelok", "Utorok", "Streda", "Štvrtok", "Piatok", "Sobota"),
"sl": ("Nedelja", "Ponedeljek", "Torek", "Sreda", "Četrtek", "Petek", "Sobota"),
"sr": ("Недеља", "Понедељак", "Уторак", "Среда", "Четвртак", "Петак", "Субота"),
"sv": ("Söndag", "Måndag", "Tisdag", "Onsdag", "Torsdag", "Fredag", "Lördag"),
"sw": ("Jumapili", "Jumatatu", "Jumanne", "Jumatano", "Alhamisi", "Ijumaa", "Jumamosi"),
"ta": ("ஞாயிறு", "திங்கள்", "செவ்வாய்", "புதன்", "வியாழன்", "வெள்ளி", "சனி"),
"th": ("วันอาทิตย์", "วันจันทร์", "วันอังคาร", "วันพุธ", "วันพฤหัสบดี", "วันศุกร์", "วันเสาร์"),
"tr": ("Pazar", "Pazartesi", "Salı", "Çarşamba", "Perşembe", "Cuma", "Cumartesi"),
"uk": ("Неділя", "Понеділок", "Вівторок", "Середа", "Четвер", "Пʼятниця", "Субота"),
"ur": ("اتوار", "پیر", "منگل", "بدھ", "جمعرات", "جمعہ", "ہفتہ"),
"vi": ("Chủ nhật", "Thứ hai", "Thứ ba", "Thứ tư", "Thứ năm", "Thứ sáu", "Thứ bảy"),
}
def get_day_names(self, string: str) -> None:
"""Parse day names from string and translate them to English.
:raises Exception: If language is not yet supported.
"""
# Parse day names.
day_names = [day_name.strip('"') for day_name in self.get_keys(string)]
# Identify language.
day_names_set = set(day_names)
for lang in self.Days:
lang_day_names = self.Days[lang]
if day_names_set.issubset(lang_day_names):
en_day_names = self.Days["en"]
self.lang = lang
# Translate day names to English.
self.day_names = [en_day_names[lang_day_names.index(d)] for d in day_names]
return
msg = (
"Could not identify language for work patterns, need to update Days dictionary."
"Please report it to IfcOpenShell Github issues"
f"\nDay names provided: {day_names}."
"\nAvailable languages:"
)
for lang in self.Days:
msg += f"\n - {lang}: {self.Days[lang]}"
raise Exception(msg)
lang: str
day_names: list[str]
def __init__(self, string: str, work_type_ids: list[int]):
self.get_day_names(string)
self.values = self.get_values(string)
self.dict_wp = []
for d, m in zip(self.values[1:], self.keys):
splt_data = d.strip().split(",")
workhours = []
self.dict_wp: list[WorkSlot] = []
for value, day_name in zip(self.values[1:], self.day_names, strict=True):
splt_data = value.strip().split(",")
workhours: list[dict[str, Any]] = []
if len(splt_data) >= 7:
number_of_work_slots = splt_data[1]
for i in range(int(number_of_work_slots)):
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_2 = datetime.strptime(splt_data[2 + i * 3 + 2].ljust(5, "0"), "%H%M%S")
work_slot_data = splt_data[2 + i * 3 :][:3]
work_type_id, start_time, end_time = work_slot_data
if int(work_type_id) in work_type_ids:
st1_1 = datetime.strptime(start_time.ljust(5, "0"), "%H%M%S")
st1_2 = datetime.strptime(end_time.ljust(5, "0"), "%H%M%S")
st = {
"Start": time(st1_1.hour, st1_1.minute),
"Finish": time(st1_2.hour, st1_2.minute),
@@ -51,21 +138,9 @@ class AstaCalendarWorkPattern:
}
workhours.append(st)
self.dict_wp.append({"DayOfWeek": m.replace('"', ""), "WorkTimes": workhours, "ifc": None})
self.dict_wp.append(WorkSlot(DayOfWeek=day_name, WorkTimes=workhours, ifc=None))
# Translate day names to english
def translate_days(days, wp):
for index, day in enumerate(days["en"]):
for lang in days.keys():
if lang == "en":
continue
if wp["DayOfWeek"] == days[lang][index]:
wp["DayOfWeek"] = days["en"][index]
return
translate_days(self.Days, self.dict_wp[-1])
for day in self.Days["en"]:
if not len(list(filter(lambda d: d["DayOfWeek"] == day, self.dict_wp))) > 0:
print("MISSING", day)
self.dict_wp.append({"DayOfWeek": day, "WorkTimes": [], "ifc": None})
# Add empty work times for missing week days.
missing_week_days = set(self.Days[self.lang]) - set(self.day_names)
for day_name in missing_week_days:
self.dict_wp.append(WorkSlot(DayOfWeek=day_name, WorkTimes=[], ifc=None))