import update_database import warnings #Wegen Pandas Deprecation eingebaut warnings.filterwarnings("ignore", category=DeprecationWarning) import argparse import os import json import pandas as pd import openpyxl """ Dieses Programm kann: - eine Excel-Datei exportieren, in der alle Teilenummern aufgelistet sind, die in Verbindung mit einer angegebenen Nummer migriert werden müssen. Sie gibt dabei Auskunft darüber, welche Teilenummern bereits migriert wurden und welche noch migriert werden müssen. - die oben beschriebene Darstellung in einer Excel-Datei auch als Treeview auf der Kommandozeile darstellen """ class Environment: 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) class Partnumbers: 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(config_path, encoding='utf-8') as json_file: return json.load(json_file) except FileNotFoundError: update_database.get_sivas_dbase(self.filename, self.file_dir) return self.load_from_json() def load_from_csv(self): """Reads partnumbers from a csv-file""" config_path = os.path.join(self.file_dir, self.filename + ".csv") try: df = pd.read_csv(config_path, sep=';', converters={"Teilenummern":str}) return tuple(df["Teilenummern"]) except FileNotFoundError: update_database.get_rd_dbase(self.file_dir) return self.load_from_csv() class SivasPartnumbers(Partnumbers): 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 super().__init__(file_dir=self.file_dir, filename=self.main_sivas_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) return ids 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 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() prt_num = [] description = [] migration_status = [] evaluation = [] comment = [] 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('') data = { "Sortierung": sorting, "Teilenummer": prt_num, "Bezeichnung": description, "Migrationsstatus": migration_status, "Beurteilung": evaluation, "Bemerkung": comment } return data def _create_sorting_string(self, parent=None, previous_parents="", sorting_strings=None) -> list: """ 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: 1000 └──1100 └──1110 └──1111 """ if parent is None: parent = self.main_id if sorting_strings is None: sorting_strings = [parent] children = self._parent_children[parent] for child in children: sorting_strings.append(f"{previous_parents}-{parent}-{child}".lstrip("-")) if child in self._parent_children: self._create_sorting_string(parent=child, previous_parents=previous_parents + f"-{parent}", sorting_strings=sorting_strings) 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): 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') parser.add_argument('-s', '--sivasid', action='store', required=True, help='Eingabe der SIVAS-Teilenummer, für die der Migrationsstatus angezeigt werden soll', metavar="id") parser.add_argument('-t', '--tree', action='store_true', help='Erzeugt Baumdarstellung aller Teilenummern. grün = Teilenummer vorhanden, rot = Migration ausstehend') parser.add_argument('-e', '--excel', action='store_true', help='Erzeugt Excel-Datei in der alle Teilenummern, die migriert werden müssen, aufgelistet sind') args = parser.parse_args() environment = Environment() if args.excel: num = args.sivasid.lstrip() excel_comparison = ExcelExport(num, environment.data_dir, environment.data_dir, environment.ruledesigner_csv_filename) excel_comparison.write_excel() elif args.tree: num = args.sivasid.lstrip() treeview_comparison = Treeview(num, environment.data_dir, environment.data_dir, environment.ruledesigner_csv_filename) treeview_comparison.generate_tree() else: parser.print_help()