anderes Errorhandling mit Claude implementiert.

lib/getpositions.py

@@ -22,6 +22,58 @@ Dieses Programm:
 """
 
 
+class ErrorCollector:
+    """Sammelt alle Fehler und Warnungen während der Verarbeitung."""
+
+    def __init__(self):
+        self.errors = {}
+        self.warnings = {}
+
+    def add_errors(self, error_dict):
+        """Fügt Fehler aus einem Dictionary hinzu."""
+        for key, value in error_dict.items():
+            if key in self.errors:
+                if isinstance(self.errors[key], list) and isinstance(value, list):
+                    self.errors[key].extend(value)
+                else:
+                    self.errors[key] = [self.errors[key], value]
+            else:
+                self.errors[key] = value
+
+    def add_warnings(self, warning_dict):
+        """Fügt Warnungen aus einem Dictionary hinzu."""
+        for key, value in warning_dict.items():
+            if key in self.warnings:
+                if isinstance(self.warnings[key], list) and isinstance(value, list):
+                    self.warnings[key].extend(value)
+                else:
+                    self.warnings[key] = [self.warnings[key], value]
+            else:
+                self.warnings[key] = value
+
+    def has_errors_or_warnings(self):
+        """Prüft ob Fehler oder Warnungen vorhanden sind."""
+        return bool(self.errors) or bool(self.warnings)
+
+    def get_all_issues(self):
+        """Gibt alle Fehler und Warnungen als Dictionary zurück."""
+        result = {}
+        if self.errors:
+            result["errors"] = self.errors
+        if self.warnings:
+            result["warnings"] = self.warnings
+        return result
+
+    def write_errorfile(self, work_dir: Path, filename: str):
+        """Schreibt Fehlerdatei nur wenn Fehler oder Warnungen vorhanden sind."""
+        if self.has_errors_or_warnings():
+            print("Fehler oder Warnungen gefunden. Schreibe Error-Datei.")
+            all_issues = self.get_all_issues()
+            write_results(to_json(all_issues), work_dir, filename)
+        else:
+            print("Keine Fehler oder Warnungen gefunden. Keine Error-Datei geschrieben.")
+
+
 def write_results(jsn_results: str, out_dir: Path, filename: str) -> None:
     """Write results to a JSON file."""
     print("writing results to a json file ...")
@@ -225,7 +277,7 @@ class CompareBuffer:
                 l.append(b)
         return l
 
-def extract_input_positions(all_inserts, all_positions) -> tuple[dict, dict, dict, dict]:
+def extract_input_positions(all_inserts, all_positions, error_collector: ErrorCollector = None) -> tuple[dict, dict]:
     """
     Extracts and organizes input positions from an iterable of inserts.
 
