m.stangl/CreoMigrate
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Großes Refactoring - Dekorator für lazy-Initialisierung durch die in Python für diese Aufgabe vorgesehenen Property-Dekoratoren ersetzt. Durch diese Dekoratoren können zukünftig auch einfach Setter-Methoden ergänzt werden. Das Holen/Laden der JSON- und CSV-Dateien wurde zudem nun so umgebaut, dass dies nicht mehr in der Klasse 'MigrationsStatus' erfolgt, sondern in den Klassen 'SivasAssemblyData' und 'RuleDesignerDatabaseConfig', da dies thematisch dort auch hingehört. Die bisherige Vorgehensweise war einerseits verwirrend und zweitens ineffizient, weil in der Klasse 'MigrationsStatus' nochmal Objekte der Klassen 'SivasAssemblyData' und 'RuleDesignerDatabaseConfig' initialisiert werden mussten. Durch die Anpassung wird nun zudem die Datei 'get_teilestamm.py' nicht länger benötigt

Browse Source
This commit is contained in:
Simon Steuer
2025-03-09 00:28:09 +01:00
parent beb1fdc0da
commit 6f188649ff
1 changed files with 147 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/manage_configs.py
+147 -103
View File
@@ -1,6 +1,6 @@
import os import os
import json import json
from configparser import ConfigParser, RawConfigParser from configparser import ConfigParser
import pandas as pd import pandas as pd
import update_database import update_database
@@ -39,153 +39,197 @@ class CSVBasedConfig:
class SivasAssemblyData(JsonBasedConfig): class SivasAssemblyData(JsonBasedConfig):
"""Auslesen der Assembly-Informationen zu einer angegebenen Baugruppennummer""" """Auslesen der Assembly-Informationen zu einer angegebenen Baugruppennummer"""
def __init__(self, config_dir, config_file): def __init__(self, config_dir, config_file, load_local_json=False):
super().__init__(config_dir, config_file) super().__init__(config_dir, config_file)
self.assembly_data = None 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 _lazy_init_assembly_data(func): def create_asm_json(self):
"""Sorgt dafür, dass json-Datei der Baugruppe nur 1x geladen wird""" """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!")
def wrapper(self, *args, **kwargs): @property
if self.assembly_data is None: def asm_data(self) -> dict:
self.assembly_data = self.load_json_from_file() """Liest die Baugruppen-JSON ein und gibt deren Informationen zurück"""
return func(self, *args, **kwargs) if self._asm_data is None:
return wrapper
@_lazy_init_assembly_data if not os.path.isfile(self._config_path):
def get_assembly_ids(self): 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)""" """Gibt alle Teilenummern der Stückliste zurück (auch Teilenummern aus tieferen Baugruppenebenen)"""
assembly_ids = set() if self._child_ids is None:
for child_id, parent_properties in self.assembly_data.items(): self._child_ids = set()
assembly_ids.add(child_id) for child_id, parent_properties in self.asm_data.items():
for parent_property in parent_properties: self._child_ids.add(child_id)
parent_id = parent_property["Parent"] for parent_property in parent_properties:
assembly_ids.add(parent_id) parent_id = parent_property["Parent"]
return assembly_ids self._child_ids.add(parent_id)
return self._child_ids
@_lazy_init_assembly_data @property
def get_parent_children(self) -> dict: def parent_children(self) -> dict:
"""Ordnet Stücklistenpositionen (Kinder) den Baugruppennummern (Eltern) zu. Die Kinder sind entsprechend ihrer Positionsnummern in den Stücklisten sortiert.""" """Ordnet Stücklistenpositionen (Kinder) den Baugruppennummern (Eltern) zu. Die Kinder sind entsprechend ihrer Positionsnummern in den Stücklisten sortiert."""
unordered_parent_children = {} if self._parent_children is None:
unordered_parent_children = {}
for child_id, parent_properties in self.assembly_data.items(): for child_id, parent_properties in self.asm_data.items():
for parent_property in parent_properties: for parent_property in parent_properties:
parent_id = parent_property["Parent"] parent_id = parent_property["Parent"]
child_posnr_in_parent_bom = int(parent_property["Position"].lstrip(".").split("/")[0]) child_posnr_in_parent_bom = int(parent_property["Position"].lstrip(".").split("/")[0])
if parent_id not in unordered_parent_children: if parent_id not in unordered_parent_children:
unordered_parent_children[parent_id] = {} unordered_parent_children[parent_id] = {}
unordered_parent_children[parent_id][child_posnr_in_parent_bom] = child_id unordered_parent_children[parent_id][child_posnr_in_parent_bom] = child_id
ordered_parent_children = {} ordered_parent_children = {}
for parent, children in unordered_parent_children.items(): for parent, children in unordered_parent_children.items():
for pos_nr, child in sorted(children.items()): for pos_nr, child in sorted(children.items()):
if parent not in ordered_parent_children: if parent not in ordered_parent_children:
ordered_parent_children[parent] = [] ordered_parent_children[parent] = []
ordered_parent_children[parent].append(child) ordered_parent_children[parent].append(child)
return ordered_parent_children self._parent_children = ordered_parent_children
return self._parent_children
def get_longest_assembly_id(self) -> str: @property
def longest_child_id(self) -> str:
"""Gibt die längste Teilenummer von allen Stücklistenpositionen zurück""" """Gibt die längste Teilenummer von allen Stücklistenpositionen zurück"""
return max(self.get_assembly_ids()) if self._longest_child_id is None:
self._longest_child_id = max(self.child_ids)
return max(self.child_ids)
@_lazy_init_assembly_data @property
def get_max_bom_depth(self) -> int: def max_bom_depth(self) -> int:
"""Gibt die tiefste Stücklistenebene zurück (0 = höchste Ebene)""" """Gibt die tiefste Stücklistenebene zurück (0 = höchste Ebene)"""
bom_levels = [] if self._max_bom_depth is None:
for parents in self.assembly_data.values(): bom_levels = []
for parent in parents: for parents in self.asm_data.values():
current_bom_level = parent["Position"].count(".") for parent in parents:
bom_levels.append(current_bom_level) current_bom_level = parent["Position"].count(".")
return max(bom_levels) bom_levels.append(current_bom_level)
self._max_bom_depth = max(bom_levels)
return self._max_bom_depth
@_lazy_init_assembly_data @property
def get_property_names(self) -> list: def property_names(self) -> list:
"""Gibt alle Eigenschaftsnamen zurück, die in der Assembly-json-Datei enthalten sind""" """Gibt alle Eigenschaftsnamen zurück, die in der Assembly-json-Datei enthalten sind"""
for parent_data in self.assembly_data.values(): if self._property_names is None:
return list(parent_data[0]) for parent_data in self.asm_data.values():
self._property_names = list(parent_data[0])
return self._property_names
@_lazy_init_assembly_data
def get_bom_property_values(self, property) -> dict: def get_bom_property_values(self, property) -> dict:
"""Gibt eine Eigenschaft zu den Teilenummern der Stückliste zurück (z.B. Bezeichnung)""" """Gibt eine Eigenschaft zu den Teilenummern der Stückliste zurück (z.B. Bezeichnung)"""
if property not in self.get_property_names(): if property not in self.property_names:
raise ValueError("Angegebener Eigenschaftsname exisitert nicht!") raise ValueError("Angegebener Eigenschaftsname exisitert nicht!")
id_property_mapping = {} id_property_mapping = {}
for child, parent_data in self.assembly_data.items(): for child, parent_data in self.asm_data.items():
for properties in parent_data: for properties in parent_data:
if property in properties: if property in properties:
id_property_mapping[child] = properties[property] id_property_mapping[child] = properties[property]
return id_property_mapping return id_property_mapping
@_lazy_init_assembly_data @property
def get_all_bom_properties(self) -> dict: def all_bom_properties(self) -> dict:
"""Gibt alle Eigenschaften zu den Teilenummern der Stückliste zurück (z.B. Bezeichnung)""" """Gibt alle Eigenschaften zu den Teilenummern der Stückliste zurück (z.B. Bezeichnung)"""
all_properties = {} if self._all_bom_properties is None:
property_names = self.get_property_names() self._all_bom_properties = {}
for property_name in property_names: for property_name in self.property_names:
all_properties[property_name] = self.get_bom_property_values(property_name) self._all_bom_properties[property_name] = self.get_bom_property_values(property_name)
return all_properties return self._all_bom_properties
class RuleDesignerDatabaseConfig(CSVBasedConfig): class RuleDesignerDatabaseConfig(CSVBasedConfig):
"""Auslesen der Angaben aus der RuleDesigner-Datenbank-CSV""" """Auslesen der CSV-Datei, in der alle RuleDesigner-Teilenummern aufgelistet sind"""
def __init__(self, config_dir=".", config_file="RD_Teilenummern.csv"): def __init__(self, config_dir=".", config_file="RD_Teilenummern.csv", load_local_csv=False):
super().__init__(config_dir, config_file) super().__init__(config_dir, config_file)
self._load_local_csv = load_local_csv
self._ids = None
def get_ruledesigner_ids(self): def copy_csv_from_server(self):
return tuple(self.load_csv_from_file()['Teilenummern']) """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: class MigrationStatus:
def __init__(self, file_dir, assembly_json): """Klasse gibt Informationen zum aktuellen Migrationsstand zurück, indem geprüft wird, ob die Nummern aus Sivas, auch in RuleDesigner vorliegen"""
self._file_dir = file_dir def __init__(self, asm_child_ids: tuple, rd_ids: tuple):
self._assembly_json = assembly_json self.asm_child_ids = asm_child_ids
self._uselocal = False self.rd_ids = rd_ids
self._overview = None
self._green_ids = None
self._red_ids = None
def use_local(self): @property
self._uselocal = True 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
def get_migration_status(self) -> dict: @property
"""Gibt zurück, ob die Teilenummern der Baugruppe bereits migriert wurden, oder nicht""" def green_ids(self) -> list:
out_dir = self._file_dir """Gibt alle Stücklistenpositionen zurück, die migriert sind"""
if self._uselocal == False:
ok = update_database.get_sivas_dbase(self._assembly_json, out_dir)
if not ok:
raise Exception("no json file could be generated")
update_database.get_rd_dbase()
else:
if not os.path.exists(os.path.join(out_dir, self._assembly_json)):
raise Exception("no local file exists "+ self._assembly_json)
self.sivas_assembly_data = SivasAssemblyData(out_dir, self._assembly_json)
sivas_assembly_ids = self.sivas_assembly_data.get_assembly_ids()
self.ruledesigner_data = RuleDesignerDatabaseConfig(out_dir)
ruledesigner_ids = self.ruledesigner_data.get_ruledesigner_ids()
migration_status = {} if self._green_ids is None:
for sivas_assembly_id in sivas_assembly_ids: self._green_ids = [id for id, status in self.id_status.items() if status is True]
migration_status[sivas_assembly_id] = True if sivas_assembly_id in ruledesigner_ids else False return self._green_ids
return migration_status
def get_migrated_ids(self) -> list: @property
"""Gibt alle Teilenummern aus der Baugruppenstückliste zurück, die bereits in RuleDesigner migriert wurden""" 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
migrated_ids = [assembly_id for assembly_id, migration_status in self.get_migration_status().items() if migration_status is True] @property
return migrated_ids def red_ids(self) -> list:
"""Gibt alle Stücklistenpositionen zurück, die noch nicht migriert sind"""
def get_unmigrated_ids(self) -> list: if self._red_ids is None:
"""Gibt alle Teilenummern aus der Baugruppenstückliste zurück, deren Migration in RuleDesigner noch aussteht""" self._red_ids = [id for id, status in self.id_status.items() if status is False]
return self._red_ids
unmigrated_ids = [assembly_id for assembly_id, migration_status in self.get_migration_status().items() if migration_status is False]
return unmigrated_ids
class TOSGroupsConfig(JsonBasedConfig): class TOSGroupsConfig(JsonBasedConfig):
@@ -525,7 +569,7 @@ class Configs:
self.TOSGruppen = TOSGroupsConfig(cfg_dir) self.TOSGruppen = TOSGroupsConfig(cfg_dir)
self.Materialien = MaterialTransConfig(cfg_dir) self.Materialien = MaterialTransConfig(cfg_dir)
self._TeileNummern = None self._teilenummern = None
self._blacklist = None self._blacklist = None
def get_cfg_path(self): def get_cfg_path(self):
@@ -539,9 +583,9 @@ class Configs:
def get_ruledesigner_ids(self): def get_ruledesigner_ids(self):
if not self._data_dir: if not self._data_dir:
raise Exception("set_data_dir not called. path not set") raise Exception("set_data_dir not called. path not set")
if not self._TeileNummern: if not self._teilenummern:
self._TeileNummern = RuleDesignerDatabaseConfig(self._data_dir).get_ruledesigner_ids() self._teilenummern = RuleDesignerDatabaseConfig(self._data_dir, load_local_csv=True).ids
return self._TeileNummern return self._teilenummern
def get_blacklist(self): def get_blacklist(self):
if not self._data_dir: if not self._data_dir: