# -*- coding: utf-8 -*- """Tests fuer lib/e_planning.py - Zuordnung Layout <-> E-Planung. Die Geometrie ist synthetisch und nachgebaut nach den Befunden an Mubea (doc/E-Planung_Bruecke.md): Speicherlinien im 800-mm-Raster, Sensor auf dem Separator, Ventil ~270 mm daneben, Stausensor konstant ~1390 mm vom Ausschleuselement. Der Lagebezug ist hier die Identitaet. """ import json import shutil from pathlib import Path import pytest import scl_skeleton from dxf_registration import Transform from e_planning import ( EDevice, EplanDocument, LayoutPoint, Rule, assign, load_positionsdraw, load_rules, pair_drives, ) from scl_context import build_context from scl_skeleton import load_document FIXTURES = Path(__file__).resolve().parent / "fixtures" / "scl" RULES = Path(__file__).resolve().parent.parent / "cfg" / "e_planning.ini" IDENT = Transform(dx=0.0, dy=0.0) def _dev(bmk: str, verw: str, x: float, y: float, **extra) -> EDevice: return EDevice(bmk=bmk, verw=verw, pos=(x, y), kennzeichnung="=A1+UH01-KF107", **extra) def _rolle(assignments, layout_id: str, role: str): return next((a for a in assignments if a.layout_id == layout_id and a.role == role), None) # --------------------------------------------------------------------------- # Einlesen # --------------------------------------------------------------------------- def test_positionsdraw_einlesen(tmp_path): datei = tmp_path / "x_positionsdraw.json" datei.write_text(json.dumps({ "sensors": { "BG0642@UNKNOWN": {"B": "BG0642", "VERW": "IN SEP", "BEZEICHNUNG": "TEIL IN SEPARATOR STR. 20", "KENNZEICHNUNG": "=A1+UH01-KF107", "pos": [1.0, 2.0]}, }, "schaltschrank_elemente": { # Derselbe Motorschutz zweimal: einmal mit, einmal ohne Attribute "FC0200@": {"IO": "FC0200", "VERW": "MOTORSCHUTZ", "BEZEICHNUNG": "MOTORSCHUTZ CONVEYER #100", "pos": [5, 5]}, "FC0200": {"B": "FC0200", "pos": [6, 6]}, }, "double_ids": {"BG0516": [[0, 0], [1, 1]]}, }), encoding="utf-8") warnungen: list[str] = [] geraete = {d.bmk: d for d in load_positionsdraw(datei, warnungen)} assert set(geraete) == {"BG0642", "FC0200"} bg = geraete["BG0642"] assert (bg.verw, bg.pos, bg.cabinet) == ("IN SEP", (1.0, 2.0), False) assert (bg.controller, bg.station, bg.prefix, bg.number) == ("UH01", "KF107", "BG", 642) assert geraete["FC0200"].verw == "MOTORSCHUTZ" assert geraete["FC0200"].cabinet assert any("BG0516" in w for w in warnungen) def test_regeln_aus_der_projektkonfiguration(): regeln, anker = load_rules(RULES) rollen = [r.role for r in regeln] assert anker == ["IN SEP"] # Reihenfolge = Dateireihenfolge; der Sensor muss vor dem Ventil kommen assert rollen.index("Separator.sensor") < rollen.index("Separator.stopper") stopper = next(r for r in regeln if r.role == "Separator.stopper") assert (stopper.layout, stopper.number_from, stopper.number_offset) == ( ["Separator"], "Separator.sensor", 1) def test_nummer_von_muss_eine_regel_sein(tmp_path): ini = tmp_path / "r.ini" ini.write_text("[Separator.stopper]\nverw = SEP\nnummer_von = Separator.gibtsnicht\n", encoding="utf-8") with pytest.raises(ValueError, match="gibtsnicht"): load_rules(ini) # --------------------------------------------------------------------------- # Zuordnung # --------------------------------------------------------------------------- SENSOR = Rule(role="Separator.sensor", layout=["Separator"], verw=["IN SEP"], prefix="BG", max_mm=300) STOPPER = Rule(role="Separator.stopper", layout=["Separator"], verw=["SEP"], prefix="MB", number_from="Separator.sensor", number_offset=1, max_mm=600, target_mm=270) def _speicher(): """Drei Speicherlinien im 800-mm-Raster. Das Ventil jeder Linie liegt 530 mm von ihrem Separator - und nur 270 mm vom Separator der Nachbarlinie. Raeumlich waere das falsch; die Nummernregel (MB = BG + 1) muss gewinnen. """ punkte = [LayoutPoint(f"00{i}", "Separator", 0.0, 800.0 * i) for i in (1, 2, 3)] geraete = [] for i, bg in ((1, 642), (2, 644), (3, 646)): geraete.append(_dev(f"BG0{bg}", "IN SEP", 0.0, 800.0 * i)) geraete.append(_dev(f"MB0{bg + 1}", "SEP", 0.0, 800.0 * i + 530.0)) return punkte, geraete def test_ventil_ueber_die_nummer_nicht_ueber_den_abstand(): punkte, geraete = _speicher() ergebnis = assign(punkte, geraete, [SENSOR, STOPPER], IDENT) for sep, bg, mb in (("001", "BG0642", "MB0643"), ("002", "BG0644", "MB0645"), ("003", "BG0646", "MB0647")): assert _rolle(ergebnis, sep, "Separator.sensor").bmk == bg stopper = _rolle(ergebnis, sep, "Separator.stopper") assert (stopper.bmk, stopper.rule) == (mb, "nummer") def test_jedes_e_bauteil_nur_einmal(): """Zwei Separatoren, ein Sensor dazwischen: nur der naehere bekommt ihn.""" punkte = [LayoutPoint("A", "Separator", 0, 0), LayoutPoint("B", "Separator", 0, 250)] geraete = [_dev("BG0100", "IN SEP", 0, 100)] ergebnis = assign(punkte, geraete, [SENSOR], IDENT) assert [(a.layout_id, a.bmk) for a in ergebnis] == [("A", "BG0100")] assert ergebnis[0].runner_up is None def test_zu_weit_weg_bleibt_offen(): punkte = [LayoutPoint("A", "Separator", 0, 0)] geraete = [_dev("BG0100", "IN SEP", 0, 400)] assert assign(punkte, geraete, [SENSOR], IDENT) == [] def test_sollabstand_schlaegt_kleinsten_abstand(): """Mubea 0080: ein fremder Stausensor liegt 269 mm nah, der eigene 1391 mm.""" stau = Rule(role="Ausschleus.stau", layout=["Ausschleus"], verw=["STAU*"], prefix="BG", max_mm=1600, target_mm=1390) punkte = [LayoutPoint("0080", "Ausschleus", 0, 0)] geraete = [_dev("BG0525", "STAU_REF-LT", 269, 0), _dev("BG0550", "STAU_REF-LT", -1391, 0)] ergebnis = assign(punkte, geraete, [stau], IDENT) assert [a.bmk for a in ergebnis] == ["BG0550"] def test_lagebezug_wird_angewendet(): """E-Koordinaten werden mit dem Lagebezug ins CSV gebracht.""" punkte = [LayoutPoint("A", "Separator", 1000, 2000)] geraete = [_dev("BG0100", "IN SEP", 0, 0)] ergebnis = assign(punkte, geraete, [SENSOR], Transform(dx=1000, dy=2000)) assert ergebnis[0].bmk == "BG0100" and ergebnis[0].distance == 0 def test_schrankelemente_werden_nicht_raeumlich_zugeordnet(): punkte = [LayoutPoint("A", "Separator", 0, 0)] geraete = [_dev("BG0100", "IN SEP", 0, 0, cabinet=True)] assert assign(punkte, geraete, [SENSOR], IDENT) == [] def test_motor_und_motorschutz_ueber_die_nummer(): geraete = [_dev("MA0200", "MOT_0,37kW", 0, 0, bezeichnung="PIN CONVEYOR#100"), EDevice(bmk="FC0200", verw="MOTORSCHUTZ", cabinet=True), _dev("MA0300", "MOT", 0, 0)] antriebe = {d.motor: d.protection for d in pair_drives(geraete)} assert antriebe == {"MA0200": "FC0200", "MA0300": ""} # --------------------------------------------------------------------------- # Bis in die SCL # --------------------------------------------------------------------------- def _eplan_fuer_mubea(csv_file: str = "mubea.csv") -> EplanDocument: """Zuordnung fuer Separator 0012 (TRO02 der Fixture) - nur Sensor und Ventil.""" punkte = [LayoutPoint("0012", "Separator", 0, 0)] geraete = [_dev("BG0712", "IN SEP", 0, 0), _dev("MB0713", "SEP", 0, 300)] return EplanDocument(csv_file=csv_file, eplan_file="Mubea_positionsdraw.json", created="2026-01-01 00:00", assignments=assign(punkte, geraete, [SENSOR, STOPPER], IDENT)) def test_kontext_setzt_namen_ein_und_loescht_den_merker(): document = load_document(FIXTURES / "mubea_tro.json") kontext = build_context(document, "mubea_tro.json", eplan=_eplan_fuer_mubea()) tro02 = next(t for t in kontext.tros if t.id == "TRO02") sep = tro02.separators[0] assert (sep.layout, sep.sensor.value, sep.sensor.todo) == ("0012", "cInBG0712", False) assert (sep.stopper.value, sep.stopper.todo) == ("MB0713", False) # Was die E-Planung nicht liefert, bleibt offen assert sep.no.todo tro01 = next(t for t in kontext.tros if t.id == "TRO01") assert tro01.separators[0].sensor.todo assert kontext.eplan_source.startswith("Mubea_positionsdraw.json") def test_kontext_setzt_namen_auch_in_speicherlinien_ein(): """PinStore_Auto: die Linien sind Separatoren im Layout - ihre E-Planung-Namen kommen in store.lines an (und damit ins PinStore-Template), nicht nur in den Separatorblock.""" from model import Analysis, DerivedTro, TroDocument from scl_skeleton import TEMPLATE_DIR, SclTemplates linien = ["0066", "0069"] store = DerivedTro(tro_id="TRO04", type_name="PinStore_Auto", label="Speicher", hosts=["0016", "0017"], separators=linien, scanners=[], reason="Test", predecessors=["TRO03"], successors=["TRO05"]) punkte = [LayoutPoint("0066", "Separator", 0, 0), LayoutPoint("0069", "Separator", 1000, 0)] geraete = [_dev("BG0766", "IN SEP", 0, 0), _dev("MB0767", "SEP", 0, 300), _dev("BG0769", "IN SEP", 1000, 0), _dev("MB0770", "SEP", 1000, 300)] eplan = EplanDocument(csv_file="mubea.csv", eplan_file="Mubea_positionsdraw.json", assignments=assign(punkte, geraete, [SENSOR, STOPPER], IDENT)) document = TroDocument.from_analysis(Analysis(tros=[store]), source="mubea.csv") kontext = build_context(document, "mubea_tro.json", eplan=eplan) zeilen = kontext.tros[0].store.lines assert [(z.layout, z.sensor.value, z.stopper.value) for z in zeilen] == [ ("0066", "cInBG0766", "MB0767"), ("0069", "cInBG0769", "MB0770")] assert not any(n.todo for z in zeilen for n in (z.sensor, z.stopper)) text = SclTemplates(TEMPLATE_DIR).render(kontext.tros[0]) assert '#TRO04.arInSep[#nActNo].stSenTrolley := "DB_Inputs".Sensors["cInBG0766"];\n' in text assert '"MB0770" := #TRO04.arOutSeparatorStore[2];\n' in text def _lauf(tmp_path, monkeypatch, eplan: EplanDocument, *weitere: str) -> tuple[int, Path]: shutil.copy(FIXTURES / "mubea_tro.json", tmp_path / "mubea_tro.json") (tmp_path / "mubea_eplan.json").write_text(eplan.to_json(), encoding="utf-8") monkeypatch.setenv("SKEL_RESULTS", str(tmp_path)) rc = scl_skeleton.main(["--json", "mubea_tro.json", "--eplan", "mubea_eplan.json", *weitere]) return rc, tmp_path / "mubea_FB_Main.scl" def test_scl_skeleton_mit_eplan(tmp_path, monkeypatch): rc, scl = _lauf(tmp_path, monkeypatch, _eplan_fuer_mubea()) text = scl.read_text(encoding="utf-8") assert rc == 0 assert '#TRO02.stInSeparator1.stSenInSep := "DB_Inputs".Sensors["cInBG0712"];\n' in text assert 'xOutStopper1 => "MB0713",\n' in text assert "// E-Planung: Mubea_positionsdraw.json" in text def test_scl_skeleton_mit_eplan_und_templates(tmp_path, monkeypatch): """Die Namen aus der E-Planung kommen auch in den Templates an (TRO02 = 1Sep).""" rc, scl = _lauf(tmp_path, monkeypatch, _eplan_fuer_mubea(), "--templates") text = scl.read_text(encoding="utf-8") assert rc == 0 region = text[text.index("REGION TRO02 "):] region = region[:region.index("\nEND_REGION")] assert "Aufbau nach Template FB_ILS_MTRO_1Sep.scl.j2" in region assert '#TRO02.stInSeparator1.stSenInSep := "DB_Inputs".Sensors["cInBG0712"];\n' in region assert 'xOutStopper1 => "MB0713",\n' in region def test_scl_skeleton_lehnt_fremde_zuordnung_ab(tmp_path, monkeypatch): rc, scl = _lauf(tmp_path, monkeypatch, _eplan_fuer_mubea("anderes.csv")) assert rc == 1 assert not scl.exists() def test_eplan_json_rundlauf(): doc = _eplan_fuer_mubea() zurueck = EplanDocument.from_json(doc.to_json()) assert zurueck == doc assert json.loads(doc.to_json())["schema"] == "sps_skel-eplan-v1" assert set(zurueck.for_layout("0012")) == {"sensor", "stopper"}