diff --git a/lib/getpositions.py b/lib/getpositions.py index b29fa40..736f3dd 100644 --- a/lib/getpositions.py +++ b/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: - missing_attribs: dict mapping IDs to a message about missing or incomplete attributes. - double_ids: dict mapping IDs to a list of positions for blocks with duplicate IDs. + Args: + wp: CompareBuffer to analyze + 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) - - 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) - + res_sens, res_schaltschrank_elemente = extract_input_positions(all_inserts, all_positions, error_collector) 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["missing_attributes"] = added_warnings - output_results["not_found"] = res_not_found - output_results["double_ids"] = res_double + res_not_found = check_existance(res_mappings, res_dist, res_sens, res_tunnel) + + # Alle weiteren Fehler zum ErrorCollector hinzufügen + 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")