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""" with open(self._config_path, 'r', encoding='utf-8') as fh: content = fh.read() return json.loads(content) 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 SivasBOM(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._bom_data = None self._child_ids = None self._parent_children = None self._longest_child_id = None self._max_bom_depth = None self._all_bom_properties = None def create_asm_json(self) -> bool: """Erstellt JSON mit den aktuellen Baugruppeninformationen""" update_database.get_sivas_bom(self._config_file.removesuffix(".json")) @property def bom_data(self) -> dict: """ Liest die JSON-Datei der Baugruppenstückliste ein und speichert die Informationen als dict. """ if self._bom_data is None: """ Falls die JSON nicht existiert oder nicht generiert werden kann, soll die Möglichkeit bestehen, diese eigenständig zu erweitern, da in der JSON z.B. die Nummer der obersten Baugruppe standardmäßig nicht vorliegt. Generierungs- und Einlese-Fehler werden daher bewusst ignoriert, um stattdessen ein leeres dict zurückzugeben, das anschließend erweitert werden kann. Die Möglichkeit zur Erweiterung der Stückliste ermöglicht es zudem, Teilenummern ohne Stückliste abzubilden. Das funktioniert, indem man die Teilenummer ohne eigene Stückliste als Kind einer fiktiven Baugruppe "None" ansieht / definiert. Dadurch ist keine separate Logik erforderlich, die sich nur mit Teilenummern ohne Stückliste befasst. """ self._bom_data = {} if not self._load_local_json: try: self.create_asm_json() except ValueError: pass try: self._bom_data = self.load_json_from_file() except FileNotFoundError: pass return self._bom_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.bom_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.bom_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": [{...}, {...} ] }""" if id not in self.bom_data: self._bom_data[id] = [] self._bom_data[id].append(metadata) return self._bom_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.bom_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 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._bom_data: raise ValueError("Teilenummer befindet sich nicht in der Baugruppe!") if property_name not in self._bom_data[id][0]: return return self._bom_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)""" id_property_mapping = {} for child_id, asm_usages in self.bom_data.items(): child_properties = asm_usages[0] id_property_mapping[child_id] = None if property_name not in child_properties else 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 child_id, asm_usages in self.bom_data.items(): child_properties = asm_usages[0] for property_name, property_val in child_properties.items(): if property_name not in self._all_bom_properties: self._all_bom_properties[property_name] = {} self._all_bom_properties[property_name][child_id] = property_val 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_ruledesigner_ids() 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 SivasTeilestamm(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._data = None @staticmethod def create_csvs(ids): """Erstellt CSV-Datei(en) mit den Teilestamminformationen zu der / den angegebenen Teilenummer(n)""" update_database.get_sivas_teilestamm(ids) @property def teilestamm_data(self): """Gibt CSV-Daten als Dataframe zurück""" if self._data is None: if not self._load_local_csv: self.create_csvs(self._config_file.removesuffix(".csv")) self._data = self.load_csv_from_file() return self._data def get_property_values(self, property_name) -> str: """Gibt Wert aus der CSV für den angegebenen Spaltennamen zurück""" ret = self.teilestamm_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