From cc664b036960d7d9e4124bf14f425d79df615ce9 Mon Sep 17 00:00:00 2001 From: foefl Date: Fri, 27 Feb 2026 10:30:29 +0100 Subject: [PATCH] add notes and simplify routines --- src/KSG_anomaly_detection/_profile.py | 6 +- src/KSG_anomaly_detection/monitor.py | 9 ++- src/KSG_anomaly_detection/preparation.py | 87 ++++++++++++++++++++---- 3 files changed, 88 insertions(+), 14 deletions(-) diff --git a/src/KSG_anomaly_detection/_profile.py b/src/KSG_anomaly_detection/_profile.py index e68ee4c..9a80999 100644 --- a/src/KSG_anomaly_detection/_profile.py +++ b/src/KSG_anomaly_detection/_profile.py @@ -8,17 +8,21 @@ profiler = cProfile.Profile() PROFILE = True USE_NEW_IMPL = True +ONLY_PREPARE = False def main() -> None: _prepare_env.main() + if ONLY_PREPARE: + return + if PROFILE: profiler.enable() monitor_folder_simple(use_new=USE_NEW_IMPL) profiler.disable() stats = pstats.Stats(profiler).sort_stats("cumtime") - stats.print_stats(15) + stats.print_stats(20) else: monitor_folder_simple(use_new=USE_NEW_IMPL) diff --git a/src/KSG_anomaly_detection/monitor.py b/src/KSG_anomaly_detection/monitor.py index 8646d5a..c21d540 100644 --- a/src/KSG_anomaly_detection/monitor.py +++ b/src/KSG_anomaly_detection/monitor.py @@ -1,5 +1,6 @@ import os import re +import sys import time import traceback from pathlib import Path @@ -146,7 +147,11 @@ def monitor_folder_simple(use_new: bool): # Aufgabe 2: NGT und check_img von Originalordner in KI kopieren # Rückgabewert: Ordner Fileserver/KI auf dem Fileserver, wo dann die Heatmaps abgelegt werden print("'copy_ngt_and_checkimg'...") - file_ki_folder, result = preparation.copy_ngt_and_checkimg() + if use_new: + file_ki_folder, result = preparation.copy_ngt_and_checkimg_new() + else: + file_ki_folder, result = preparation.copy_ngt_and_checkimg() + if result: # d. h. Fehler ist aufgetreten continue # zu nächstem neuen folder springen @@ -157,6 +162,8 @@ def monitor_folder_simple(use_new: bool): else: preparation.change_image_to_yellow() + # sys.exit(0) + # Aufgabe 4: AOI-Bilder in RGB überführen und zwischenspeichern # wir erhalten hier den Speicherort sowie ggf. Fehlermeldungen zurück print("'create_rgb_images_and_patches'...") diff --git a/src/KSG_anomaly_detection/preparation.py b/src/KSG_anomaly_detection/preparation.py index ce0cd75..7917199 100644 --- a/src/KSG_anomaly_detection/preparation.py +++ b/src/KSG_anomaly_detection/preparation.py @@ -18,16 +18,20 @@ COLOUR_ASSIGNMENT = {"R": [255, 0, 0], "G": [0, 255, 0], "B": [0, 0, 0]} class Preparation: def __init__(self, folder): - self.folder_path = ( - folder # der aktuelle Ordner mit neuen AOI-Bilddateien auf dem KI-Rechner - ) - self.visper = Path(self.folder_path).parts[-4] - self.original_data_path = ( - Path(config.STORING_PATH) - / self.visper - / Path(self.folder_path).parent.parent.name - / Path(self.folder_path).parent.name - ) # Pfad zu Fileserver/KI/... + # der aktuelle Ordner mit neuen AOI-Bilddateien auf dem KI-Rechner + self.folder_path = folder + + self.path = Path(self.folder_path) + # ?? verify existence --> fail early? + + self.visper = self.path.parts[-4] + # self.original_data_path = ( + # Path(config.STORING_PATH) + # / self.visper + # / Path(self.folder_path).parents[1].name + # / Path(self.folder_path).parent.name + # ) # Pfad zu Fileserver/KI/... + self.original_data_path = Path(config.STORING_PATH).joinpath(*self.path.parts[-4:-1]) print(f"[{self.visper}] {Path(self.folder_path).name} Vorbereitung gestartet...") @@ -46,7 +50,6 @@ class Preparation: folders_to_copy.append(entry.path) for src in folders_to_copy: - # TODO duplicate -> "self.original_data_path" dst = ( Path(config.STORING_PATH) / self.visper @@ -70,6 +73,59 @@ class Preparation: no = tb[-1].lineno return None, f"Fehlermeldung: {e}, {no}" + def copy_ngt_and_checkimg_new(self): + try: + # extract last level name because we need to copy all folders containing this name + # target_name = os.path.basename(self.folder_path) + # target_name = self.path.name + # base_path = self.path.parent + + # ?? Warum kopieren wir nicht gleich den ganzen Ordner? --> Das Ergebnis der unten stehenden + # ?? Operation ist eine Liste, in der nur der Name des Ordners (Päckchen-Ebene) steht, die also eigentlich + # ?? immer nur einen Eintrag enthält --> self.path + # Einzige Ausnahme: mehr als 10 Einträge/Päckchen, dann enthalten die Päckchen 10-19 auch den Namen des + # ersten Päckchens + # Annahme hier: Es soll nur die Daten des entsprechenden Päckchens kopiert werden + + # folders_to_copy = [] + # with os.scandir(base_path) as entries: + # for entry in entries: + # if entry.is_dir() and target_name in entry.name: + # folders_to_copy.append(entry.path) + + # for src in folders_to_copy: + # # duplicate -> self.original_data_path + # dst = ( + # Path(config.STORING_PATH) + # / self.visper + # / Path(self.folder_path).parent.parent.name + # / Path(self.folder_path).parent.name + # / Path(src).name + # ) + # copytree(src, dst, dirs_exist_ok=True) + + src = self.path + dst = self.original_data_path / self.path.name + copytree(src, dst, dirs_exist_ok=True) + + # ?? Soll hier der Pfad mit den kopierten Originaldaten zurückgegeben werden? + # Das ist aktuell nicht der Fall, da die letzten Pfadkomponenten als absolute Pfade das + # Ergebnis komplett überschreiben + # return Path(config.STORING_PATH) / self.visper / Path( + # self.folder_path + # ).parent.parent / Path(self.folder_path).parent, None + return self.original_data_path, None + + except FileExistsError: + return ( + None, + f"Fehlermeldung: Ordner {Path(self.folder_path).parts[-1]} existiert bereits.", + ) + except Exception as e: + tb = traceback.extract_tb(e.__traceback__) + no = tb[-1].lineno + return None, f"Fehlermeldung: {e}, {no}" + # --------------------------- Dritte Aufgabe: check_img auf dem Fileserver auf Gelb ("Work in Progress") ändern -------------------------- def change_image_to_yellow(self): # first we define for R, G and B which coour has to be adapted @@ -108,7 +164,14 @@ class Preparation: return img def change_image_to_yellow_new(self): - base_path = Path(self.folder_path).parent + # ?? Hier sind wir nicht mehr auf Ebene des Päckchens, sondern im übergeordneteten Bereich. Ist das + # ?? richtig so? Beim Kopieren waren wir nur auf Päckchen-Ebene unterwegs. + # Wenn jetzt neue Instanzen dieser Klasse mit anderen Päckchen desselben Basisordners erstellt werden, werden + # die Berechnungen an dieser Stelle immer erneut durchgeführt, auch für die anderen Päckchen, egal ob bereits + # durchgeführt oder durch Nutzer abgewählt + # Annahme hier: Es sollen nur Dateien des aktuellen Päckchens bearbeitet werden. + base_path = self.path + # base_path = Path(self.folder_path).parent # iterate over all 'checkimg' folders recursively # !! check needed