650 lines
20 KiB
Python
650 lines
20 KiB
Python
"""
|
|
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 draw_symbol_frames(doc, msp, symbols, symbol_width=1210, symbol_height=381, frame_layer="SYMBOL_FRAMES"):
|
|
"""
|
|
Zeichnet Rahmen um alle Symbole zur Visualisierung der Symbol-Grenzen.
|
|
|
|
Args:
|
|
doc: DXF-Dokument
|
|
msp: Modelspace
|
|
symbols: Liste von Symbol-Dictionaries mit 'entity', 'position', 'attributes'
|
|
symbol_width: Feste Breite des Symbols (Standard: 1210 für MA-Frame)
|
|
symbol_height: Feste Höhe des Symbols (Standard: 381)
|
|
frame_layer: Layer-Name für die Rahmen (Standard: "SYMBOL_FRAMES")
|
|
|
|
Returns:
|
|
Anzahl der gezeichneten Rahmen
|
|
"""
|
|
print("\n" + "="*60)
|
|
print("Zeichne Symbol-Rahmen zur Visualisierung")
|
|
print("="*60)
|
|
|
|
# Erstelle Layer falls nicht vorhanden
|
|
if frame_layer not in doc.layers:
|
|
doc.layers.add(frame_layer, color=3) # Farbe 3 = Grün
|
|
print(f" Layer '{frame_layer}' erstellt (Farbe: Grün)")
|
|
|
|
frame_count = 0
|
|
|
|
for symbol in symbols:
|
|
# Hole Position und Attribute
|
|
pos = symbol['position']
|
|
attribs = symbol['attributes']
|
|
|
|
# Hole IO-Attribut um die Zeichen-Anzahl zu bestimmen
|
|
io_value = attribs.get('IO', '')
|
|
if not io_value:
|
|
# Versuche andere Attribute als Fallback
|
|
for key in ['KENNZEICHNUNG', 'NAME']:
|
|
if key in attribs and attribs[key]:
|
|
io_value = attribs[key]
|
|
break
|
|
|
|
# Bestimme Dimensionen basierend auf Zeichen-Anzahl
|
|
# 6 Zeichen (MA-1@@) = 1210, 7 Zeichen (BG-1@@@) = 1410
|
|
char_count = len(str(io_value))
|
|
if char_count >= 7:
|
|
width = 1410 # Multi-Frame (BG/MB/POT)
|
|
else:
|
|
width = symbol_width # MA-Frame (Standard)
|
|
height = symbol_height
|
|
|
|
# Berechne Eckpunkte (Position = linke untere Ecke des Symbols)
|
|
# Symbol erstreckt sich von (x, y) nach rechts und nach oben
|
|
x_min = pos[0]
|
|
x_max = pos[0] + width
|
|
y_min = pos[1]
|
|
y_max = pos[1] + height
|
|
|
|
# Erstelle Rechteck als LWPOLYLINE
|
|
points = [
|
|
(x_min, y_min),
|
|
(x_max, y_min),
|
|
(x_max, y_max),
|
|
(x_min, y_max),
|
|
(x_min, y_min) # Schließe das Rechteck
|
|
]
|
|
|
|
msp.add_lwpolyline(points, dxfattribs={
|
|
'layer': frame_layer,
|
|
'color': 3, # Grün
|
|
'closed': True
|
|
})
|
|
|
|
frame_count += 1
|
|
|
|
print(f" {frame_count} Symbol-Rahmen gezeichnet")
|
|
return frame_count
|
|
|
|
|
|
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
|