Files
CreoMigrate/lib/manage_configs.py
T

600 lines
23 KiB
Python

import os
import json
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, config_file):
"""merke dir den Pfad zu den Configdateien"""
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, config_file):
"""merke dir den Pfad zu den Configdateien"""
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 SivasAssemblyData(JsonBasedConfig):
"""Auslesen der Assembly-Informationen zu einer angegebenen Baugruppennummer"""
def __init__(self, config_dir, config_file, 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_path, os.environ.get("CREMIG_DATA"))
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 os.path.isfile(self._config_path):
raise FileNotFoundError(f"{self._config_path} existiert nicht. Bitte prüfen, ob die Nummer richtig geschrieben wurde.")
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()
for child_id, parent_properties in self.asm_data.items():
self._child_ids.add(child_id)
for parent_property in parent_properties:
parent_id = parent_property["Parent"]
self._child_ids.add(parent_id)
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
@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_bom_property_values(self, property) -> dict:
"""Gibt eine Eigenschaft zu den Teilenummern der Stückliste zurück (z.B. Bezeichnung)"""
if property 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 in properties:
id_property_mapping[child] = properties[property]
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_bom_property_values(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 os.path.isfile(self._config_path):
raise FileNotFoundError(f"{self._config_path} existiert nicht. Bitte Batchdatei unter 'bin/01_get_rd_dbase.bat' ausführen!")
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._overview = None
self._green_ids = None
self._red_ids = 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 green_ids(self) -> list:
"""Gibt alle Stücklistenpositionen zurück, die migriert sind"""
if self._green_ids is None:
self._green_ids = [id for id, status in self.id_status.items() if status is True]
return self._green_ids
@property
def yellow_ids(self) -> list:
"""Gibt migrierte Stücklistenpositionen zurück, die aktuell eine Freigabe der Baugruppe verhindern, da sie sich derzeit in Arbeit befinden"""
pass
@property
def red_ids(self) -> list:
"""Gibt alle Stücklistenpositionen zurück, die noch nicht migriert sind"""
if self._red_ids is None:
self._red_ids = [id for id, status in self.id_status.items() if status is False]
return self._red_ids
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