from manage_configs import SivasBOM, RuleDesignerDatabaseConfig, MigrationStatus, SivasTeilestamm, BOMConfig import argparse import os import openpyxl from itertools import zip_longest """ 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 Darstellung in der Excel-Datei als Baum in der Konsole ausgeben """ class GetExcludedIDsFromCFG(argparse.Action): def __call__(self, parser, namespace, values, option_string = None): bom_config = BOMConfig(os.path.dirname(values[0]), os.path.basename(values[0])) excluded_ids = bom_config.get_excluded_ids(values[1]) setattr(namespace, self.dest, excluded_ids) class LStrip(argparse.Action): def __call__(self, parser, namespace, values, option_string = None): setattr(namespace, self.dest, values.lstrip()) class LocateSQLCredentials(argparse.Action): def __call__(self, parser, namespace, values, option_string = None): if not os.path.isfile(os.environ.get("RD_SQL_DATABASE_CREDENTIALS")): print(f"ACHTUNG: Blau hinterlegte Nummern (Teilenummern die ins PDM migriert wurden, aber aktuell nicht freigegeben sind) werden nicht dargestellt, da nicht auf die Login-Daten für die RD-SQL-Datenbank zugegriffen werden kann. Bitte prüfen, ob die Login-Datei '{os.environ.get("RD_SQL_DATABASE_CREDENTIALS")}' existiert. Auch prüfen, ob die Netzlaufwerke verbunden sind!") setattr(namespace, self.dest, True) class ExcelExport: def __init__(self, asm_id, asm_json_dir, asm_json_filename, asm_csv_dir, asm_csv_filename, rd_dbase_dir, rd_dbase_filename, load_local_asm_json, load_local_asm_csv, load_local_rd_dbase, no_blue_ids): self.asm_id = asm_id self.asm_data = SivasBOM(asm_json_dir, asm_json_filename, load_local_asm_json) self.asm_metadata = SivasTeilestamm(asm_csv_dir, asm_csv_filename, load_local_asm_csv) self.rd_data = RuleDesignerDatabaseConfig(rd_dbase_dir, rd_dbase_filename, load_local_rd_dbase) self._migration_status = None @property def migration_status(self): if self._migration_status is None: self._migration_status = MigrationStatus(self.asm_data.child_ids, self.rd_data.ids) return self._migration_status def write_excel(self): data = self._create_excel_data() out_dir = os.environ.get("CREMIG_WORK") out_path = os.path.join(out_dir, self.asm_id + ".xlsm") workbook_path = os.path.join(os.environ.get("CREMIG_LIB"), os.environ.get("VBA_MACRO_NAME")) workbook = openpyxl.load_workbook(workbook_path, keep_vba=True) worksheet = workbook.active # Spaltenüberschriften 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 Spaltenüberschriften 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 # Umbenennung des Blattes erforderlich, da die Umbenennung die Ausführung des VBA-Makros triggert. # Die Umbenennung muss NACH dem Einfügen der Daten in das Blatt erfolgen, da die bedingte Formatierung, # die ebenfalls Bestandteil des Makros ist, nicht ausgeführt werden kann, solange sich keine Daten im Blatt befinden worksheet.title = "Ergebnis" workbook.save(out_path) print(f"Excel-Datei erfolgreich in '{out_dir}' geschrieben...") def _create_excel_data(self) -> dict: # Spalten im Arbeitsblatt: Sortierung, Teilenummer, Bezeichnung, Bestandsabfrage, Teilebeurteilung, Zusatzbemerkung # Sortierung: Minus-separierte Hierarchie der Teilenummern in der Stücklistenstruktur # Teilenummer: SIVAS-Teilenummer # Bezeichnung: SIVAS-Bezeichnung der Teilenummer # Migrationsstatus: "IMPORTIEREN" = Migration ausstehend, "VERFÜGBAR" = Migration abgeschlossen # Alle weiteren Spalten bleiben leer, da sie händisch vom User zu befüllen sind sorting = self._create_sorting_string() migrated_ids = self.migration_status.ids_in_rd child_descriptions:dict = self.asm_data.get_prop_val_for_all_ids("Bezeichnung") parent_description:str = self.asm_metadata.get_property_values("Bezeichnung1") descriptions = child_descriptions.copy() descriptions[asm_id] = parent_description prt_num = [] description = [] migration_status = [] assessment = [] comment = [] for id_str in sorting: id = id_str.split("-")[-1] prt_num.append(id) assessment.append('') comment.append('') if id not in descriptions: description.append("") else: description.append(descriptions[id]) if id in migrated_ids: if id in self.migration_status.ids_in_progress: migration_status.append("PDM-Migration abgeschlossen") else: migration_status.append("Migration abgeschlossen (freigegeben)") else: migration_status.append("Migration ausstehend") data = { "Sortierung": sorting, "Teilenummer": prt_num, "Bezeichnung": description, "Migrationsstatus": migration_status, "Beurteilung": assessment, "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.asm_id if sorting_strings is None: sorting_strings = [parent] parent_children = self.asm_data.parent_children children = parent_children[parent] for child in children: sorting_strings.append(f"{previous_parents}-{parent}-{child}".lstrip("-")) if child in 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", "BLUE": "\033[44m", "RESET": "\033[0m" } def __init__(self, asm_id:str, asm_json_dir:str, asm_json_filename:str, asm_csv_dir:str, asm_csv_filename:str, rd_dbase_dir:str, rd_dbase_filename:bool=False, load_local_asm_json:bool=False, load_local_asm_csv:bool=False, load_local_rd_dbase:bool=False, no_blue_ids:bool=False, excluded_bom_ids:list=None, excluded_summary_ids:list=None): self.asm_id = asm_id self.asm_bom = SivasBOM(asm_id, asm_json_dir, asm_json_filename, load_local_asm_json, excluded_bom_ids) self.asm_teilestamm = SivasTeilestamm(asm_csv_dir, asm_csv_filename, load_local_asm_csv) self.rd_data = RuleDesignerDatabaseConfig(rd_dbase_dir, rd_dbase_filename, load_local_rd_dbase) self.all_excluded_summary_ids = excluded_summary_ids self._excluded_summary_ids_in_bom = None self.no_blue_ids = no_blue_ids self._migration_status = None self._longest_entry = None @property def migration_status(self): if self._migration_status is None: self._migration_status = MigrationStatus(self.asm_bom.child_ids, self.rd_data.ids) return self._migration_status @property def longest_entry(self): if self._longest_entry is None: longest_color_code = max(Treeview.console_colors.values(), key=len) longest_connector_symbol = max(Treeview.connector_symbols.values(), key=len) padding_after_id = 2 if self._longest_entry is None: self._longest_entry = (self.asm_bom.max_bom_depth) * len(Treeview.prefix_types["PIPE"]) \ + len(longest_connector_symbol) \ + len(longest_color_code) \ + len(self.asm_bom.longest_child_id) \ + len(Treeview.console_colors["RESET"]) \ + padding_after_id return self._longest_entry def _add_asm_id_to_bom(self): """Standardmäßig ist die oberste Baugruppennummer kein Bestandteil der JSON-Datei, in der die Baugruppenstruktur aufgelistet ist und die über das Programm 'Sivas2Json' von Hr. Jakob generiert wird. Damit der Baum die Bezeichnung der obersten Baugruppennummer im Baum angezeigt wird ist es jedoch erforderlich, dass die Nummer mitsamt ihrer Bezeichnung in der JSON enthalten ist""" metadata = {"Bezeichnung": self.asm_teilestamm.get_property_values("Bezeichnung1"), "Parent": None, "Position": "10/0"} self.asm_bom.extend_asm_bom(self.asm_id, metadata) def generate_tree(self): self._add_asm_id_to_bom() self._print_tree() self._print_summary() def _print_tree(self, parent_id=None, prefix=""): """Gibt die Baugruppenstruktur in Form eines Baumes in der Konsole aus""" parent_children = self.asm_bom.parent_children children_ids = parent_children[parent_id] for i, id in enumerate(children_ids): last_child = self._last_child_check(i, children_ids) connector_symbol = self._set_connector_symbol(id, last_child) background_color = self._set_background_color(id) self._print_entry(id, prefix, connector_symbol, background_color) if id in parent_children: next_prefix = self._set_next_prefix(prefix, id, last_child) self._print_tree(parent_id=id, prefix=next_prefix) def _remove_ids_from_summary_amount(self, ids: list) -> list: """ Reduziert die Anzahl in der Zusammenfassung (z.B. der nicht migrierten Teilenummern), um die Anzahl an Teilenummern, die vom User aus der Zusammenfassung ausgeschlossen wurden """ if self.all_excluded_summary_ids is None: return ids return list(filter(lambda id: id not in self.excluded_summary_ids_in_bom, ids)) @property def excluded_summary_ids_in_bom(self) -> list: """ Gibt alle Teilenummern zurück, die im Baum vorliegen, aber nicht in der Anzahl der Zusammenfassung enthalten sein sollen""" if self._excluded_summary_ids_in_bom is None: self._excluded_summary_ids_in_bom = list(filter(lambda id: id in self.asm_bom.bom_data, self.all_excluded_summary_ids)) return self._excluded_summary_ids_in_bom def _print_summary(self): ids_in_rd = self.migration_status.ids_in_rd ids_pending = self.migration_status.ids_pending amount_in_rd = len(self._remove_ids_from_summary_amount(ids_in_rd)) amount_pending = len(self._remove_ids_from_summary_amount(ids_pending)) if not self.no_blue_ids: ids_in_progress = self.migration_status.ids_in_progress amount_in_work = len(self._remove_ids_from_summary_amount(ids_in_progress)) amount_released = amount_in_rd - amount_in_work child_ids = self.asm_bom.child_ids progress = int(amount_in_rd / len(self._remove_ids_from_summary_amount(child_ids)) * 100) column_width = 5 border_padding = "".ljust(column_width, "─") empty_padding = "".ljust(column_width, " ") red_entry = f"│{Treeview.console_colors['RED'] } {Treeview.console_colors['RESET']}│ Migration ausstehend │ {str(amount_pending).ljust(column_width)}│" if not self.no_blue_ids: blue_entry = f"│{Treeview.console_colors['BLUE'] } {Treeview.console_colors['RESET']}│ PDM-Migration abgeschlossen │ {str(amount_in_work).ljust(column_width)}│\n" green_entry = f"│{Treeview.console_colors['GREEN']} {Treeview.console_colors['RESET']}│ Migration abgeschlossen (Freigegeben) │ {str(amount_released).ljust(column_width)}│\n" else: blue_entry = "" green_entry = f"│{Treeview.console_colors['GREEN']} {Treeview.console_colors['RESET']}│ Migration abgeschlossen │ {str(amount_in_rd).ljust(column_width)}│\n" red_entry = f"│{Treeview.console_colors['RED'] } {Treeview.console_colors['RESET']}│ Migration ausstehend │ {str(amount_pending).ljust(column_width)}│" header = f""" ╭───────────────────────────────────────────{border_padding}╮ │ Auswertung {empty_padding}│ ├─┬───────────────────────────────────────┬─{border_padding}┤ """ progress_segment = f""" ├─┴───────────────────────────────────────┼─{border_padding}┤ │ Migrationsfortschritt │ {f"{progress}%".ljust(column_width)}│ """ bottom_row = f"╰─────────────────────────────────────────┴─{border_padding}╯" exclusion_description = "" excluded_id_list = "" if self.asm_bom.excluded_ids_in_bom or self.excluded_summary_ids_in_bom: exclusion_description = f"""├─────────────────────────────────────────┴─{border_padding}┤ │ {empty_padding}│ ├───────────────────────────────────────────{border_padding}┤ │ Ausgeschlossene Teilenummern {empty_padding}│ │──────────────────────┬────────────────────{border_padding}┤ │ Baum │ Auswertung {empty_padding}│ ├──────────────────────┼────────────────────{border_padding}┤ """ for bom_id, smry_id in zip_longest(self.asm_bom.excluded_ids_in_bom, self.excluded_summary_ids_in_bom, fillvalue=9*" "): excluded_id_list += f"│ { bom_id} │ {smry_id} {empty_padding}│\n" bottom_row = f"╰──────────────────────┴────────────────────{border_padding}╯" print(header + green_entry + blue_entry + red_entry + progress_segment + exclusion_description + excluded_id_list + bottom_row) def _print_entry(self, child_id, prefix, connector, color): """Eintrag in die Konsole drucken""" text_before_description = f"{prefix}{connector}{color}{child_id}{Treeview.console_colors['RESET']}".ljust(self.longest_entry) description = self.asm_bom.get_prop_val_for_single_id(child_id, "Bezeichnung") print(f"{text_before_description}{description}") def _last_child_check(self, child_idx, children_ids): """Prüft, ob aktuelle ID das letzte Kind der Baugruppe ist""" return True if child_idx == len(children_ids) - 1 else False def _set_connector_symbol(self, child_id, last_child): """Legt Verbindungssymbol für aktuellen Eintrag im Baum fest""" if child_id == self.asm_id: # Oberste Baugruppennummer wird immer ohne Verbindersymbol dargestellt return "" if not last_child: return Treeview.connector_symbols["T"] return Treeview.connector_symbols["ELBOW"] def _set_next_prefix(self, prefix, child_id, last_child): """Legt den Teil VOR dem Verbindersymbol für den nächsten Eintrag im Baum fest""" if child_id == self.asm_id: # Erste Stücklistenposition der obersten Baugruppennummer hat nie Präfix return "" prefix += Treeview.prefix_types["EMPTY"] if last_child else Treeview.prefix_types["PIPE"] return prefix def _set_background_color(self, id): """Legt Hintergrundfarbe der Teilenummer basierend auf ihrem Migrationsstatus fest: - grün: Migration vollständig abgeschlossen und freigegeben - blau: Migration ins PDM abgeschlossen - rot: Migration ausstehend""" if id in self.migration_status.ids_pending: return Treeview.console_colors["RED"] if not self.no_blue_ids: if id in self.migration_status.ids_in_progress: return Treeview.console_colors["BLUE"] return Treeview.console_colors["GREEN"] if __name__ == '__main__': def main(): parser = argparse.ArgumentParser(description='Programm gibt Auskunft über den aktuellen Migrationsstand einer SIVAS-Teilenummer', prog='compare_lists', formatter_class=lambda prog: argparse.HelpFormatter(prog, max_help_position=50)) # Hilfetext auch bei längeren Argumenten in gleicher Zeile wie Argument darstellen # Custom Actions parser.register("action", "GetIDsFromCFG", GetExcludedIDsFromCFG) parser.register("action", "lstrip", LStrip) parser.register("action", "LocateSQLCredentials", LocateSQLCredentials) parser.add_argument('sivas_id', action='lstrip', help='Teilenummer, für die Migrationsüberblick generiert werden soll.', metavar="SIVAS-Teilenummer") parser.add_argument('output_type', choices=["tree", "excel"], help="'tree' = Darstellung des Überblicks als Baum in der Konsole. 'excel' = Export des Überblicks als Excel") parser.add_argument('--rd_ids_filepath', action='store', default=os.path.join(os.environ.get("CREMIG_DATA"), "RD_Teilenummern.csv"), help='Verzeichnis der CSV-Datei mit den RuleDesigner-Nummern. Default: $CREMIG_DATA', metavar="filepath") parser.add_argument('--local_rd_ids', action='store_true', default=False, help='Lokales laden der CSV mit den RuleDesigner-Nummern (RD-Teilenummer.csv) statt Kopieren von Server') parser.add_argument('--bom_filedir', action='store', default=os.environ.get("CREMIG_DATA"),help='Verzeichnis der JSON-Datei mit den Stücklisteninformationen. Default: $CREMIG_DATA', metavar='dir') parser.add_argument('--local_bom', action='store_true', default=False, help='Lokales laden der JSON mit den Stücklisteninformationen statt Generierung') parser.add_argument('--teilestamm_filedir', action='store', default=os.environ.get("CREMIG_DATA"), help='Verzeichnis der CSV mit den SIVAS-Teilestamminformationen. Default: $CREMIG_DATA', metavar='dir') parser.add_argument('--local_teilestamm', action='store_true', default=False, help='Lokales Laden der CSV mit den Teilestamminformationen statt Generierung') parser.add_argument('--no_blue_ids', action="store_true", default=False, help='Nummern, die in RD vorliegen, aber nicht freigegeben sind, gleich wie freigegeben Nummern einfärben') exclude_from_overview = parser.add_mutually_exclusive_group() exclude_from_overview.add_argument('--rm_bom_ids_str', action="store", nargs="*", default=None, dest="excluded_bom_ids", help="Angabe aller Nummern, die vom Migrationsüberblick ausgeschlossen sein sollen", metavar="id1 id2") exclude_from_overview.add_argument('--rm_bom_ids_cfg', action="GetIDsFromCFG", nargs=2, default=None, dest="excluded_bom_ids", help="cfg-Pfad & Sektion, in der Nummern aufgelistet sind, die nicht im Überblick dargestellt sein sollen", metavar=("path", "section")) exclude_from_summary = parser.add_mutually_exclusive_group() exclude_from_summary.add_argument('--rm_summary_ids_str', action="store", nargs="*", default=None, dest="excluded_summary_ids", help="Angabe aller Nummern, die von der Zusammenfassung ausgeschlossen sein sollen", metavar="id1 id2") exclude_from_summary.add_argument('--rm_summary_ids_cfg', action="GetIDsFromCFG", nargs=2, default=None, dest="excluded_summary_ids", help="cfg-Pfad & Sektion, in der Nummern aufgelistet sind, die von der Zusammenfassung ausgeschlossen sein sollen", metavar=("path", "section")) args = parser.parse_args() match args.output_type: case "excel": excel_comparison = ExcelExport(asm_id=args.sivas_id, asm_json_dir=args.bom_filedir, asm_json_filename=f'{args.sivas_id}.json', asm_csv_dir=args.teilestamm_filedir, asm_csv_filename=f"{args.sivas_id}.csv", rd_dbase_dir=os.path.dirname(args.rd_ids_filepath), rd_dbase_filename=os.path.basename(args.rd_ids_filepath), load_local_asm_json=args.local_bom, load_local_asm_csv=args.local_teilestamm, load_local_rd_dbase=args.local_rd_ids, no_blue_ids=args.no_blue_ids) excel_comparison.write_excel() case "tree": treeview_comparison = Treeview(asm_id=args.sivas_id, asm_json_dir=args.bom_filedir, asm_json_filename=f'{args.sivas_id}.json', asm_csv_dir=args.teilestamm_filedir, asm_csv_filename=f'{args.sivas_id}.csv', rd_dbase_dir=os.path.dirname(args.rd_ids_filepath), rd_dbase_filename=os.path.basename(args.rd_ids_filepath), load_local_asm_json=args.local_bom, load_local_asm_csv=args.local_teilestamm, load_local_rd_dbase=args.local_rd_ids, no_blue_ids=args.no_blue_ids, excluded_bom_ids=args.excluded_bom_ids, excluded_summary_ids=args.excluded_summary_ids) treeview_comparison.generate_tree() main()