diff --git a/src/ifc4d/ifc4d/common.py b/src/ifc4d/ifc4d/common.py index 96ffe7a0af..13a4f4beea 100644 --- a/src/ifc4d/ifc4d/common.py +++ b/src/ifc4d/ifc4d/common.py @@ -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") diff --git a/src/ifc4d/ifc4d/pp2ifc.py b/src/ifc4d/ifc4d/pp2ifc.py index b0faddc22d..408ab32a9c 100644 --- a/src/ifc4d/ifc4d/pp2ifc.py +++ b/src/ifc4d/ifc4d/pp2ifc.py @@ -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"] diff --git a/src/ifc4d/ifc4d/wpattern.py b/src/ifc4d/ifc4d/wpattern.py index 304f03777a..72a150aea9 100644 --- a/src/ifc4d/ifc4d/wpattern.py +++ b/src/ifc4d/ifc4d/wpattern.py @@ -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))