459 lines
18 KiB
Python
459 lines
18 KiB
Python
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('-n', '--number', 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.number.lstrip()
|
|
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 = Treeview(num, environment.data_dir, environment.data_dir, environment.ruledesigner_csv_filename)
|
|
treeview_comparison.generate_tree()
|
|
else:
|
|
parser.print_help()
|