Aufgaben in mehrere Klassen aufgeteilt + Treeview-Generierung korrigiert (an neues json-Format von Hr. Jakob angepasst) + Treeview wird nun rekursiv generiert, da dessen Erstellung dadurch verstaendlicher ist + Excel-Export korrigiert (ebenfalls an neues json-Format von Hr. Jakob angepasst).

This commit is contained in:
2024-01-08 11:07:16 +01:00
parent 3ad27bf084
commit 099c44395f
+369 -169
View File
@@ -1,4 +1,5 @@
import update_database
import argparse
import os
import json
@@ -12,98 +13,247 @@ Dieses Programm kann:
- die oben beschriebene Darstellung in einer Excel-Datei auch als Treeview auf der Kommandozeile darstellen
"""
class Compare:
class Environment:
lib_dir = os.environ.get('CREMIG_LIB')
data_dir = os.environ.get('CREMIG_DATA')
out_dir = os.environ.get('CREMIG_WORK')
ruledesigner_csv_filename = os.environ.get('RD_DATABASE_NAME')
ruledesigner_csv_filepath = os.path.join(data_dir, ruledesigner_csv_filename)
vba_macro_filename = os.environ.get('VBA_MACRO_NAME')
vba_macro_filepath = os.path.join(lib_dir, vba_macro_filename)
def __init__(self, main_sivas_id):
self.main_sivas_id = main_sivas_id
self.sivas_json_path = os.path.join(Compare.data_dir, main_sivas_id + ".json")
self.sivas_ids = self.read_sivas_ids()
self.ruledesigner_ids = self.read_ruledesigner_ids()
def __init__(self):
self.lib_dir = os.environ.get('CREMIG_LIB')
self.data_dir = os.environ.get('CREMIG_DATA')
self.out_dir = os.environ.get('CREMIG_WORK')
self.ruledesigner_csv_filename = os.environ.get('RD_DATABASE_NAME')
self.vba_macro_filename = os.environ.get('VBA_MACRO_NAME')
self.vba_macro_filepath = os.path.join(self.lib_dir, self.vba_macro_filename)
def read_sivas_ids(self):
class Partnumbers:
# Metohde liest die JSON-Datei ein, in der alle Sivas-Teilenummern aufgelistet sind, die zusammen mit der Hauptnummer migriert werden müssen
def __init__(self, filename: str, file_dir: str):
self.file_dir = file_dir
self.filename = filename.rsplit(".")[0]
def load_from_json(self):
"""Reads partnumbers from a json-file"""
config_path = os.path.join(self.file_dir, self.filename + ".json")
try:
with open(self.sivas_json_path, encoding='utf-8') as json_file:
with open(config_path, encoding='utf-8') as json_file:
return json.load(json_file)
except FileNotFoundError:
# JSON-Datei wurde noch nicht generiert
update_database.get_sivas_dbase(self.main_sivas_id, Compare.data_dir)
return self.read_sivas_ids()
update_database.get_sivas_dbase(self.filename, self.file_dir)
return self.load_from_json()
def load_from_csv(self):
def read_ruledesigner_ids(self):
"""Reads partnumbers from a csv-file"""
# Methode liest die CSV-Datei ein, in der alle RuleDesigner-Nummern aufgelistet sind
config_path = os.path.join(self.file_dir, self.filename + ".csv")
try:
df = pd.read_csv(Compare.ruledesigner_csv_filepath, sep=';', converters={"Teilenummern":str})
df = pd.read_csv(config_path, sep=';', converters={"Teilenummern":str})
return tuple(df["Teilenummern"])
except FileNotFoundError:
# CSV-Datei wurde noch nicht ins %CREMIG_DATA% kopiert
update_database.get_rd_dbase(Compare.data_dir)
return self.read_ruledesigner_ids()
update_database.get_rd_dbase(self.file_dir)
return self.load_from_csv()
def check_migration_status(self):
class SivasPartnumbers(Partnumbers):
# Methode prüft, ob die Sivas-Nummern bereits migriert wurden
def __init__(self, main_sivas_id: str, file_dir: str):
self.file_dir = file_dir
self.main_sivas_id = main_sivas_id
self._partnumbers = None
migrated = {}
for id in self.sivas_ids:
migrated[id] = True if id in self.ruledesigner_ids else False
return migrated
super().__init__(file_dir=self.file_dir, filename=self.main_sivas_id)
def create_sorting_string(self, id):
def lazy_partnumber_init(function):
"""Decorator prevents time-consuming initialization of sivas partnumbers if method doesn't require them"""
def wrapper(self):
if self._partnumbers is None:
self._partnumbers = self.__class__.mro()[1].load_from_json(self) # Using super() isn't possible here
return function(self)
return wrapper
@lazy_partnumber_init
def get_partnumbers(self) -> set:
"""Returns all partnumbers (id's) of the provided json-file"""
ids = set()
for child_id, top_level_assemblies_of_child in self._partnumbers.items():
ids.add(child_id)
for top_level_assembly in top_level_assemblies_of_child:
top_level_assembly_id = top_level_assembly["Parent"]
ids.add(top_level_assembly_id)
# Methode setzt rekursiv einen String zusammen, der anzeigt, wo sich eine Teilenummer in der Hierarchie der obersten Baugruppe befindet.
# Beispiel:
# Die Teilenummer "123456" ist das Halbzeug der Teilenummer "234567". "234567" ist wiederum in der Baugruppe "345678" eingebaut.
# Die Methode gibt dann folgenden String zurück: 345678-234567-123456
return ids
parent = self.sivas_ids[id][0]['Parent']
if parent not in self.sivas_ids:
return f"{parent}-{id}"
recur_val = self.create_sorting_string(parent)
return f"{recur_val}-{id}"
def get_longest_partnumber(self) -> str:
"""Returns the longest partnumber in the provided json-file"""
return max(self.get_partnumbers())
@lazy_partnumber_init
def get_ordered_parent_children(self, no_duplicates: bool = False) -> dict:
"""Replicates the partnumber structure of the given BOM (bill of material) from sivas. The Order of the partnumbers is identical to the position numbers in the BOM"""
unordered_parent_children = {}
for parents in self._partnumbers:
for children in self._partnumbers[parents]:
if children["Parent"] not in unordered_parent_children:
unordered_parent_children[children["Parent"]] = {}
pos = children["Position"].lstrip(".").split("/")[0]
pos_in_list = int(pos)
unordered_parent_children[children["Parent"]][pos_in_list] = parents
# Second dict necessary, because children aren't sorted according to their position numbers in the SIVAS-BOM
ordered_parent_children = {}
for key, values in unordered_parent_children.items():
for subkey, subvalue in sorted(values.items()):
if key not in ordered_parent_children:
ordered_parent_children[key] = []
ordered_parent_children[key].append(subvalue)
return ordered_parent_children
@lazy_partnumber_init
def get_max_bom_depth(self) -> int:
"""Returns the maximum nesting-depth (level) of the provided assembly partnumber as an integer"""
bom_levels = []
for values in self._partnumbers.values():
for value in values:
current_bom_level = value["Position"].count(".")
bom_levels.append(current_bom_level)
return max(bom_levels)
@lazy_partnumber_init
def get_partnumber_descriptions(self) -> dict:
descriptions = {}
for parent, values in self._partnumbers.items():
for value in values:
descriptions[parent] = value["Bezeichnung"]
return descriptions
def create_excel_export(self):
# Spalten in Excel-Datei: Sortierung, Teilenummer, Bezeichnung, Bestandsabfrage, Teilebeurteilung, Zusatzbemerkung
# Sortierung: Minus-separierter "Pfad" dieser Teilenummer im Assembly / im Halbzeug
# Teilenummer: Sivas-Teilenummer
# Bezeichnung: SIVAS-Bezeichnung der Teilenummer
# Migrationsstatus: "IMPORTIEREN" = Teilenummer muss noch migriert werden, "VERFÜGBAR" = Teilenummer wurde bereits migriert
# Restliche Spalten bleiben leer, weil sie in Excel befüllt werden
class RuleDesignerPartnumbers(Partnumbers):
def __init__(self, ruledesigner_ids_dir: str, ruledesigner_ids_filename: str):
self.ruledesigner_ids_dir = ruledesigner_ids_dir
self.ruledesigner_ids_filename = ruledesigner_ids_filename
super().__init__(file_dir=self.ruledesigner_ids_dir, filename=self.ruledesigner_ids_filename)
def get_partnumbers(self):
return super().load_from_csv()
def update_database(self):
return update_database.get_rd_dbase(self.ruledesigner_ids_dir)
class SivasRuleDesignerComparison:
def __init__(self, sivas_ids_dir, sivas_main_id, rd_ids_dir, rd_ids_filename):
self.sivas = SivasPartnumbers(main_sivas_id=sivas_main_id, file_dir=sivas_ids_dir)
self.ruledesigner = RuleDesignerPartnumbers(ruledesigner_ids_dir=rd_ids_dir, ruledesigner_ids_filename=rd_ids_filename)
def get_migration_status(self) -> dict:
"""Checks if given sivas-partnumbers have been migrated been migrated into RuleDesigner"""
sivas_ids = self.sivas.get_partnumbers()
ruledesigner_ids = self.ruledesigner.get_partnumbers()
migration_completed = {}
for sivas_partnumber in sivas_ids:
migration_completed[sivas_partnumber] = True if sivas_partnumber in ruledesigner_ids else False
return migration_completed
class ExcelExport:
def __init__(self, main_id, sivas_ids_file_dir, rd_ids_file_dir, rd_ids_filename):
self.main_id = main_id
self.environment = Environment()
self._sivas = SivasPartnumbers(main_sivas_id=main_id, file_dir=sivas_ids_file_dir)
self._ruledesigner = RuleDesignerPartnumbers(ruledesigner_ids_dir=rd_ids_file_dir, ruledesigner_ids_filename=rd_ids_filename)
self._comparison = SivasRuleDesignerComparison(sivas_ids_dir=sivas_ids_file_dir, sivas_main_id=main_id, rd_ids_dir=rd_ids_file_dir, rd_ids_filename=rd_ids_filename)
self._parent_children = self._sivas.get_ordered_parent_children()
self._ruledesigner_ids = self._ruledesigner.get_partnumbers()
def write_excel(self):
data = self._create_excel_data()
output_dir = environment.out_dir
destiny_filepath = os.path.join(output_dir, self.main_id + ".xlsm")
workbook = openpyxl.load_workbook(self.environment.vba_macro_filepath, keep_vba=True)
worksheet = workbook.active
# Headers
for header, column in zip(data, worksheet.iter_cols(min_row=1, max_col=len(data), max_row=1)):
for cell in column:
cell.value = header
# Values below Headers
for value, column in zip(data.values(), worksheet.iter_cols(min_row=2, max_col=len(data), max_row=len(data["Sortierung"])+1)):
for cell, element in zip(column, value):
cell.value = element
# Renaming the sheet is necessary, because it triggers the execution of the VBA-Macro
# The renaming must be performed AFTER inserting the data into the spreadsheet, because the conditional formatting, which is part of the macro, can't be performed
# when no data is in the spreadsheet
worksheet.title = "Ergebnis"
workbook.save(destiny_filepath)
print(f"Excel-Datei erfolgreich in '{output_dir}' geschrieben...")
def _create_excel_data(self) -> dict:
# Columns in excel-spreadsheet: Sortierung, Teilenummer, Bezeichnung, Bestandsabfrage, Teilebeurteilung, Zusatzbemerkung
# Sortierung: Minus-separated hierarchy of the partnumber in the BOM structure
# Teilenummer: SIVAS-partnumber
# Bezeichnung: SIVAS-description of the partnumber
# Migrationsstatus: "IMPORTIEREN" = partnumber hasn't been migrated yet, "VERFÜGBAR" = partnumber has been migrated
# All the other columns remain empty, because their values are entered in the excel spreadsheet
sorting = self._create_sorting_string()
migrated = self._comparison.get_migration_status()
descriptions = self._sivas.get_partnumber_descriptions()
sorting = []
prt_num = []
description = []
migration_status = []
evaluation = []
comment = []
migrated = self.check_migration_status()
for id in self.sivas_ids:
sorting.append(self.create_sorting_string(id))
prt_num.append(id)
description.append(self.sivas_ids[id][0]['Bezeichnung'])
migration_status.append("VERFÜGBAR" if migrated[id] is True else "IMPORTIEREN")
for partnumber_string in sorting:
partnumber = partnumber_string.split("-")[-1]
prt_num.append(partnumber)
if partnumber not in descriptions:
description.append("")
else:
description.append(descriptions[partnumber])
migration_status.append("VERFÜGBAR" if migrated[partnumber] is True else "IMPORTIEREN")
evaluation.append('')
comment.append('')
@@ -115,125 +265,173 @@ class Compare:
"Beurteilung": evaluation,
"Bemerkung": comment
}
destiny_filepath = os.path.join(Compare.out_dir, self.main_sivas_id + ".xlsm")
workbook = openpyxl.load_workbook(Compare.vba_macro_filepath, keep_vba=True)
worksheet = workbook.active
# Überschriften eintragen
for header, column in zip(data, worksheet.iter_cols(min_row=1, max_col=len(data), max_row=1)):
for cell in column:
cell.value = header
# Werte (unter Überschriften) eintragen
for value, column in (zip(data.values(), worksheet.iter_cols(min_row=2, max_col=len(data), max_row=len(sorting)))):
for cell, element in zip(column, value):
cell.value = element
# Umbenennung des Blattes ist erforderlich, weil VBA Makro erst wirksam wird, wenn das Blatt nicht mehr "Sheet1" heißt
# Die Umbenennung darf nicht vor der Eintragung der Daten erfolgen, weil es sonst Probleme mit der bedingten Formatierung gibt
worksheet.title = "Ergebnis"
workbook.save(destiny_filepath)
print(f"Excel-Datei erfolgreich in '{Compare.out_dir}' geschrieben...")
def parent_children_mapping(self):
# Methode erstellt dict mit allen Parents (Keys) der json-Datei und weist den Parents deren Kinder als Werte zu.
parent_children = {}
for id in self.sivas_ids:
parent = self.sivas_ids[id][0]["Parent"]
if parent not in parent_children:
parent_children[parent] = []
parent_children[parent].append(id)
return parent_children
def get_excel_comments_for_treeview(self):
workbook = openpyxl.load()
pass
def create_treeview(self):
symbols = {
"pipe": "",
"empty": " ",
"elbow": "└──",
"t": "├──"
}
# ANSI-Escape-Sequenzen für Einfärbung des Terminals abhängig vom Migrationsstatus.
# os.system wird für die Verarbeitung der Sequenzen benötigt
os.system("color")
colors = {
"RED": "\033[41m",
"GREEN": "\033[42m",
"RESET": "\033[0m"
}
prefix = ""
parent_children = self.parent_children_mapping()
max_tree_depth = max(id_value[0]["Position"].count(".") for id_value in self.sivas_ids.values())
max_sivas_num_len = max([len(id) for id in self.sivas_ids.keys()])
max_description_len = max([len(id_value[0]["Bezeichnung"]) for id_value in self.sivas_ids.values()])
for i, id in enumerate(self.sivas_ids):
parent = self.sivas_ids[id][0]["Parent"]
description = self.sivas_ids[id][0]["Bezeichnung"]
current_key_level = self.sivas_ids[id][0]["Position"].count(".")
return data
# Aktueller Verbinder: └── , wenn letzter Knotenpunkt, sonst ├──
connector = symbols['elbow'] if id == parent_children[parent][-1] else symbols['t']
def _create_sorting_string(self, parent=None, previous_parents="", sorting_strings=None) -> list:
# grün = Teilenummer migriert, rot = Teilenummer noch nicht migriert
out_color = f"{colors['GREEN']}" if id in self.ruledesigner_ids else f"{colors['RED']}"
"""
Creates a string which represents the hierarchy of a partnumber in the SIVAS-BOM (bill of material).
Example:
The string "1000-1100-1110-1111" for the partnumber "1111" would represent the following BOM structure:
if i == 0:
# Bei oberster Teilenummer soll kein Verbinder dargestellt werden.
print(f"{out_color}{parent}{colors['RESET']}")
1000
└──1100
└──1110
└──1111
"""
tree_str = f"{prefix}{connector}{out_color}{id}{colors['RESET']}"
len_tree_str = len(tree_str) - len(out_color + colors['RESET'])
if parent is None:
parent = self.main_id
if sorting_strings is None:
sorting_strings = [parent]
pad_tree = (max_tree_depth * len(symbols['pipe']) + len(symbols['t']) + max_sivas_num_len - len_tree_str + 2) * " "
pad_description = (max_description_len - len(description) + 2) * " "
children = self._parent_children[parent]
print(f"{tree_str}{pad_tree}{description}{pad_description}")
for child in children:
if (i + 1) != len(self.sivas_ids):
next_key = list(self.sivas_ids)[i + 1]
next_key_level = self.sivas_ids[next_key][0]["Position"].count(".")
else:
# Zu migrierende Nummer ist ein bearbeitetes Halbzeug. Programm muss hier beendet werden, da ansonsten später die
# nächste Teilenummer der json benötigt wird. Diese exisitert aber nicht, weil die json nur eine Teilenummer enthält
break
sorting_strings.append(f"{previous_parents}-{parent}-{child}".lstrip("-"))
if id == parent_children[parent][-1]:
# Prüft, ob die aktuelle Nummer das letzte Kind des Parents ist
if child in self._parent_children:
self._create_sorting_string(parent=child, previous_parents=previous_parents + f"-{parent}", sorting_strings=sorting_strings)
if id in parent_children:
# Prüft, ob das aktuelle Teil selbst ein Parent ist.
# Wenn ja, muss unter dem connector-Symbol der aktuellen Nummer (also im nächsten Durchlauf) Leerraum stehen
prefix += symbols["empty"]
return sorting_strings
class Treeview:
# For correct display of ANSI escape sequences
os.system("color")
prefix_types = {"PIPE": "",
"EMPTY": " "
}
connector_symbols = {"ELBOW": "└──",
"T": "├──"
}
console_colors = {"RED": "\033[41m",
"GREEN": "\033[42m",
"RESET": "\033[0m"
}
def __init__(self, main_partnumber, sivas_src_file, rd_src_dir, rd_src_file):
self.main_partnumber = main_partnumber
self._sivas = SivasPartnumbers(main_sivas_id=main_partnumber, file_dir=sivas_src_file)
self._ruledesigner = RuleDesignerPartnumbers(ruledesigner_ids_dir=rd_src_dir, ruledesigner_ids_filename=rd_src_file)
self._sivas_rd_comparison = SivasRuleDesignerComparison(sivas_main_id=main_partnumber, sivas_ids_dir=sivas_src_file, rd_ids_dir=rd_src_dir, rd_ids_filename=rd_src_file)
self._parent_children = self._sivas.get_ordered_parent_children()
self._descriptions = self._sivas.get_partnumber_descriptions()
self._max_tree_depth = self._sivas.get_max_bom_depth()
self._longest_partnumber = self._sivas.get_longest_partnumber()
self._migration_status = self._sivas_rd_comparison.get_migration_status()
def generate_tree(self):
self._tree_header()
self._tree_body()
self._tree_summary()
def _tree_header(self):
"""Prints root number (number for which the user wants to see the tree) to the console"""
background_color = self._set_text_background_color(self.main_partnumber)
print(f"{background_color}{self.main_partnumber}{Treeview.console_colors["RESET"]}")
def _tree_body(self, parent_id=None, prefix=""):
"""Prints the body (all partnumbers except root/main-partnumber) of the tree to the console"""
if parent_id is None:
parent_id = self.main_partnumber
children_ids = self._parent_children[parent_id]
for index, child_id in enumerate(children_ids):
else:
# Ende der aktuellen Gruppe an Teilenummern, die sich auf einer gemeinsamen Ebene befinden ist erreicht.
# Die nächste Teilenummer liegt eine oder mehrere Ebenen höher als die aktuelle Teilenummer (außer das Ende des gesamten Baumes ist erreicht).
# Der bisherige Präfix-String wird deshalb gekürzt. Die Anzahl der zu kürzenden Zeichen ergibt sich durch die Anzahl der Ebenen, die zwischen
# der aktuellen Teilenummer und der darauffolgenden Teilenummer liegen und der Länge des Pipe-Symbols, bzw. eines anderen Symbols (alle Symbole
# haben die gleiche Länge)
prefix_trim_len = (next_key_level - current_key_level) * len(symbols["pipe"])
prefix = prefix[:prefix_trim_len]
else:
if id in parent_children:
# Prüft erneut, ob das aktuelle Teil auch ein Parent ist.
# Wenn ja, muss dieses Mal unter der nächsten Nummer ein Pipe-Symbol kommen
prefix += symbols["pipe"]
is_last_assembly_child = self._last_child_check(index, children_ids)
connector_symbol = self._set_connector_symbol(is_last_assembly_child)
partnumber_background_color = self._set_text_background_color(child_id)
self._print_entry(child_id, prefix, connector_symbol, partnumber_background_color)
if child_id in self._parent_children:
next_prefix = self._prefix(prefix, is_last_assembly_child)
self._tree_body(parent_id=child_id, prefix=next_prefix)
def _tree_summary(self):
amount_all_sivas_ids = len(self._migration_status)
amount_migrated_ids = sum(migration_status == True for migration_status in self._migration_status.values())
amount_unmigrated_ids = amount_all_sivas_ids - amount_migrated_ids
progress = int(amount_migrated_ids / amount_all_sivas_ids * 100)
column_width = 5
border_padding = "".ljust(column_width, "")
empty_padding = "".ljust(column_width, " ")
print(f"""
╭─────────────────────────────────────{border_padding}
│ Auswertung {empty_padding}
├───────────────────────────────────┬─{border_padding}
│ Anzahl abgeschlossene Migrationen │ {str(amount_migrated_ids).ljust(column_width)}
│ Anzahl fehlende Migrationen │ {str(amount_unmigrated_ids).ljust(column_width)}
├───────────────────────────────────┼─{border_padding}
│ Migrationsfortschritt │ {f"{progress}%".ljust(column_width)}
╰───────────────────────────────────┴─{border_padding}
""")
def _print_entry(self, child_id, prefix, connector, color):
"""Prints tree entry"""
# Necessary for calculating the padding-width after the partnumber
longest_entry = self._max_tree_depth * len(Treeview.prefix_types["PIPE"]) \
+ len(connector) \
+ len(self._longest_partnumber) \
+ len(Treeview.console_colors["RESET"]) \
+ len(Treeview.console_colors["RED"]) # other color codes (ansi escape sequences) could have been used instead since they all have the same length
text_before_description = f"{prefix}{connector}{color}{child_id}{Treeview.console_colors["RESET"]}".ljust(longest_entry)
description = self._descriptions[child_id]
print(f"{text_before_description} {description}")
def _last_child_check(self, child_index, children):
"""Checks, if current child is the last child of the assembly"""
return True if child_index == len(children) - 1 else False
def _set_connector_symbol(self, child_is_last_child):
"""Chooses connector symbol for current tree entry"""
return Treeview.connector_symbols["ELBOW"] if child_is_last_child else Treeview.connector_symbols["T"]
def _prefix(self, prefix, child_is_last_child):
"""Creates prefix (part before the connector symbol) for each tree entry"""
prefix += Treeview.prefix_types["EMPTY"] if child_is_last_child else Treeview.prefix_types["PIPE"]
return prefix
def _set_text_background_color(self, partnumber):
"""Sets background color of partnumber in console depending on its migration status
- Migration completed: green
- Migration pending: red"""
return Treeview.console_colors["GREEN"] if self._migration_status[partnumber] is True else Treeview.console_colors["RED"]
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='Das Programm gibt Auskunft über den aktuellen Migrationsstand', prog='compare_lists')
@@ -243,13 +441,15 @@ if __name__ == '__main__':
args = parser.parse_args()
environment = Environment()
if args.excel:
num = args.number.lstrip()
excel_comparison = Compare(num)
excel_comparison.create_excel_export()
excel_comparison = ExcelExport(num, environment.data_dir, environment.data_dir, environment.ruledesigner_csv_filename)
excel_comparison.write_excel()
elif args.tree:
num = args.number.lstrip()
treeview_comparison = Compare(num)
treeview_comparison.create_treeview()
treeview_comparison = Treeview(num, environment.data_dir, environment.data_dir, environment.ruledesigner_csv_filename)
treeview_comparison.generate_tree()
else:
parser.print_help()