@@ -236,14 +288,14 @@ def extract_input_positions(all_inserts, all_positions) -> tuple[dict, dict, dic
     attributes and duplicate IDs for further error handling.
 
     Args:
-        insert_iterable: An iterable of insert objects to process.
+        all_inserts: List of insert objects to process.
+        all_positions: List of position objects corresponding to inserts.
+        error_collector: ErrorCollector instance to collect errors and warnings.
 
     Returns:
         A tuple containing:
             - all_sensors: dict of sensor IDs to sensor attribute dicts
             - all_schaltschrank: dict of Schaltschrankelement IDs to their attribute dicts
-            - double_ids: dict of IDs with multiple associated blocks (potential duplicates)
-            - missing_attribs: dict of IDs with missing or incomplete attributes
     """
     all_sensors = dict()
     all_cables = dict()
@@ -275,16 +327,16 @@ def extract_input_positions(all_inserts, all_positions) -> tuple[dict, dict, dic
 
     # spezialbehandlung der Sensoren, da diese in IO und A,B,C Blöcke geteilt sind
     # Die Funktion sucht die übereinanderliegenen Elemente und baut ein Dict daraus 
-    allocate_blocks_together(all_sensors, wp)
+    allocate_blocks_together(all_sensors, wp, error_collector)
 
     set_single_frames_with_unique_sps(all_schaltschrank, wp)
 
     # die noch übrigen Blöcke melden
-    missing_attribs, double_ids = get_errors_double_and_attributes(wp) 
+    get_errors_double_and_attributes(wp, error_collector)
 
-    return all_sensors, all_schaltschrank, double_ids, missing_attribs
+    return all_sensors, all_schaltschrank
 
-def get_errors_double_and_attributes(wp: CompareBuffer) -> tuple[dict, dict]:
+def get_errors_double_and_attributes(wp: CompareBuffer, error_collector: ErrorCollector = None):
     """
     Analyze the CompareBuffer for blocks that could not be merged or assigned.
 
@@ -292,9 +344,9 @@ def get_errors_double_and_attributes(wp: CompareBuffer) -> tuple[dict, dict]:
       - IDs with only a single associated block, which likely indicates missing or incomplete attributes.
       - IDs with multiple associated blocks, which may indicate duplicate or ambiguous entries.
 
-    Returns:
+    Args:
-        missing_attribs: dict mapping IDs to a message about missing or incomplete attributes.
+        wp: CompareBuffer to analyze
-        double_ids: dict mapping IDs to a list of positions for blocks with duplicate IDs.
+        error_collector: ErrorCollector instance to collect errors and warnings
     """
     missing_attribs = dict()
     double_ids = dict()
@@ -311,7 +363,13 @@ def get_errors_double_and_attributes(wp: CompareBuffer) -> tuple[dict, dict]:
                 if id_ not in double_ids:
                     double_ids[id_] = []
                 double_ids[id_].append(block['pos'])
-    return missing_attribs, double_ids
+
+    # Füge Fehler und Warnungen zum ErrorCollector hinzu, falls vorhanden
+    if error_collector:
+        if double_ids:
+            error_collector.add_errors({"double_ids": double_ids})
+        if missing_attribs:
+            error_collector.add_warnings({"missing_attributes": missing_attribs})
 
 def set_single_frames_with_unique_sps(all_sensors: dict, wp: CompareBuffer):
     all_sensors_ids = wp.get_block_ids() 
@@ -327,7 +385,7 @@ def set_single_frames_with_unique_sps(all_sensors: dict, wp: CompareBuffer):
                 all_sensors[new_id] = block_with_sps
                 wp.remove_block(id_, block_with_sps)
 
-def allocate_blocks_together(all_sensors: dict, wp: CompareBuffer) -> None:
+def allocate_blocks_together(all_sensors: dict, wp: CompareBuffer, error_collector: ErrorCollector = None) -> None:
     """
     Merge sensor blocks with the same ID that are split across multiple DXF blocks.
 
@@ -352,19 +410,21 @@ def allocate_blocks_together(all_sensors: dict, wp: CompareBuffer) -> None:
             for block_without_sps in blks_other:
                 # Vergleiche alle Blöcke mit SPS und denen ohne auf die gleiche Position
                 if wp.positions_are_close(block_with_sps, block_without_sps, 1000):
-                    new_id = create_new_id(id_, block_with_sps, block_without_sps) # hier das Präfix davor
+                    new_id = create_new_id(id_, block_with_sps, block_without_sps, error_collector) # hier das Präfix davor
                     all_sensors[new_id] = merge_two_dicts(block_without_sps, block_with_sps) #Kombiniert alle infos aus dxf und "pos"
                     wp.remove_block(id_, block_with_sps)
                     wp.remove_block(id_, block_without_sps)
     
-def create_new_id(id_: str, dict1: dict, dict2: dict) -> str:
+def create_new_id(id_: str, dict1: dict, dict2: dict, error_collector: ErrorCollector = None) -> str:
     sps_praefix = None
     if "SPS" in dict1:
         sps_praefix = dict1["SPS"]
     if "SPS" in dict2:
         sps_praefix = dict2["SPS"]
     if not sps_praefix:
-        raise Exception
+        if error_collector:
+            error_collector.add_errors({"missing_sps_prefix": {id_: "SPS Präfix fehlt für Block-ID"}})
+        return f"{id_}@UNKNOWN"  # Fallback für fehlenden SPS-Präfix
     return f"{id_}@{sps_praefix}"
 
 def attribs_to_dicts(insert_iterable) -> tuple[list, list]:
@@ -514,7 +574,7 @@ def get_subdistributor_positions_from_symbols(all_inserts:list, all_positions:li
             continue
 
         if res in dist2sensors:
-            ret[id_] = ld["pos"]
+            ret[res] = ld["pos"]
     
     return ret
 
@@ -877,19 +937,17 @@ if __name__ == '__main__':
        
         output_results = dict()
 
+        # ErrorCollector für alle Fehler und Warnungen initialisieren
+        error_collector = ErrorCollector()
+
         if args.sensors:
             # Sensoren auslesen aus den Symbolen
+
             if use_iter:
                 all_inserts, all_positions = attribs_to_dicts(iterdxf.modelspace(dxf_path))
             else:
                 all_inserts, all_positions = attribs_to_dicts(msp)
-            res_sens, res_schaltschrank_elemente, res_double, missing_attribs = extract_input_positions(all_inserts, all_positions)
+            res_sens, res_schaltschrank_elemente = extract_input_positions(all_inserts, all_positions, error_collector)
-            
-            if args.errors and len(res_double) > 0:
-                print("Duplicate blocks found. Writing errors-file.")
-                err_ids = list(res_double.keys())
-                write_results(to_json(list(err_ids)), work_dir, args.errors)
-
 
             output_results['sensors'] = res_sens
             output_results['schaltschrank_elemente'] = res_schaltschrank_elemente
@@ -901,6 +959,9 @@ if __name__ == '__main__':
             # Mapping zu Sensoren auslesen
             (res_mappings, warnings) = create_mappings(res_sens)
             output_results['mappings'] = res_mappings
+            # Mapping-Warnungen zum ErrorCollector hinzufügen
+            if warnings:
+                error_collector.add_warnings({"mapping_warnings": warnings})
             if args.console:
                 print(to_json(res_mappings))
 
@@ -946,16 +1007,24 @@ if __name__ == '__main__':
 
         if args.write:
             basename = os.path.splitext(args.write)[0]
-            res_not_found = check_existance(res_mappings, res_dist, res_sens, res_tunnel)
-            if args.errors and "overdefined_tunnel" in res_not_found:
-                print("unorrect tunnels found. Writing errors-file.")
-                err_ids = list(res_not_found["overdefined_tunnel"])
-                write_results(to_json(list(err_ids)), work_dir, args.errors)
 
-            added_warnings = merge_two_dicts(warnings, missing_attribs)
+            res_not_found = check_existance(res_mappings, res_dist, res_sens, res_tunnel)
-            res_not_found["missing_attributes"] = added_warnings
+
-            output_results["not_found"] = res_not_found
+            # Alle weiteren Fehler zum ErrorCollector hinzufügen
-            output_results["double_ids"] = res_double
+            error_collector.add_errors(res_not_found)
+
+            # ErrorCollector schreibt Datei nur wenn Fehler/Warnungen vorhanden
+            if args.errors:
+                error_collector.write_errorfile(work_dir, args.errors)
+
+            # Für Kompatibilität mit Ausgabe-JSON: Fehler und Warnungen aus ErrorCollector holen
+            all_issues = error_collector.get_all_issues()
+            if "errors" in all_issues:
+                output_results["not_found"] = all_issues["errors"].get("not_found", {})
+                if "double_ids" in all_issues["errors"]:
+                    output_results["double_ids"] = all_issues["errors"]["double_ids"]
+                if "missing_attributes" in all_issues.get("warnings", {}):
+                    output_results["not_found"]["missing_attributes"] = all_issues["warnings"]["missing_attributes"]
             
 
             write_results(to_json(output_results), work_dir, f"{basename}.json")