Neue Suche nach Doppelrahmen über eigene Pufferklasse
This commit is contained in:
+201
-112
@@ -8,6 +8,7 @@ import re
|
||||
from ezdxf.addons import iterdxf
|
||||
import re
|
||||
from pathlib import Path
|
||||
from shapely.geometry import Point
|
||||
|
||||
|
||||
"""
|
||||
@@ -33,41 +34,6 @@ def merge_two_dicts(x, y):
|
||||
z.update(y)
|
||||
return z
|
||||
|
||||
def get_type_of_name(name):
|
||||
|
||||
SpecialKeys = ["MB", # Ventil
|
||||
"MA", # Motor
|
||||
"BG", # Stausensor
|
||||
"BP", # Schalter Druckluft
|
||||
"QM", # Ventile
|
||||
"BX" # Scanner
|
||||
]
|
||||
|
||||
KabelKey = ["WD", "WF"]
|
||||
DropKeys = [
|
||||
"FC", # Motorschutzschalter
|
||||
"PF", # Leuchtmelder
|
||||
"DI", # Feedback vom Gerät
|
||||
"QA", # Hauptschütz
|
||||
"SF" # Drucktaster
|
||||
]
|
||||
|
||||
prefix = name[:2]
|
||||
if prefix in SpecialKeys:
|
||||
typ = "Sensor"
|
||||
|
||||
# Suche nach Kabel
|
||||
elif prefix in KabelKey:
|
||||
typ = "Kabel"
|
||||
|
||||
# suche nach Items die wir nicht weiter verfolgen
|
||||
elif prefix in DropKeys:
|
||||
typ = "Schaltschrankelement"
|
||||
|
||||
else:
|
||||
typ = "unknown"
|
||||
|
||||
return typ
|
||||
|
||||
def get_type_of_name_cfg(name):
|
||||
prefix = name[:2]
|
||||
@@ -79,7 +45,7 @@ def get_type_of_name_cfg(name):
|
||||
else:
|
||||
return "unknown"
|
||||
|
||||
def get_attributes_of_insert(insert):
|
||||
def get_attributes_of_insert(dinsert, dpos):
|
||||
"""
|
||||
Diese Funktion schaut nach den aktuell definierten Attributen in den Blöcken
|
||||
|
||||
@@ -95,94 +61,223 @@ def get_attributes_of_insert(insert):
|
||||
|
||||
"""
|
||||
id = ""
|
||||
ld = dict()
|
||||
ld = dinsert
|
||||
typ = 'unknown'
|
||||
for attrib in insert.attribs:
|
||||
attr_tag = attrib.dxf.tag
|
||||
attr_text = attrib.dxf.text
|
||||
if len(insert.attribs) == 0:
|
||||
continue # Überspringe Blöcke ohne Attribute
|
||||
#print(f"Attribut Name: {attrib.dxf.tag}, Wert: {attrib.dxf.text}")
|
||||
ld[attr_tag] = attr_text
|
||||
|
||||
if attr_tag == "NAME": # die neueren Bläcke heissen nicht IO, sondern haben einen Namen
|
||||
typ = get_type_of_name_cfg(attr_text)
|
||||
id = attr_text
|
||||
# pos = attrib.dxf.insert #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben
|
||||
# ld["pos"] = (round(pos.x, 1), round(pos.y, 1))
|
||||
if attr_tag == "IO":
|
||||
typ = get_type_of_name_cfg(attr_text)
|
||||
id = attr_text
|
||||
#print(f"-- coord io {id}--: {attrib.dxf.insert}") # position des Blocks
|
||||
pos = attrib.dxf.insert #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben
|
||||
ld["pos"] = (round(pos.x, 1), round(pos.y, 1))
|
||||
|
||||
if attr_tag == "B":
|
||||
# Suche nach Sensoren
|
||||
typ = get_type_of_name_cfg(attr_text)
|
||||
id = attr_text
|
||||
pos = attrib.dxf.insert #Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben
|
||||
ld["pos"] = (round(pos.x, 1), round(pos.y, 1))
|
||||
# die neueren Bläcke heissen nicht IO, sondern haben einen Namen
|
||||
if "NAME" in dinsert:
|
||||
typ = get_type_of_name_cfg(dinsert["NAME"])
|
||||
id = dinsert["NAME"]
|
||||
|
||||
if "IO" in dinsert:
|
||||
attr_text = dinsert["IO"]
|
||||
typ = get_type_of_name_cfg(attr_text)
|
||||
id = dinsert["IO"]
|
||||
pos = dpos["IO"]
|
||||
|
||||
if "REALE_POSITION" in dinsert and dinsert["REALE_POSITION"] == 'x':
|
||||
pos = dpos["REALE_POSITION"]
|
||||
|
||||
if attr_tag == "REALE_POSITION" and attr_text == "x":
|
||||
#typ = get_type_of_name(attr_text)
|
||||
#print(f"-- coord real --: {attrib.dxf.insert}")
|
||||
pos = attrib.dxf.insert #Position Ecke unten links von "x"-Marker auslesen
|
||||
|
||||
# Hoehe und Breite von "x" addieren, um Mittelpunkt zu finden
|
||||
breite_marker = config.getfloat("GetPos-Geom-Sensor", "Breite")
|
||||
hoehe_marker = config.getfloat("GetPos-Geom-Sensor", "Hoehe")
|
||||
midx = pos[0] + breite_marker * 0.5
|
||||
midy = pos[1] + hoehe_marker * 0.5
|
||||
ld["pos"] = (round(midx, 1), round(midy, 1))
|
||||
|
||||
return (ld, id, typ)
|
||||
else:
|
||||
ld["pos"] = (pos[0], pos[1])
|
||||
|
||||
def check_double_ids(ld, id, allSensors):
|
||||
if id in allSensors and "NAME" in ld:
|
||||
return True
|
||||
if id in allSensors and "B" in ld and "B" in allSensors[id]:
|
||||
return True
|
||||
if id in allSensors and "IO" in ld and "IO" in allSensors[id]:
|
||||
return True
|
||||
return False
|
||||
if "B" in dinsert:
|
||||
attr_text = dinsert["B"]
|
||||
typ = get_type_of_name_cfg(attr_text)
|
||||
id = attr_text
|
||||
#Position aufzeichnen und bei Bedarf später mit REAL_POS überschreiben
|
||||
pos = dpos["B"]
|
||||
ld["pos"] = (pos[0], pos[1])
|
||||
|
||||
return (ld, id, typ)
|
||||
|
||||
|
||||
class CompareBuffer:
|
||||
"""ein Puffer um alle als Doppelt zugewiesenen blöcke zwischenzulagern.
|
||||
Anhand von ähnlicher Position und gleicher ID wird dann entschieden,
|
||||
ob die Daten beider dicts zusammen gehören"""
|
||||
def __init__(self):
|
||||
self._wartepuffer = dict()
|
||||
|
||||
def add_block(self, id, buffer):
|
||||
"""adds one block under the buffer list under this id"""
|
||||
if id not in self._wartepuffer:
|
||||
self._wartepuffer[id] = list()
|
||||
self._wartepuffer[id].append(buffer)
|
||||
|
||||
def get_blocks(self, id):
|
||||
if id in self._wartepuffer:
|
||||
return self._wartepuffer[id]
|
||||
else:
|
||||
return []
|
||||
|
||||
def get_block_ids(self)->list:
|
||||
return list(self._wartepuffer.keys())
|
||||
|
||||
|
||||
def set_block(self, id, buffer):
|
||||
if id in self._wartepuffer:
|
||||
del self._wartepuffer[id]
|
||||
self._wartepuffer[id] = buffer
|
||||
|
||||
def dict_compare(self, d1, d2):
|
||||
str1 = json.dumps(d1)
|
||||
str2 = json.dumps(d2)
|
||||
return str1 == str2
|
||||
|
||||
def remove_block(self, id, toRemove):
|
||||
"""removes this block from the list under the given id"""
|
||||
buffers = self.get_blocks(id)
|
||||
l = list()
|
||||
for b in buffers:
|
||||
if self.dict_compare(b, toRemove):
|
||||
pass
|
||||
else:
|
||||
l.append(b)
|
||||
self.set_block(id, l)
|
||||
|
||||
def positions_are_close(self, dict1:dict, dict2:dict, border:float):
|
||||
pos1 = None
|
||||
pos2 = None
|
||||
if "pos" in dict1:
|
||||
pos1 = dict1["pos"]
|
||||
if "pos" in dict2:
|
||||
pos2 = dict2["pos"]
|
||||
if not (pos1 and pos2):
|
||||
return False
|
||||
p1 = Point(pos1[0], pos1[1])
|
||||
p2 = Point(pos2[0], pos2[1] )
|
||||
dist = p1.distance(p2)
|
||||
if dist < border:
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_all_sps_blocks(self, id):
|
||||
"""gives back all blocks with SPS inside to the given id"""
|
||||
buffers = self.get_blocks(id)
|
||||
l = list()
|
||||
for b in buffers:
|
||||
if "SPS" in b:
|
||||
l.append(b)
|
||||
return l
|
||||
|
||||
def get_non_sps_blocks(self, id):
|
||||
"""gives back all blocks without SPS inside to the given id"""
|
||||
buffers = self.get_blocks(id)
|
||||
l = list()
|
||||
for b in buffers:
|
||||
if "SPS" not in b:
|
||||
l.append(b)
|
||||
return l
|
||||
|
||||
def extract_input_positions(insert_iterable) -> tuple:
|
||||
allSensors = dict()
|
||||
allCables = dict()
|
||||
allunknowns = dict()
|
||||
doubleIds = dict()
|
||||
allSchaltschrank = dict()
|
||||
allunknowns = list()
|
||||
|
||||
for insert in insert_iterable:
|
||||
if insert.dxftype() != 'INSERT':
|
||||
continue
|
||||
WP = CompareBuffer()
|
||||
all_inserts, all_positions = attribs_to_dicts(insert_iterable)
|
||||
|
||||
ld, id, typ = get_attributes_of_insert(insert)
|
||||
for insert, pos in zip(all_inserts, all_positions):
|
||||
ld, id, typ = get_attributes_of_insert(insert, pos)
|
||||
|
||||
if typ == "Sensor":
|
||||
if id and "pos" in ld and isinstance(ld["pos"], tuple) and len(ld["pos"]) == 2:
|
||||
if id in allSensors:
|
||||
is_double = check_double_ids(ld, id, allSensors)
|
||||
if not is_double:
|
||||
allSensors[id] = merge_two_dicts(allSensors[id], ld) #Kombiniert alle infos aus dxf und "pos"
|
||||
else:
|
||||
# gib aus wo sich beide gleichlautenden Elemente auf der Zeichnung befinden
|
||||
if id not in doubleIds:
|
||||
doubleIds[id] = []
|
||||
doubleIds[id].append(allSensors[id]["pos"]) # erste Position
|
||||
doubleIds[id].append(ld["pos"]) # zweite Position
|
||||
else:
|
||||
allSensors[id] = ld
|
||||
# wenn NAME enthalten ist, dann ist das ein eindeutiger Bezeichner
|
||||
if "NAME" in ld:
|
||||
allSensors[id] = ld
|
||||
else:
|
||||
# alle anderen Sachen werden dann aus mehreren Rahmen zusammen gesammelt
|
||||
WP.add_block(id, ld)
|
||||
|
||||
elif typ == "Kabel":
|
||||
allCables[id] = ld
|
||||
elif typ == "Schaltschrankelement":
|
||||
allSchaltschrank[id] = ld
|
||||
else:
|
||||
allunknowns[id] = ld
|
||||
allunknowns.append(ld)
|
||||
|
||||
return allSensors, doubleIds
|
||||
# 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(allSensors, WP)
|
||||
|
||||
# die noch übrigen Blöcke melden
|
||||
missing_attribs, doubleIds = get_errors_double_and_attributes(WP)
|
||||
|
||||
return allSensors, doubleIds, missing_attribs
|
||||
|
||||
def get_errors_double_and_attributes(WP):
|
||||
missing_attribs = dict()
|
||||
doubleIds = dict()
|
||||
for id in WP.get_block_ids():
|
||||
blocks = WP.get_blocks(id)
|
||||
# einzelne Blöcke deren Zuordnung nicht gelungen ist, fehlen Angaben in den Attributen, z.B. SPS, B, u.ä.
|
||||
if len(blocks) == 1:
|
||||
given_keys = str(blocks[0].keys())
|
||||
missing_attribs[id] = f"Nur ein Block und/oder fehlende Attribute: {given_keys}"
|
||||
WP.remove_block(id, blocks[0])
|
||||
# alle anderen Angaben sind mindestens zwei oder mehr Blöcke mit derselben Id
|
||||
else:
|
||||
for block in blocks:
|
||||
if id not in doubleIds:
|
||||
doubleIds[id] = []
|
||||
doubleIds[id].append(block['pos'])
|
||||
return missing_attribs, doubleIds
|
||||
|
||||
def allocate_blocks_together(allSensors:dict, WP:CompareBuffer):
|
||||
"""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()
|
||||
for id in all_sensors_ids:
|
||||
blks_sps = WP.get_all_sps_blocks(id)
|
||||
blks_other = WP.get_non_sps_blocks(id)
|
||||
|
||||
for block_with_sps in blks_sps:
|
||||
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):
|
||||
newId = create_new_id(id, block_with_sps, block_without_sps) # hier das Präfix davor
|
||||
allSensors[newId] = 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, dict1, dict2):
|
||||
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
|
||||
return f"{sps_praefix}:{id}"
|
||||
|
||||
def attribs_to_dicts(insert_iterable):
|
||||
all_inserts = list()
|
||||
all_positions = list()
|
||||
for insert in insert_iterable:
|
||||
if insert.dxftype() != 'INSERT':
|
||||
continue
|
||||
itemdata = dict()
|
||||
positions = dict()
|
||||
typ = 'unknown'
|
||||
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, 1), round(pos.y, 1), round(pos.z, 1))
|
||||
if len(itemdata) > 0:
|
||||
all_inserts.append(itemdata)
|
||||
all_positions.append(positions)
|
||||
return all_inserts, all_positions
|
||||
|
||||
def get_input_positions(msp) -> tuple:
|
||||
return extract_input_positions(msp.query('INSERT'))
|
||||
@@ -193,7 +288,7 @@ def get_input_positions_iter(dxf_path) -> tuple:
|
||||
def create_mappings(positions:dict) -> tuple:
|
||||
unterverteiler_pfad = ""
|
||||
dnamen = dict()
|
||||
# sammle die Sensoren mit ihren zugehörigen Unterverteilern
|
||||
# s
|
||||
sensor2unterverteiler = dict()
|
||||
warnings = dict()
|
||||
for sensorname,v in positions.items():
|
||||
@@ -205,17 +300,11 @@ def create_mappings(positions:dict) -> tuple:
|
||||
#print(unterverteiler_pfad)
|
||||
|
||||
# Pfad zur Karte splitten. Dieser hat z.B. den Inhalt "=AH01+UH02-KF1FDI7"
|
||||
pattern = r"^=([A-Z]+\d+)([+\-])([A-Z]+\d+)([+\-])([A-Z]+\d+.)$"
|
||||
match = re.match(pattern, unterverteiler_pfad)
|
||||
if match:
|
||||
anlage = match.group(1)
|
||||
verteiler = match.group(3)
|
||||
karte = match.group(5)
|
||||
# match.group(1) # AH01
|
||||
# match.group(2) # +
|
||||
# match.group(3) # UH02
|
||||
# match.group(4) # -
|
||||
# match.group(5) # KF1FDI7
|
||||
matches = re.findall(r'[^\-+=]+', unterverteiler_pfad.lstrip('='))
|
||||
if matches:
|
||||
anlage = matches[0]
|
||||
verteiler = matches[1]
|
||||
karte = matches[2]
|
||||
else:
|
||||
warnings[sensorname] = f"Ungültiger Pfad in Kennzeichnung: {unterverteiler_pfad}"
|
||||
continue
|
||||
@@ -586,9 +675,9 @@ if __name__ == '__main__':
|
||||
if args.sensors:
|
||||
# Sensoren auslesen
|
||||
if use_iter:
|
||||
res_sens, res_double = get_input_positions_iter(dxf_path)
|
||||
res_sens, res_double, missing_attribs = get_input_positions_iter(dxf_path)
|
||||
else:
|
||||
res_sens, res_double = get_input_positions(msp)
|
||||
res_sens, res_double, missing_attribs = get_input_positions(msp)
|
||||
|
||||
if args.errors and len(res_double) > 0:
|
||||
print("Duplicate blocks found. Writing errors-file.")
|
||||
|
||||
Reference in New Issue
Block a user