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

Klassen 'Partnumbers', 'SivasPartnumbers' und 'RuleDesignerPartnumbers' entfernt und stattdessen mit den Klassen in manage_configs.py verknüpft, um Redundanzen zu vermeiden. Zudem doppeltes Laden der genannten Klassen in der Klasse 'RuleDesignerComparison', die jetzt 'MigrationStatus' heißt, gefixt. Abgesehen davon durch Aufrufen von Methoden in den __init__-Methoden mittels lazy-Initialisierung per property-Dekoratoren behoben.

Browse Source
This commit is contained in:
Simon Steuer
2025-03-08 18:16:17 +01:00
parent 5bd83324b2
commit da4eed34c3
1 changed files with 159 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compare_lists.py
+159 -219
View File
@@ -1,12 +1,10 @@
import update_database
from manage_configs import SivasAssemblyData, RuleDesignerDatabaseConfig
import warnings
#Wegen Pandas Deprecation eingebaut
warnings.filterwarnings("ignore", category=DeprecationWarning)
import argparse
import os
import json
import pandas as pd
import openpyxl
"""
@@ -16,200 +14,100 @@ Dieses Programm kann:
- die oben beschriebene Darstellung in einer Excel-Datei auch als Treeview auf der Kommandozeile darstellen
"""
class Environment:
class MigrationStatus:
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
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)
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)
def __init__(self, asm_child_ids: tuple, rd_ids: tuple):
self.asm_child_ids = asm_child_ids
self.rd_ids = rd_ids
self._overview = None
self._green_ids = None
self._yellow_ids = None
self._red_ids = None
return ids
@property
def overview(self) -> dict:
"""Returns dict with children of the given assembly and their migration status (True = migration complete, False = migration pending)"""
def get_longest_partnumber(self) -> str:
"""Returns the longest partnumber in the provided json-file"""
return max(self.get_partnumbers())
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
@lazy_partnumber_init
def get_ordered_parent_children(self, no_duplicates: bool = False) -> dict:
@property
def green_ids(self) -> list:
"""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"""
"""Returns migrated child-ids of given assembly"""
unordered_parent_children = {}
if self._green_ids is None:
self._green_ids = [id for id, status in self.overview.items() if status is True]
return self._green_ids
for parents in self._partnumbers:
for children in self._partnumbers[parents]:
@property
def red_ids(self) -> list:
if children["Parent"] not in unordered_parent_children:
unordered_parent_children[children["Parent"]] = {}
"""Returns unmigrated child-ids of given assembly"""
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
if self._red_ids is None:
self._red_ids = [id for id, status in self.overview.items() if status is False]
return self._red_ids
@property
def yellow_ids(self) -> list:
"""Returns migrated child-ids of given assembly, which prevent the asm from being released, because they are currently being modified"""
pass
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 __init__(self, asm_id, asm_json_dir, asm_json_filename, rd_dbase_dir, rd_dbase_filename):
self.asm_id = asm_id
self._asm_data = SivasAssemblyData(asm_json_dir, asm_json_filename)
self._rd_dbase = RuleDesignerDatabaseConfig(rd_dbase_dir, rd_dbase_filename)
self._asm_child_ids = None
self._rd_ids = None
self._migration_status = None
self._parent_children = None
self._id_descriptions = None
@property
def rd_ids(self):
if self._rd_ids is None:
self._rd_ids = self._rd_dbase.get_ruledesigner_ids()
return self._rd_ids
@property
def migration_status(self):
if self._migration_status is None:
self._migration_status = MigrationStatus(self.asm_child_ids, self.rd_ids)
return self._migration_status
@property
def parent_children(self) -> dict:
if self._parent_children is None:
self._parent_children = self._asm_data.get_parent_children()
return self._parent_children
@property
def id_descriptions(self):
if self._id_descriptions is None:
self._id_descriptions = self._asm_data.get_bom_property_values("Bezeichnung")
return self._id_descriptions
@property
def asm_child_ids(self):
if self._asm_child_ids is None:
self._asm_child_ids = self._asm_data.get_assembly_ids()
return self._asm_child_ids
def write_excel(self):
data = self._create_excel_data()
output_dir = environment.out_dir
out_dir = os.environ.get("CREMIG_WORK")
out_path = os.path.join(out_dir, self.asm_id + ".xlsm")
destiny_filepath = os.path.join(output_dir, self.main_id + ".xlsm")
workbook = openpyxl.load_workbook(self.environment.vba_macro_filepath, keep_vba=True)
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
# Headers
@@ -222,13 +120,13 @@ class ExcelExport:
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
# Renaming the sheet is necessary, because that 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 vba-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...")
workbook.save(out_path)
print(f"Excel-Datei erfolgreich in '{out_dir}' geschrieben...")
def _create_excel_data(self) -> dict:
@@ -240,8 +138,8 @@ class ExcelExport:
# 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()
migrated = self.migration_status.green_ids
descriptions = self.id_descriptions
prt_num = []
description = []
@@ -256,7 +154,7 @@ class ExcelExport:
description.append("")
else:
description.append(descriptions[partnumber])
migration_status.append("VERFÜGBAR" if migrated[partnumber] is True else "IMPORTIEREN")
migration_status.append("VERFÜGBAR" if partnumber in migrated else "IMPORTIEREN")
evaluation.append('')
comment.append('')
@@ -284,21 +182,20 @@ class ExcelExport:
"""
if parent is None:
parent = self.main_id
parent = self.asm_id
if sorting_strings is None:
sorting_strings = [parent]
children = self._parent_children[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:
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:
@@ -319,16 +216,59 @@ class Treeview:
}
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 __init__(self, asm_id, asm_json_dir, asm_json_filename, rd_dbase_dir, rd_dbase_filename):
self.asm_id = asm_id
self._asm_data = SivasAssemblyData(asm_json_dir, asm_json_filename)
self._rd_dbase = RuleDesignerDatabaseConfig(rd_dbase_dir, rd_dbase_filename)
self._asm_child_ids = None
self._rd_ids = None
self._migration_status = None
self._parent_children = None
self._id_descriptions = None
self._max_tree_depth = None
self._longest_partnumber = None
@property
def asm_child_ids(self):
if self._asm_child_ids is None:
self._asm_child_ids = self._asm_data.get_assembly_ids()
return self._asm_child_ids
@property
def rd_ids(self):
if self._rd_ids is None:
self._rd_ids = self._rd_dbase.get_ruledesigner_ids()
return self._rd_ids
@property
def migration_status(self):
if self._migration_status is None:
self._migration_status = MigrationStatus(self.asm_child_ids, self.rd_ids)
return self._migration_status
@property
def parent_children(self) -> dict:
if self._parent_children is None:
self._parent_children = self._asm_data.get_parent_children()
return self._parent_children
@property
def id_descriptions(self):
if self._id_descriptions is None:
self._id_descriptions = self._asm_data.get_bom_property_values("Bezeichnung")
return self._id_descriptions
@property
def max_tree_depth(self):
if self._max_tree_depth is None:
self._max_tree_depth = self._asm_data.get_max_bom_depth()
return self._max_tree_depth
@property
def longest_partnumber(self):
if self._longest_partnumber is None:
self._longest_partnumber = self._asm_data.get_longest_assembly_id()
return self._longest_partnumber
def generate_tree(self):
self._tree_header()
@@ -339,17 +279,17 @@ class Treeview:
"""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']}")
background_color = self._set_text_background_color(self.asm_id)
print(f"{background_color}{self.asm_id}{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
parent_id = self.asm_id
children_ids = self._parent_children[parent_id]
children_ids = self.parent_children[parent_id]
for index, child_id in enumerate(children_ids):
@@ -366,11 +306,10 @@ class Treeview:
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
amount_migrated_ids = len(self.migration_status.green_ids)
amount_unmigrated_ids = len(self.migration_status.red_ids)
progress = int(amount_migrated_ids / amount_all_sivas_ids * 100)
progress = int(amount_migrated_ids / len(self.asm_child_ids) * 100)
column_width = 5
border_padding = "".ljust(column_width, "─")
@@ -392,14 +331,14 @@ class Treeview:
"""Prints tree entry"""
# Necessary for calculating the padding-width after the partnumber
longest_entry = self._max_tree_depth * len(Treeview.prefix_types["PIPE"]) \
longest_entry = self.max_tree_depth * len(Treeview.prefix_types["PIPE"]) \
+ len(connector) \
+ len(self._longest_partnumber) \
+ 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]
description = self.id_descriptions[child_id]
print(f"{text_before_description} {description}")
@@ -432,27 +371,28 @@ class Treeview:
- Migration completed: green
- Migration pending: red"""
return Treeview.console_colors["GREEN"] if self._migration_status[partnumber] is True else Treeview.console_colors["RED"]
return Treeview.console_colors["GREEN"] if self.migration_status.overview[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')
parser.add_argument('-rddir', '--rd_dbase_dir', action='store', default=os.environ.get("CREMIG_DATA"))
parser.add_argument('-rdfn', '--rd_dbase_filename', action='store', default=os.environ.get("RD_DATABASE_NAME"))
parser.add_argument('-asmdir', '--asm_json_dir', action='store', default=os.environ.get("CREMIG_DATA"))
args = parser.parse_args()
environment = Environment()
asm_id = args.sivasid.lstrip()
asm_json_filename = f'{asm_id}.json'
if args.excel:
num = args.sivasid.lstrip()
excel_comparison = ExcelExport(num, environment.data_dir, environment.data_dir, environment.ruledesigner_csv_filename)
excel_comparison = ExcelExport(asm_id, args.asm_json_dir, asm_json_filename, args.rd_dbase_dir, args.rd_dbase_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 = Treeview(asm_id, args.asm_json_dir, asm_json_filename, args.rd_dbase_dir, args.rd_dbase_filename)
treeview_comparison.generate_tree()
else:
parser.print_help()
parser.print_help()