""" Common utility functions for the cable routing system. This module contains frequently used helper functions for: - File and path operations - Environment variable handling - JSON operations - DXF file handling - Dictionary operations """ import json import os import sys from pathlib import Path from typing import Any, Tuple import ezdxf from ezdxf.lldxf.const import DXFStructureError # ============================================================================ # File Operations # ============================================================================ def check_file_in_work(work_dir, filename) -> Tuple[Path, bool]: """ Check if a file exists, either at the given path or in the work directory. Args: work_dir: Working directory to check if file not found at filename path (str or Path) filename: Path to the file to check (str or Path) Returns: Tuple of (resolved_path, exists_flag) - resolved_path: Path where file was found (or would be) - exists_flag: True if file exists, False otherwise """ work_dir = Path(work_dir) filename = Path(filename) fexists = True if not filename.exists(): mypath = work_dir.joinpath(filename) if not mypath.exists(): fexists = False else: mypath = filename return mypath, fexists # ============================================================================ # Environment Variable Operations # ============================================================================ def check_environment_var(env_str: str) -> Path: """ Get and validate an environment variable as a Path. Args: env_str: Name of the environment variable Returns: Path object from the environment variable Exits: Exits the program if the environment variable is not set or empty """ out_path = os.environ.get(env_str) if out_path: return Path(out_path) else: print(f"Umgebungsvariable {env_str} ist nicht gesetzt oder leer.") exit() # ============================================================================ # JSON Operations # ============================================================================ def to_json(d: Any, pretty: bool = True) -> str: """ Convert a Python object to a JSON string. Args: d: Object to convert to JSON pretty: If True, format with indentation for readability Returns: JSON string representation of the object Note: Uses ensure_ascii=False to properly display German umlauts (ä, ö, ü) """ return json.dumps(d, indent=2 if pretty else None, ensure_ascii=False, default=str) def load_json(jsonfilename: str) -> dict: """ Load JSON data from a file. Args: jsonfilename: Path to the JSON file Returns: Parsed JSON data as a dictionary """ with open(jsonfilename, encoding='utf-8') as fh: return json.load(fh) def write_results(jsn_results: str, out_dir: Path, filename: str) -> None: """ Write JSON results to a file. Args: jsn_results: JSON string to write out_dir: Output directory path filename: Name of the output file """ print("writing results to a json file ...") outfile = Path(out_dir) / filename with open(outfile, 'w', encoding='utf-8') as fh: fh.write(jsn_results) print("done") # ============================================================================ # DXF File Operations # ============================================================================ def dxf_is_binary(dxf_path: Path) -> bool: """ Check if a DXF file is in binary format. Args: dxf_path: Path to the DXF file Returns: True if the file is a binary DXF, False otherwise """ with open(dxf_path, 'rb') as f: header = f.read(22) return b'AutoCAD Binary DXF' in header def get_dxf_file(filepath: Path): """ Load a DXF file using ezdxf. Args: filepath: Path to the DXF file Returns: ezdxf Drawing object Exits: Exits with code 1 for I/O errors Exits with code 2 for invalid/corrupted DXF files """ try: print("reading file ..", end='') doc = ezdxf.filemanagement.readfile(filepath) print("done") except IOError: print("Not a DXF file or a generic I/O error.") sys.exit(1) except DXFStructureError: print("Invalid or corrupted DXF file.") sys.exit(2) return doc # ============================================================================ # ezdxf Operations (High-Level DXF Helpers) # ============================================================================ def extract_insert_attributes(insert, error_collector=None) -> dict: """ Extrahiert alle Attribute aus einem INSERT-Block. Unterstützt zweistufige Blockstruktur: - Äußerer Block enthält LWPOLYLINE und INSERT zu Attribut-Block - Innerer Attribut-Block enthält die eigentlichen Attribute Args: insert: INSERT-Entity des Blocks error_collector: Optional ErrorCollector für Warnungen Returns: Dictionary mit allen gefundenen Attributen {tag: text} """ attributes = {} if insert.dxftype() != 'INSERT': return attributes # Prüfe zuerst ob der Block direkt Attribute hat (alte Struktur) direct_attribs = list(insert.attribs) if direct_attribs: for attrib in direct_attribs: tag = attrib.dxf.tag value = attrib.dxf.text attributes[tag] = value return attributes # Neue zweistufige Struktur: Suche im Block nach einem INSERT mit Attributen # Dies erfordert das doc-Objekt, daher nur wenn verfügbar if hasattr(insert, 'doc') and insert.doc: try: block_layout = insert.doc.blocks.get(insert.dxf.name) for entity in block_layout: if entity.dxftype() == 'INSERT': inner_attribs = list(entity.attribs) if inner_attribs: # Gefunden: innerer Block mit Attributen for attrib in inner_attribs: tag = attrib.dxf.tag value = attrib.dxf.text attributes[tag] = value break # Nur das erste INSERT mit Attributen verwenden except Exception as e: if error_collector: error_msg = f"Fehler beim Extrahieren der Attribute aus Block '{insert.dxf.name}': {e}" if hasattr(error_collector, 'add_warnings'): error_collector.add_warnings({"attribute_extraction_error": error_msg}) return attributes def extract_insert_attributes_with_doc(doc, insert, error_collector=None) -> dict: """ Extrahiert alle Attribute aus einem INSERT-Block mit explizitem doc-Parameter. Empfohlen für neue Code, da es zweistufige Blockstrukturen besser unterstützt. Args: doc: DXF-Dokument insert: INSERT-Entity des Blocks error_collector: Optional ErrorCollector für Warnungen Returns: Dictionary mit allen gefundenen Attributen {tag: text} """ attributes = {} if insert.dxftype() != 'INSERT': return attributes # Prüfe zuerst ob der Block direkt Attribute hat (alte Struktur) direct_attribs = list(insert.attribs) if direct_attribs: for attrib in direct_attribs: tag = attrib.dxf.tag value = attrib.dxf.text attributes[tag] = value return attributes # Neue zweistufige Struktur: Suche im Block nach einem INSERT mit Attributen try: block_layout = doc.blocks.get(insert.dxf.name) for entity in block_layout: if entity.dxftype() == 'INSERT': inner_attribs = list(entity.attribs) if inner_attribs: # Gefunden: innerer Block mit Attributen for attrib in inner_attribs: tag = attrib.dxf.tag value = attrib.dxf.text attributes[tag] = value break # Nur das erste INSERT mit Attributen verwenden except Exception as e: if error_collector: error_msg = f"Fehler beim Extrahieren der Attribute aus Block '{insert.dxf.name}': {e}" if hasattr(error_collector, 'add_warnings'): error_collector.add_warnings({"attribute_extraction_error": error_msg}) return attributes def ensure_layer_exists(doc, layer_name: str, color: int = None, linetype: str = None, lineweight: int = None, description: str = None): """ Stellt sicher, dass ein Layer existiert und erstellt ihn falls nötig. Args: doc: DXF-Dokument layer_name: Name des Layers color: ACI Color Index (1-255), optional linetype: Name des Linientyps (z.B. 'CONTINUOUS', 'DASHED'), optional lineweight: Linienbreite in mm*100 (z.B. 25 für 0.25mm), optional description: Layer-Beschreibung, optional Returns: Layer-Objekt (existierend oder neu erstellt) """ if layer_name in doc.layers: layer = doc.layers.get(layer_name) else: layer = doc.layers.add(layer_name) # Setze optionale Attribute if color is not None: layer.color = color if linetype is not None: layer.dxf.linetype = linetype if lineweight is not None: layer.dxf.lineweight = lineweight if description is not None: layer.dxf.description = description return layer def add_rectangle_lwpolyline(msp, x: float, y: float, width: float, height: float, layer: str = '0', color: int = None, rgb: tuple = None, closed: bool = True): """ Fügt eine rechteckige LWPOLYLINE hinzu. Args: msp: Modelspace des DXF-Dokuments x: X-Koordinate der unteren linken Ecke y: Y-Koordinate der unteren linken Ecke width: Breite des Rechtecks height: Höhe des Rechtecks layer: Layer-Name (default: '0') color: ACI Color Index (1-255), optional rgb: RGB-Tuple (r, g, b) mit Werten 0-255, optional closed: Ob die Polylinie geschlossen sein soll (default: True) Returns: LWPOLYLINE-Entity """ points = [ (x, y), (x + width, y), (x + width, y + height), (x, y + height) ] polyline = msp.add_lwpolyline(points, close=closed) polyline.dxf.layer = layer if color is not None: polyline.dxf.color = color if rgb is not None: polyline.rgb = rgb return polyline def add_text_with_alignment(msp, text: str, x: float, y: float, height: float = 50, layer: str = '0', color: int = None, halign: int = 0, valign: int = 0, rotation: float = 0.0, style: str = 'Standard'): """ Fügt einen Text mit Ausrichtung hinzu. Args: msp: Modelspace des DXF-Dokuments text: Textinhalt x: X-Koordinate y: Y-Koordinate height: Texthöhe (default: 50) layer: Layer-Name (default: '0') color: ACI Color Index (1-255), optional halign: Horizontale Ausrichtung (0=links, 1=mitte, 2=rechts) valign: Vertikale Ausrichtung (0=basis, 1=unten, 2=mitte, 3=oben) rotation: Rotation in Grad (default: 0.0) style: Textstil-Name (default: 'Standard') Returns: TEXT-Entity """ text_entity = msp.add_text(text, height=height) text_entity.dxf.layer = layer text_entity.dxf.style = style text_entity.dxf.rotation = rotation if color is not None: text_entity.dxf.color = color # Setze Ausrichtung text_entity.set_placement((x, y), align=f"{'LEFT' if halign == 0 else 'CENTER' if halign == 1 else 'RIGHT'}_{'BASELINE' if valign == 0 else 'BOTTOM' if valign == 1 else 'MIDDLE' if valign == 2 else 'TOP'}") return text_entity def extract_attributes_with_positions(insert_iterable, round_decimals: int = 1): """ Wandelt eine Iterable von INSERT-Objekten in zwei Listen um. Kompatibilität mit getpositions.py attribs_to_dicts(). Args: insert_iterable: Iterable von INSERT-Entities (z.B. msp oder iterdxf.modelspace()) round_decimals: Anzahl Nachkommastellen für Positions-Rundung (default: 1) Returns: Tuple von (all_inserts, all_positions): - all_inserts: Liste von Dicts mit Attribut-Tags und deren Textwerten - all_positions: Liste von Dicts mit Attribut-Tags und deren (x, y, z)-Positionen Note: Blöcke ohne Attribute werden übersprungen. Diese Funktion ist kompatibel mit der alten attribs_to_dicts() aus getpositions.py """ all_inserts = [] all_positions = [] for insert in insert_iterable: if insert.dxftype() != 'INSERT': continue itemdata = {} positions = {} # Direkte Attribute (alte Struktur) for attrib in insert.attribs: if len(insert.attribs) == 0: continue # Überspringe Blöcke ohne Attribute attr_tag = attrib.dxf.tag attr_text = attrib.dxf.text pos = attrib.dxf.insert itemdata[attr_tag] = attr_text positions[attr_tag] = ( round(pos.x, round_decimals), round(pos.y, round_decimals), round(pos.z, round_decimals) ) if len(itemdata) > 0: all_inserts.append(itemdata) all_positions.append(positions) return all_inserts, all_positions def add_blockref_with_attributes(msp, block_name: str, insert_point: tuple, attributes: dict = None, layer: str = '0', scale: float = 1.0, rotation: float = 0.0): """ Fügt eine Block-Referenz (INSERT) mit Attributen hinzu. Args: msp: Modelspace des DXF-Dokuments block_name: Name des Blocks insert_point: Einfügepunkt als (x, y) oder (x, y, z) Tuple attributes: Dictionary mit Attribut-Namen und -Werten (optional) layer: Layer-Name (default: '0') scale: Skalierungsfaktor (default: 1.0) rotation: Rotation in Grad (default: 0.0) Returns: INSERT-Entity mit gesetzten Attributen """ # Füge Block-Referenz hinzu blockref = msp.add_blockref(block_name, insert_point) blockref.dxf.layer = layer if scale != 1.0: blockref.dxf.xscale = scale blockref.dxf.yscale = scale blockref.dxf.zscale = scale if rotation != 0.0: blockref.dxf.rotation = rotation # Füge Attribute hinzu falls vorhanden if attributes: blockref.add_auto_attribs(attributes) return blockref def query_entities_by_layer(msp, entity_type: str = None, layers: list = None, exclude_layers: list = None): """ Filtert Entities nach Typ und Layer. Args: msp: Modelspace des DXF-Dokuments entity_type: Entity-Typ (z.B. 'INSERT', 'LINE', 'TEXT'), None für alle layers: Liste der zu inkludierenden Layer (None = alle außer exclude_layers) exclude_layers: Liste der zu exkludierenden Layer (None = keine Exklusion) Returns: Liste der gefilterten Entities Examples: # Alle INSERTs auf Layer 'ILS_MOTOR' query_entities_by_layer(msp, 'INSERT', layers=['ILS_MOTOR']) # Alle Entities außer auf Layer '0' query_entities_by_layer(msp, exclude_layers=['0']) # Alle LINEs auf ILS_* Layern (manuelles Filtern nötig) all_lines = query_entities_by_layer(msp, 'LINE') ils_lines = [e for e in all_lines if e.dxf.layer.startswith('ILS_')] """ # Baue Query-String if entity_type: entities = msp.query(entity_type) else: entities = msp # Filtere nach Layern result = [] for entity in entities: layer = entity.dxf.layer # Prüfe Exklusion if exclude_layers and layer in exclude_layers: continue # Prüfe Inklusion if layers is None or layer in layers: result.append(entity) return result def create_layers_from_config(doc, layer_definitions: dict): """ Erstellt mehrere Layer aus einer Konfiguration. Args: doc: DXF-Dokument layer_definitions: Dictionary mit Layer-Definitionen Format: { 'layer_name': { 'color': int (optional), 'linetype': str (optional), 'lineweight': int (optional), 'description': str (optional) }, ... } Example: layer_config = { 'ILS_MOTOR': {'color': 1, 'description': 'Motor layer'}, 'ILS_SENSOR': {'color': 3, 'lineweight': 25}, 'ILS_CABLE': {'color': 5, 'linetype': 'DASHED'} } create_layers_from_config(doc, layer_config) """ for layer_name, props in layer_definitions.items(): ensure_layer_exists( doc, layer_name, color=props.get('color'), linetype=props.get('linetype'), lineweight=props.get('lineweight'), description=props.get('description') ) # ============================================================================ # Dictionary Operations # ============================================================================ def merge_two_dicts(x: dict, y: dict) -> dict: """ Merge two dictionaries, with values from y overwriting those in x. Args: x: First dictionary (base) y: Second dictionary (override values) Returns: New dictionary with merged values """ z = x.copy() z.update(y) return z