import os import json import pymssql from configparser import ConfigParser import pandas as pd import update_database """ Diese Lib verwaltet alle Config dateien (als .cfg oder als .json) von CreoMigrate """ class JsonBasedConfig: """Klasse mit generellen Methoden für alle Config Objekte mit json files""" def __init__(self, config_dir: str, config_file: str): """merke dir den Pfad zu den Configdateien""" self._config_dir = config_dir self._config_file = config_file self._config_path = os.path.join(config_dir, config_file) def load_json_from_file(self): """lade das json file vom gegeben Pfad in ein json Objekt""" try: with open(self._config_path, 'r', encoding='utf-8') as fh: content = fh.read() return json.loads(content) except FileNotFoundError: raise ValueError("Bitte prüfen, ob korrekte Teilenummer, bzw. korrekter Dateiname angegeben wurde.") def from_json(self, json_str): """nimm den Json String zur Initialisierung des Objektes""" return json.loads(json_str) class CSVBasedConfig: def __init__(self, config_dir: str, config_file: str): """merke dir den Pfad zu den Configdateien""" self._config_dir = config_dir self._config_file = config_file self._config_path = os.path.join(config_dir, config_file) def load_csv_from_file(self): """lade die csv aus dem angegebenen Dateipfad in ein Dataframe""" return pd.read_csv(self._config_path, sep=";", dtype="object") class SQLBasedConfig: def __init__(self, server, user, pw, dbase): self._server = server self._user = user self._pw = pw self._dbase = dbase def query_db(self, sql_statement) -> list: """Führt Datenbankabfrage aus und gibt Daten zurück""" with pymssql.connect(self._server, self._user, self._pw, self._dbase) as conn: with conn.cursor(as_dict=True) as cursor: try: cursor.execute(sql_statement) except pymssql.ProgrammingError: print("SQL-Statement konnte nicht verarbeitet werden!") raise ret = cursor.fetchall() return ret class RuleDesignerSQLData(SQLBasedConfig): def __init__(self, server=os.environ.get("RD_SQL_DATABASE_SERVER"), user=os.environ.get("RD_SQL_DATABASE_USER"), pw=os.environ.get("RD_SQL_DATABASE_PW"), dbase=os.environ.get("RD_SQL_DATABASE_NAME")): super().__init__(server, user, pw, dbase) self._rd_statuses = None @property def rd_statuses(self) -> dict: """Gibt Teilenummer und zugehörigen Status zurück""" if self._rd_statuses is None: # Datenbankabfrage aller Teilenummern (codCodice) und der zugehörigen Status (codStatus) sql_statement = "SELECT [codCodice] AS ID, [codStatus] AS STATUS FROM [FUSION_PDM].[dbo].[CODICI]" sql_data = self.query_db(sql_statement) # Jeder Teilenummer den sprechenden Stsatusnamen, statt der internen Statusnummer zuweisen # Dafür laden der Mapping-Datei für die Statusnummern (z.B. 3 = "FREIGEGEBEN") rd_status_cfg = IniBasedConfig(os.environ.get("CREMIG_CFG"), "RuleDesignerStatusnamen.cfg") rd_status_sections = rd_status_cfg.sections() rd_status_names = rd_status_cfg.get_dict(rd_status_sections[0]) self._rd_statuses = {} for id_status_dict in sql_data: id = id_status_dict["ID"] status = id_status_dict["STATUS"] self._rd_statuses[id] = rd_status_names[str(status)] return self._rd_statuses class SivasAssemblyData(JsonBasedConfig): """Auslesen der Assembly-Informationen zu einer angegebenen Baugruppennummer""" def __init__(self, config_dir: str, config_file: str, load_local_json=False): super().__init__(config_dir, config_file) self._load_local_json = load_local_json self._asm_data = None self._child_ids = None self._parent_children = None self._longest_child_id = None self._max_bom_depth = None self._property_names = None self._all_bom_properties = None def create_asm_json(self): """Erstellt JSON mit den aktuellen Baugruppeninformationen""" try: update_database.get_sivas_dbase(self._config_file.removesuffix(".json")) except: raise ValueError("JSON-Datei mit Baugruppeninformationen konnte nicht erstellt werden!") @property def asm_data(self) -> dict: """Liest die Baugruppen-JSON ein und gibt deren Informationen zurück""" if self._asm_data is None: if not self._load_local_json: self.create_asm_json() self._asm_data = self.load_json_from_file() return self._asm_data @property def child_ids(self) -> set: """Gibt alle Teilenummern der Stückliste zurück (auch Teilenummern aus tieferen Baugruppenebenen)""" if self._child_ids is None: self._child_ids = set(self.asm_data.keys()) return self._child_ids @property def parent_children(self) -> dict: """Ordnet Stücklistenpositionen (Kinder) den Baugruppennummern (Eltern) zu. Die Kinder sind entsprechend ihrer Positionsnummern in den Stücklisten sortiert.""" if self._parent_children is None: unordered_parent_children = {} for child_id, parent_properties in self.asm_data.items(): for parent_property in parent_properties: parent_id = parent_property["Parent"] child_posnr_in_parent_bom = int(parent_property["Position"].lstrip(".").split("/")[0]) if parent_id not in unordered_parent_children: unordered_parent_children[parent_id] = {} unordered_parent_children[parent_id][child_posnr_in_parent_bom] = child_id ordered_parent_children = {} for parent, children in unordered_parent_children.items(): for pos_nr, child in sorted(children.items()): if parent not in ordered_parent_children: ordered_parent_children[parent] = [] ordered_parent_children[parent].append(child) self._parent_children = ordered_parent_children return self._parent_children def extend_asm_bom(self, id, metadata:dict) -> dict: """Ergänzt die Stückliste um eine weitere Teilenummer unter Berücksichtigung des vorgegebenen Formats der ausgegebenen JSON des Programms 'Sivas2Json' von Hr. Jakob Aktuell ist die JSON wie folgt aufgebaut: { "Stücklistenposition 1": [{Metadaten aus 1. SIVAS-Stammstückliste, in der die Stücklistenposition verbaut ist}, {Metadaten aus 2. SIVAS-Stammstückliste, in der die Stücklistenposition verbaut ist (falls die oberste Baugruppe weitere Unterbaugruppen enthält, in denen diese Nummer ebenfalls verbaut ist)}, ... ] "Stücklistenposition 2": [{...}, {...} ] }""" property_names = self.property_names asm_data = self.asm_data if id not in asm_data: self._asm_data[id] = [] for property_name in property_names: if property_name not in metadata: metadata[property_name] = None self._asm_data[id].append(metadata) return self._asm_data @property def longest_child_id(self) -> str: """Gibt die längste Teilenummer von allen Stücklistenpositionen zurück""" if self._longest_child_id is None: self._longest_child_id = max(self.child_ids) return max(self.child_ids) @property def max_bom_depth(self) -> int: """Gibt die tiefste Stücklistenebene zurück (0 = höchste Ebene)""" if self._max_bom_depth is None: bom_levels = [] for parents in self.asm_data.values(): for parent in parents: current_bom_level = parent["Position"].count(".") bom_levels.append(current_bom_level) self._max_bom_depth = max(bom_levels) return self._max_bom_depth @property def property_names(self) -> list: """Gibt alle Eigenschaftsnamen zurück, die in der Assembly-json-Datei enthalten sind""" if self._property_names is None: for parent_data in self.asm_data.values(): self._property_names = list(parent_data[0]) return self._property_names def get_prop_val_for_single_id(self, id:str, property_name:str) -> str: """Gibt eine Eigenschaft von einer einzelnen Teilenummer aus der Stückliste zurück""" if id not in self._asm_data: raise KeyError("Angegebene Teilenummer befindet sich nicht in der Baugruppe!") if property_name not in self._asm_data[id][0]: raise ValueError(f"Teilenummer besitzt nicht die angegebene Eigenschaft '{property_name}'") return self._asm_data[id][0][property_name] def get_prop_val_for_all_ids(self, property_name) -> dict: """Gibt eine Eigenschaft zu allen Teilenummern der Stückliste zurück (z.B. Bezeichnung)""" if property_name not in self.property_names: raise ValueError("Angegebener Eigenschaftsname exisitert nicht!") id_property_mapping = {} for child, parent_data in self.asm_data.items(): for properties in parent_data: if property_name in properties: id_property_mapping[child] = properties[property_name] return id_property_mapping @property def all_bom_properties(self) -> dict: """Gibt alle Eigenschaften zu den Teilenummern der Stückliste zurück (z.B. Bezeichnung)""" if self._all_bom_properties is None: self._all_bom_properties = {} for property_name in self.property_names: self._all_bom_properties[property_name] = self.get_prop_val_for_all_ids(property_name) return self._all_bom_properties class RuleDesignerDatabaseConfig(CSVBasedConfig): """Auslesen der CSV-Datei, in der alle RuleDesigner-Teilenummern aufgelistet sind""" def __init__(self, config_dir=".", config_file="RD_Teilenummern.csv", load_local_csv=False): super().__init__(config_dir, config_file) self._load_local_csv = load_local_csv self._ids = None def copy_csv_from_server(self): """Kopiert CSV-Datei vom Server, in der alle Teilenummern, die sich in RuleDesigner befinden""" try: update_database.get_rd_dbase() except: raise ValueError("CSV mit RuleDesigner-Teilenummern konnte nicht ins lokale Verzeichnis kopiert werden!") @property def ids(self): if self._ids is None: if not self._load_local_csv: self.copy_csv_from_server() self._ids = tuple(self.load_csv_from_file()['Teilenummern']) return self._ids class MigrationStatus: """Klasse gibt Informationen zum aktuellen Migrationsstand zurück, indem geprüft wird, ob die Nummern aus Sivas, auch in RuleDesigner vorliegen""" def __init__(self, asm_child_ids: tuple, rd_ids: tuple): self.asm_child_ids = asm_child_ids self.rd_ids = rd_ids self._rd_statuses = None self._overview = None self._ids_in_rd = None self._ids_in_progress = None self._ids_pending = None @property def id_status(self) -> dict: """Gibt zurück, ob die Stücklistenpositionen der Baugruppe migriert sind, oder nicht""" if self._overview is None: self._overview = {} for child_id in self.asm_child_ids: self._overview[child_id] = True if child_id in self.rd_ids else False return self._overview @property def ids_in_rd(self) -> list: """Gibt alle Stücklistenpositionen zurück, die migriert sind""" if self._ids_in_rd is None: self._ids_in_rd = [id for id, status in self.id_status.items() if status is True] return self._ids_in_rd @property def ids_in_progress(self) -> list: """Gibt migrierte Stücklistenpositionen zurück, die aktuell eine Freigabe der Baugruppe verhindern, da sie sich in Arbeit befinden""" if self._ids_in_progress is None: rd_sql_data = RuleDesignerSQLData() rd_statuses = rd_sql_data.rd_statuses self._ids_in_progress = [] for id in self.ids_in_rd: if rd_statuses[id] != "FREIGEGEBEN": self._ids_in_progress.append(id) return self._ids_in_progress @property def ids_pending(self) -> list: """Gibt alle Stücklistenpositionen zurück, die noch nicht migriert sind""" if self._ids_pending is None: self._ids_pending = [id for id, status in self.id_status.items() if status is False] return self._ids_pending class TeilestammMetadata(CSVBasedConfig): def __init__(self, config_dir, config_file, load_local_csv=False): super().__init__(config_dir, config_file) self._load_local_csv = load_local_csv self._csv_data = None @staticmethod def create_csvs(ids): """Erstellt CSV-Datei(en) mit den Teilestamminformationen zu der / den angegebenen Teilenummer(n)""" try: update_database.get_sivas_teilestamm(ids) except: raise ValueError("CSV(s) konnte nicht generiert werden!") @property def csv_data(self): """Gibt CSV-Daten als Dataframe zurück""" if self._csv_data is None: if not self._load_local_csv: self.create_csvs(self._config_file.removesuffix(".csv")) self._csv_data = self.load_csv_from_file() return self._csv_data def get_property_values(self, property_name) -> str: """Gibt Wert aus der CSV für den angegebenen Spaltennamen zurück""" ret = self.csv_data[property_name][0] return ret class TOSGroupsConfig(JsonBasedConfig): """Auslesen der Config Angaben zu den TOS Gruppen""" def __init__(self, config_dir='.', config_file='TOS-Gruppen.json'): self.tos_hierarchy = None super().__init__(config_dir, config_file) def get_children_keys(self, tos_hierarchy, currentId, result): # wenn eine Referenz nicht weiter in der TOS Hierarchy vorkommt, dann nicht mit einsammeln if currentId not in tos_hierarchy: return None childIds = tos_hierarchy[currentId]["reference-ids"].values() level = tos_hierarchy[currentId]["TOS-level"] if len(childIds) > 0: for cId in childIds: # prüfen, ob die Referenz sonst noch verwendet wird if cId in tos_hierarchy: result[cId] = level for cId in childIds: self.get_children_keys(tos_hierarchy, cId, result) def get_Id2TosGroup(self): "hole alle ids zur jeweiligen Gruppe" # lazy. nur einmal laden, wenn noch nicht geschehen if not self.tos_hierarchy: self.tos_hierarchy = super().load_json_from_file() tos_hierarchy = self.tos_hierarchy num2refId = dict() num2child = dict() allkwds = self.get_all_keywords() for kw in allkwds.keys(): refids = tos_hierarchy[kw]["reference-ids"] for cId in refids.keys(): num2refId[cId] = refids[cId] children = tos_hierarchy[kw]["children"] for cId in children.keys(): num2child[cId] = children[cId] return (num2refId, num2child) def get_all_keywords(self): """hole alle TOS Keywords, welche in der config Datei definiert sind""" # lazy. nur einmal laden, wenn noch nicht geschehen if not self.tos_hierarchy: self.tos_hierarchy = super().load_json_from_file() # beginne mit root als ersten, vordefiniertem Keyword rootid = self.tos_hierarchy["root"] # und von dort aus folge dann allen Referenzen result = dict() # root als Nullte Ebene dazu result[rootid] = 0 self.get_children_keys(self.tos_hierarchy, rootid, result) return result def get_keywords_of_level(self, allKeywords, levelnum): """hole alle TOS Keywords der gegebenen Hierarchiestufe "levelnum" """ ret = list() for kw in allKeywords: if allKeywords[kw] == levelnum: ret.append(kw) return ret def from_json(self, json_str): """lade den Json String zur Initialisierung des Objektes""" self.tos_hierarchy = super().from_json(json_str) class MaterialTransConfig(JsonBasedConfig): """Auslesen der Config zu allen Materialangaben""" def __init__(self, config_dir='.', config_file='Materialübersetzung.json'): self._materials = None super().__init__(config_dir, config_file) def from_json(self, json_str): """lade den Json String zur Initialisierung des Objektes""" self._materials = super().from_json(json_str) def get_all_materials(self): """hole alle Materialien, welche in der config Datei definiert sind""" if not self._materials: self._materials = super().load_json_from_file() ret = list() for k in self._materials.keys(): ret.append(k) return ret def get_material_of_type(self, mat): """hole alle Materialien, welche zu diesem Werkstofftyp existieren""" if not self._materials: self.load_materials() return self._materials[mat] class IniBasedConfig: """Klasse mit generellen Methoden für alle Config Objekte mit .cfg/.ini files""" def __init__(self, config_dir, config_file): """merke dir den Pfad zu den Configdateien""" self.config_path = os.path.join(config_dir, config_file) if not os.path.exists(self.config_path): raise Exception("path does not exist") self._isInitalized = False self._config = ConfigParser(allow_no_value=True, delimiters=("=")) self._config.optionxform = lambda option: option # preserve case for letters def from_cfg_str(self, cfg_str): """nimm den config String zur Initialisierung des Objektes""" self._config.read_string(cfg_str) self._isInitalized = True def load_cfg_from_file(self): """lade das config file vom gegeben Pfad""" self._config.read(self.config_path, encoding='utf-8') self._isInitalized = True def get_section_options(self, section_name): if not self._isInitalized: self.load_cfg_from_file() return self._config.options(section_name) def sections(self): if not self._isInitalized: self.load_cfg_from_file() return self._config.sections() def get(self, section_name, option): if not self._isInitalized: self.load_cfg_from_file() return self._config.get(section_name, option) def get_int(self, section_name, option): if not self._isInitalized: self.load_cfg_from_file() return self._config.getint(section_name, option) def get_dict(self, section_name): if not self._isInitalized: self.load_cfg_from_file() return dict(self._config.items(section_name)) class AllgemeinConfig(IniBasedConfig): """Auslesen der Config für alle allgemeinen Angaben""" def __init__(self, config_dir='.', config_file='Allgemein.cfg'): super().__init__(config_dir, config_file) def get_sections(self): """ gibt alle Sektionen zurück wie Kategorien, Dispogruppe, Dispolaufgruppe, EinheitAusw, ... """ return super().sections() def get_options(self, section_name): """ get all options below""" return super().get_section_options(section_name) class BezeichnerConfig(IniBasedConfig): """Auslesen der Config zu allen möglichen Bezeichner""" def __init__(self, config_dir='.', config_file='Bezeichner.cfg'): super().__init__(config_dir, config_file) def get_all(self): """hole alle möglichen Bezeichner """ return super().get_section_options("Bezeichner") class FarbenConfig(IniBasedConfig): """Auslesen der Config zu allen Farbensarten""" def __init__(self, config_dir='.', config_file='Farben.cfg'): super().__init__(config_dir, config_file) def get_farbauswahl(self): """hole alle Farben die möglich sind -> [RAL 1004 Goldgelb, RAL 2000 Gelborange, RAL 3002 Kaminrot, RAL 5010 Enzianblau] """ return super().get_section_options("Farbauswahl") def get_farben_eloxieren(self): """hole alle Farben fürs eloxieren -> [Rot, Grün, Blau, Gelb, Schwarz, Violett, Natur] """ return super().get_section_options("FarbeEloxieren") class FertigungsartConfig(IniBasedConfig): """Auslesen der Config zu allen Fertigungsarten""" def __init__(self, config_dir='.', config_file='Fertigungsarten.cfg'): super().__init__(config_dir, config_file) def get_fertigungsarten(self): """hole alle Fertigungsarten, welche als Sektionen in der config gegeben sind, also Fertigungsart, Allgemeintoleranz3DDruck, AllgemeintoleranzGiessen, AllgemeintoleranzExtrudieren, ... """ return super().sections() def get_list_in_section(self, section): """hole alle möglichen Oberflächen Behandlungsarten zum gegebenen Material Allgemeintoleranz3DDruck ftype -> [ISO 2768:1991-fH, ISO 2768:1991-fK, ISO 2768:1991-fL, ISO 2768:1991-mH, , ...] """ return super().get_section_options(section) class HerstellerConfig(IniBasedConfig): """Auslesen der Config zu allen möglichen Hersteller""" def __init__(self, config_dir='.', config_file='Hersteller.cfg'): super().__init__(config_dir, config_file) def get_all(self): """hole alle möglichen Bezeichner """ return super().get_section_options("Hersteller") class NormenConfig(IniBasedConfig): """Auslesen der Config zu allen Normen""" def __init__(self, config_dir='.', config_file='Normen.cfg'): super().__init__(config_dir, config_file) def get_normen(self): """hole alle Normen, welche als Sektionen in der config gegeben sind, also Oberflaechennorm, Kantennorm, Massnorm, Tolerierungsnorm,.. """ return super().sections() def get_normen_of_type(self, otype): """hole alle möglichen Oberflächen Behandlungsarten zum gegebenen Material Oberfläche otype Oberflaechennorm -> [ISO 1302:2002, ISO 21920] Tolerierungsnorm, ... """ return super().get_section_options(otype) class OberflächeConfig(IniBasedConfig): """Auslesen der Config zu allen Oberflächen""" def __init__(self, config_dir='.', config_file='Oberflächenbehandlung.cfg'): super().__init__(config_dir, config_file) def get_oberflächen(self): """hole alle Oberflächen, welche als Sektionen in der config gegeben sind, also OberflächenbehandlungStahl, OberflächenbehandlungAluminium, ... """ return super().sections() def get_oberflächen_behandlungen(self): """hole alle möglichen Arten von Oberflächenbehandlungen""" return super().get_section_options('Oberflaechenbehandlungen') def get_oberfläche_of_type(self, otype): """hole alle möglichen Oberflächen Behandlungsarten zum gegebenen Material Oberfläche otype OberflächenbehandlungStahl --> [lackieren, pulverbeschichten, verzinken, vernickeln, brünieren, härten, bürsten, sandstrahlen] OberflächenbehandlungAluminium ... """ return super().get_section_options(otype) class WerkstoffeConfig(IniBasedConfig): """Auslesen der Config zu allen Werkstoffen""" def __init__(self, config_dir='.', config_file='Werkstoff.cfg'): super().__init__(config_dir, config_file) def get_werkstoff_namen(self): """hole alle Werkstoffe, welche als Sektionen in der config gegeben sind, also z.B. Stahl, Aluminium, Kunst-/ Verbundstoff, ... """ return super().get_section_options('Werkstofftyp') def get_all(self): ret = list() l = self.get_werkstoff_namen() for item in l: ltemp = self.get_werkstoffe_of_name(l) ret.append(ltemp) return ret def get_werkstoffe_of_name(self, wName): """hole alle Werkstoffe unter der gegebenen Sektion Werkstofftyp wName Stahl --> [ 1.0035 (S185), 1.0038 (S235JR), 1.0122 (S235JRC), 1.0122+C (S235JRC+C), 1.0330 (DC01), ... ] Aluminium, ... """ return super().get_section_options(wName) class PfadConfig(IniBasedConfig): """Auslesen der Config zu allen Pfaden""" def __init__(self, config_dir='.', config_file='Zielpfade.cfg'): super().__init__(config_dir, config_file) class ZusatzConfig(IniBasedConfig): """Auslesen der Zusatzinformationen die aus dem Änderungsindex abgeleitet werden""" def __init__(self, config_dir='.', config_file='Zusatzinformationen.cfg'): super().__init__(config_dir, config_file) class InvaldCharsConfig(): """Auslesen der Zusatzinformationen die aus dem Änderungsindex abgeleitet werden""" def __init__(self, config_dir='.', config_file='InvalidCharacters.txt'): self.config_path = os.path.join(config_dir, config_file) if not os.path.exists(self.config_path): raise Exception("path does not exist") self._isInitalized = False file = open(self.config_path,'r') line = file.read() line = line.rstrip() self._characters = line.split(',') def get(self): return self._characters class GuiConfig(IniBasedConfig): """Auslesen der Zusatzinformationen die aus dem Änderungsindex abgeleitet werden""" def __init__(self, config_dir='.', config_file='gui.cfg'): username = os.environ.get("USERNAME") spezific_file = 'gui-' + username + '.cfg' spezific_path = os.path.join(config_dir,spezific_file) if os.path.exists(spezific_path): config_file=spezific_file super().__init__(config_dir, config_file=config_file) class Configs: """Sammlung aller Configdateien """ def __init__(self, cfg_dir): self._cfg_dir = cfg_dir self.Allgemein = AllgemeinConfig(cfg_dir) self.Bezeichner = BezeichnerConfig(cfg_dir) self.Farben = FarbenConfig(cfg_dir) self.Fertigung = FertigungsartConfig(cfg_dir) self.Hersteller = HerstellerConfig(cfg_dir) self.Normen = NormenConfig(cfg_dir) self.Oberfläche = OberflächeConfig(cfg_dir) self.Pfade = PfadConfig(cfg_dir) self.Werkstoffe = WerkstoffeConfig(cfg_dir) self.Zusatz = ZusatzConfig(cfg_dir) self.InvalidChars = InvaldCharsConfig(cfg_dir) self.Gui = GuiConfig(cfg_dir) self.TOSGruppen = TOSGroupsConfig(cfg_dir) self.Materialien = MaterialTransConfig(cfg_dir) self._teilenummern = None self._blacklist = None def get_cfg_path(self): return self._cfg_dir def set_data_dir(self, data_dir): self._data_dir = data_dir def get_data_dir(self): return self._data_dir def get_ruledesigner_ids(self): if not self._data_dir: raise Exception("set_data_dir not called. path not set") if not self._teilenummern: self._teilenummern = RuleDesignerDatabaseConfig(self._data_dir, load_local_csv=True).ids return self._teilenummern def get_blacklist(self): if not self._data_dir: raise Exception("set_data_dir not called. path not set") if not self._blacklist: with open(os.path.join(self._data_dir, 'blacklist.txt')) as f: lines = f.read().splitlines() self._blacklist = lines return self._blacklist