FIX: getpositions verwendet jetzt auch das gegeben SPS Präfix. Ausgeschlossene Schaltschrankelemente können jetzt auch von ioconvert verwendet werden.
This commit is contained in:
+58
-4
@@ -81,6 +81,9 @@ def get_attributes_of_insert(d_insert: dict, d_pos: dict) -> tuple[dict, str, st
|
|||||||
attr_text = d_insert["IO"]
|
attr_text = d_insert["IO"]
|
||||||
typ = get_type_of_name_cfg(attr_text)
|
typ = get_type_of_name_cfg(attr_text)
|
||||||
id_ = d_insert["IO"]
|
id_ = d_insert["IO"]
|
||||||
|
# Sensoren werden später gemerged mit den anderen Blöcken des Rahmens mit A,B,C, usw
|
||||||
|
if "SPS" in d_insert and typ != "Sensor":
|
||||||
|
id_ = id_+"@"+d_insert["SPS"]
|
||||||
pos = d_pos["IO"]
|
pos = d_pos["IO"]
|
||||||
|
|
||||||
if "REALE_POSITION" in d_insert and d_insert["REALE_POSITION"] == 'x':
|
if "REALE_POSITION" in d_insert and d_insert["REALE_POSITION"] == 'x':
|
||||||
@@ -183,6 +186,25 @@ class CompareBuffer:
|
|||||||
return l
|
return l
|
||||||
|
|
||||||
def extract_input_positions(insert_iterable) -> tuple[dict, dict, dict, dict]:
|
def extract_input_positions(insert_iterable) -> tuple[dict, dict, dict, dict]:
|
||||||
|
"""
|
||||||
|
Extracts and organizes input positions from an iterable of inserts.
|
||||||
|
|
||||||
|
This function processes a list of insert objects (e.g., from a DXF file), classifies them by type
|
||||||
|
(Sensor, Kabel, Schaltschrankelement, or unknown), and organizes them into dictionaries keyed by their IDs.
|
||||||
|
For sensors, it handles the special case where sensor information may be split across multiple blocks
|
||||||
|
(e.g., IO and A,B,C blocks) and merges them if necessary. It also collects information about missing
|
||||||
|
attributes and duplicate IDs for further error handling.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
insert_iterable: An iterable of insert objects to process.
|
||||||
|
|
||||||
|
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_sensors = dict()
|
||||||
all_cables = dict()
|
all_cables = dict()
|
||||||
all_schaltschrank = dict()
|
all_schaltschrank = dict()
|
||||||
@@ -219,6 +241,17 @@ def extract_input_positions(insert_iterable) -> tuple[dict, dict, dict, dict]:
|
|||||||
return all_sensors, all_schaltschrank, double_ids, missing_attribs
|
return all_sensors, all_schaltschrank, double_ids, missing_attribs
|
||||||
|
|
||||||
def get_errors_double_and_attributes(wp: CompareBuffer) -> tuple[dict, dict]:
|
def get_errors_double_and_attributes(wp: CompareBuffer) -> tuple[dict, dict]:
|
||||||
|
"""
|
||||||
|
Analyze the CompareBuffer for blocks that could not be merged or assigned.
|
||||||
|
|
||||||
|
This function inspects the CompareBuffer for:
|
||||||
|
- 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.
|
||||||
|
"""
|
||||||
missing_attribs = dict()
|
missing_attribs = dict()
|
||||||
double_ids = dict()
|
double_ids = dict()
|
||||||
for id_ in wp.get_block_ids():
|
for id_ in wp.get_block_ids():
|
||||||
@@ -237,8 +270,21 @@ def get_errors_double_and_attributes(wp: CompareBuffer) -> tuple[dict, dict]:
|
|||||||
return missing_attribs, double_ids
|
return missing_attribs, double_ids
|
||||||
|
|
||||||
def allocate_blocks_together(all_sensors: dict, wp: CompareBuffer) -> None:
|
def allocate_blocks_together(all_sensors: dict, wp: CompareBuffer) -> None:
|
||||||
"""geht alle gemerkten Sensoren durch die gleich heissen.
|
"""
|
||||||
Falls ein SPS Präfix angegeben wird, wird es zur Id hinzugefügt und als neuer Name gemerkt """
|
Merge sensor blocks with the same ID that are split across multiple DXF blocks.
|
||||||
|
|
||||||
|
This function iterates over all sensor IDs stored in the CompareBuffer. For each ID, it looks for blocks
|
||||||
|
with and without an "SPS" prefix. If two such blocks have positions that are close to each other (within a
|
||||||
|
specified tolerance), they are considered to represent the same physical sensor. The function then merges
|
||||||
|
their information into a single dictionary, creates a new unique sensor ID by appending the SPS prefix,
|
||||||
|
and adds the merged sensor to the all_sensors dictionary. The merged blocks are then removed from the buffer.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
all_sensors (dict): Dictionary to store merged sensor information, keyed by unique sensor IDs.
|
||||||
|
wp (CompareBuffer): Buffer containing blocks that could not be directly assigned, grouped by ID.
|
||||||
|
"""
|
||||||
|
#geht alle gemerkten Sensoren durch die gleich heissen.
|
||||||
|
# Falls ein SPS Präfix angegeben wird, wird es zur Id hinzugefügt und als neuer Name gemerkt
|
||||||
all_sensors_ids = wp.get_block_ids()
|
all_sensors_ids = wp.get_block_ids()
|
||||||
for id_ in all_sensors_ids:
|
for id_ in all_sensors_ids:
|
||||||
blks_sps = wp.get_all_sps_blocks(id_)
|
blks_sps = wp.get_all_sps_blocks(id_)
|
||||||
@@ -264,6 +310,14 @@ def create_new_id(id_: str, dict1: dict, dict2: dict) -> str:
|
|||||||
return f"{id_}@{sps_praefix}"
|
return f"{id_}@{sps_praefix}"
|
||||||
|
|
||||||
def attribs_to_dicts(insert_iterable) -> tuple[list, list]:
|
def attribs_to_dicts(insert_iterable) -> tuple[list, list]:
|
||||||
|
"""
|
||||||
|
Wandelt eine Iterable von INSERT-Objekten in zwei Listen um:
|
||||||
|
- all_inserts: Liste von Dictionaries mit Attribut-Tags und deren Textwerten
|
||||||
|
- all_positions: Liste von Dictionaries mit Attribut-Tags und deren (x, y, z)-Positionen (jeweils gerundet auf eine Nachkommastelle)
|
||||||
|
|
||||||
|
Jeder Eintrag in den Listen entspricht einem INSERT-Block mit Attributen.
|
||||||
|
Blöcke ohne Attribute werden übersprungen.
|
||||||
|
"""
|
||||||
all_inserts = list()
|
all_inserts = list()
|
||||||
all_positions = list()
|
all_positions = list()
|
||||||
for insert in insert_iterable:
|
for insert in insert_iterable:
|
||||||
@@ -285,10 +339,10 @@ def attribs_to_dicts(insert_iterable) -> tuple[list, list]:
|
|||||||
all_positions.append(positions)
|
all_positions.append(positions)
|
||||||
return all_inserts, all_positions
|
return all_inserts, all_positions
|
||||||
|
|
||||||
def get_input_positions(msp) -> tuple[dict, dict, dict]:
|
def get_input_positions(msp) -> tuple[dict, dict, dict, dict]:
|
||||||
return extract_input_positions(msp.query('INSERT'))
|
return extract_input_positions(msp.query('INSERT'))
|
||||||
|
|
||||||
def get_input_positions_iter(dxf_path) -> tuple[dict, dict, dict]:
|
def get_input_positions_iter(dxf_path) -> tuple[dict, dict, dict, dict]:
|
||||||
return extract_input_positions(iterdxf.modelspace(dxf_path))
|
return extract_input_positions(iterdxf.modelspace(dxf_path))
|
||||||
|
|
||||||
def create_mappings(positions: dict) -> tuple[dict, dict]:
|
def create_mappings(positions: dict) -> tuple[dict, dict]:
|
||||||
|
|||||||
+17
-17
@@ -6,24 +6,24 @@ import json
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
def process_item(sname, sdata):
|
def process_item(sname, sdata):
|
||||||
data = {
|
data = {
|
||||||
"id": sdata.get("IO", ""),
|
"id": sdata.get("IO", ""),
|
||||||
"bezeichnung": sdata.get("BEZEICHNUNG", ""),
|
"bezeichnung": sdata.get("BEZEICHNUNG", ""),
|
||||||
"verwendung": sdata.get("VERW", ""),
|
"verwendung": sdata.get("VERW", ""),
|
||||||
"kennzeichnung": sdata.get("KENNZEICHNUNG", ""),
|
"kennzeichnung": sdata.get("KENNZEICHNUNG", ""),
|
||||||
"text_d": sdata.get("TEXT-D", ""),
|
"text_d": sdata.get("TEXT-D", ""),
|
||||||
"text_e": sdata.get("TEXT-E", ""),
|
"text_e": sdata.get("TEXT-E", ""),
|
||||||
"text_es": sdata.get("TEXT-ES", ""),
|
"text_es": sdata.get("TEXT-ES", ""),
|
||||||
"text_f": sdata.get("TEXT-F", ""),
|
"text_f": sdata.get("TEXT-F", ""),
|
||||||
"sps": sdata.get("SPS", None),
|
"sps": sdata.get("SPS", None),
|
||||||
}
|
}
|
||||||
reihenfolge = ["id", "bezeichnung", "verwendung", "kennzeichnung", "text_d", "text_e", "text_es", "text_f"]
|
reihenfolge = ["id", "bezeichnung", "verwendung", "kennzeichnung", "text_d", "text_e", "text_es", "text_f"]
|
||||||
# CSV-String erzeugen
|
# CSV-String erzeugen
|
||||||
csvstr = "','".join(str(data[key]) for key in reihenfolge)
|
csvstr = "','".join(str(data[key]) for key in reihenfolge)
|
||||||
csvstr = "'"+csvstr+"'"
|
csvstr = "'"+csvstr+"'"
|
||||||
|
|
||||||
sps_nr = sdata.get("SPS", None)
|
sps_nr = sdata.get("SPS", None)
|
||||||
return csvstr, sps_nr
|
return csvstr, sps_nr
|
||||||
|
|
||||||
def prepare_data(rawdata:dict):
|
def prepare_data(rawdata:dict):
|
||||||
sensors = rawdata["sensors"]
|
sensors = rawdata["sensors"]
|
||||||
|
|||||||
+5
-9
@@ -1,16 +1,12 @@
|
|||||||
import os
|
|
||||||
import json
|
import json
|
||||||
import argparse
|
|
||||||
import heapq
|
|
||||||
import math
|
|
||||||
import matplotlib.pyplot as plt
|
|
||||||
import networkx as nx
|
import networkx as nx
|
||||||
import shapely
|
from shapely.geometry import Point
|
||||||
from shapely.geometry import LineString, Point
|
|
||||||
from shapely.ops import nearest_points
|
|
||||||
from plant import Anlage
|
from plant import Anlage
|
||||||
import configparser
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
import argparse
|
||||||
|
import os
|
||||||
|
import configparser
|
||||||
|
import matplotlib.pyplot as plt
|
||||||
|
|
||||||
|
|
||||||
# Funktionen
|
# Funktionen
|
||||||
|
|||||||
Reference in New Issue
Block a user