# -*- coding: utf-8 -*- """Tests fuer lib/merge_hw_layout.py - TRO-JSON + Hardware-JSON -> Anlagen-JSON. Synthetische Anlage nach den Mubea-Befunden (doc/E-Planung_Bruecke.md): Separator-Sensor auf dem Separator, Stopperventil ~270 mm daneben mit Nummer = Sensor + 1, Stausensor ~1390 mm vom Ausschleuselement, Kreisel mit ueberlappender Speicherlinie (Gefaellestrecke), Vario-Strecke mit Antrieb. """ import json from pathlib import Path import pytest import merge_hw_layout from dxf_registration import Transform, load_transform, save_transform from merge_hw_layout import ( build_anlage, load_overrides, load_rules, ) from hw_import import HardwareDocument, HwDevice, conveyor_kind_of, conveyor_of, pair_drives from model import AnlageDocument, Analysis, DerivedTro, LayoutPart, TroDocument, derive_slots RULES = Path(__file__).resolve().parent.parent / "cfg" / "e_planning.ini" IDENT = Transform(dx=0.0, dy=0.0) SEPS = {f"000{n}": ((n - 1) * 800.0, 0.0) for n in range(1, 7)} def _part(layout_id, kind, x, y, bbox=None, name=""): return LayoutPart(layout_id=layout_id, kind=kind, name=name or f"{kind}{layout_id}", position=(x, y, 0.0), bbox=bbox) def _tro(tro_id, type_name, seps, hosts=()): return DerivedTro(tro_id=tro_id, type_name=type_name, label=tro_id, hosts=list(hosts), separators=list(seps), scanners=[], reason="Test") def layout() -> TroDocument: parts = [_part(i, "Separator", x, y) for i, (x, y) in SEPS.items()] parts += [ _part("0030", "Ausschleus", 0.0, 2000.0), _part("0010", "Kreisel", 2000.0, 5000.0, (6000.0, 800.0, 1.0), "Kreisel3"), # Speicherlinie, deren Huellbox den Kreisel ueberdeckt (wie bei Mubea) _part("0040", "Gefaellestrecke", 2000.0, 5000.0, (6000.0, 1200.0, 1.0), "GF_1"), _part("0020", "Strecke", 2000.0, -3000.0, (6000.0, 800.0, 1.0), "VF_1"), ] tros = [ _tro("TRO01", "1Sep", ["0001"], ["0010-R"]), _tro("TRO02", "1Sep1Swi", ["0002"], ["0010-R"]), _tro("TRO03", "Vario", ["0003"], ["0020"]), _tro("TRO04", "1Sep", ["0004"]), _tro("TRO05", "1Sep", ["0005"]), _tro("TRO06", "1Sep", ["0006"]), ] return TroDocument.from_analysis(Analysis(tros=tros), source="x.csv", parts=parts) def _dev(bmk, verw, pos=None, text="", tia=True, cabinet=False, direction="in"): return HwDevice(bmk=bmk, verw=verw, bezeichnung=text, kennzeichnung="=A1+UH01-KF102", pos=pos, cabinet=cabinet, direction=direction, address="%E000.0" if direction == "in" else "%A000.0", constant=(f"cIn{bmk}" if direction == "in" else f"cOut{bmk}") if tia else "", conveyor=conveyor_of(text), conveyor_kind=conveyor_kind_of(text), sources=["tia", "positions"] if tia else ["positions"]) def hardware(shift=(0.0, 0.0)) -> HardwareDocument: """E-Zeichnung = Layout - shift (shift = Lagebezug E -> Layout).""" def at(x, y): return (x - shift[0], y - shift[1]) devices = [] for n in range(1, 6): x, y = SEPS[f"000{n}"] devices.append(_dev(f"BG0{n}01", "IN SEP", at(x, y))) if n != 3: # MB0502 fehlt im TIA-Import -> Widerspruch devices.append(_dev(f"MB0{n}02", "SEP", at(x, y + 270.0), tia=(n != 5), direction="out")) devices += [ _dev("BG0601", "STAU_REF-LT", at(0.0, 2000.0 + 1390.0)), _dev("BG0701", "SYNC P1", at(1000.0, 5100.0), "PRUEFUNG BT (KR-K0102 PUFFER)"), _dev("BG0702", "BT BELADEN", at(3000.0, 4900.0), "BT BELADEN KR-K0102"), _dev("BG0801", "KETTENBRUCH", at(1000.0, -3000.0), "KETTENBRUCH CONVEYOR #106"), _dev("BG0802", "UEBERGABE", at(3000.0, -3000.0), "UEBERGABE CONVEYOR #106"), _dev("BG0900", "UEBERNAHME", at(50000.0, 0.0), "SENSOR UEBERNAHME"), _dev("MA0202", "MOT", None, "ANTRIEB KREISEL-K0102", cabinet=True, direction="out"), _dev("FC0202", "", None, "MOTORSCHUTZ KREISEL-K0102", cabinet=True), _dev("MA0210", "MOT", None, "PIN CONVEYOR-#106", cabinet=True, direction="out"), _dev("FC0210", "", None, "MOTORSCHUTZ CONVEYOR #106", cabinet=True), ] return HardwareDocument(project="X", positions_file="X_positionsconv.json", positions_sha256="abc", devices=devices, drives=pair_drives(devices)) def run(doc=None, hw=None, overrides=None, transform=IDENT, tmp_path=None): rules, anchors = load_rules(RULES) return build_anlage(doc or layout(), hw or hardware(), rules, anchors, tmp_path or Path("."), "x_tro.json", "X_hw.json", overrides, transform=transform) # --------------------------------------------------------------------------- # Slots # --------------------------------------------------------------------------- def test_separator_sensor_raeumlich_stopper_ueber_nummer(): anlage = run() sensor = anlage.binding("TRO01", "separator1", "sensor") stopper = anlage.binding("TRO01", "separator1", "stopper") assert (sensor.bmk, sensor.constant, sensor.rule, sensor.status) == ( "BG0101", "cInBG0101", "raeumlich", "zugeordnet") assert (stopper.bmk, stopper.constant, stopper.rule) == ("MB0102", "cOutMB0102", "nummer") assert stopper.layout_id == "0001" def test_bmk_nicht_im_tia_ist_widerspruch(): b = run().binding("TRO05", "separator1", "stopper") assert (b.bmk, b.status, b.note) == ("MB0502", "widerspruch", "BMK nicht im TIA-Import") def test_pflichtrollen_ohne_bmk_sind_offen_optionale_nicht(): anlage = run() roles = {(b.slot, b.role): b.status for b in anlage.bindings if b.tro == "TRO06"} assert roles == {("separator1", "sensor"): "offen", ("separator1", "stopper"): "offen"} # die Weiche hat noch keine Layout-ID -> Ventil offen assert anlage.binding("TRO02", "switch1", "valve").status == "offen" def test_bauteil_ohne_slot_bleibt_sichtbar(): loose = [b for b in run().bindings if not b.tro] assert [(b.role, b.layout_id, b.bmk, b.status) for b in loose] == [ ("Ausschleus.stau", "0030", "BG0601", "ohne_slot")] def test_unzugeordnete_feldgeraete(): assert "BG0900" in run().unassigned assert "BG0101" not in run().unassigned # --------------------------------------------------------------------------- # Foerderer # --------------------------------------------------------------------------- def test_foerderer_per_mehrheit_und_art(): """K0102 nennt "KR" -> nur Kreisel stimmen, obwohl die Speicherlinie den Kreisel ueberdeckt; #106 nennt "CONVEYOR" -> Strecke.""" anlage = run() by_name = {c.eplan_name: c for c in anlage.conveyors} k = by_name["K0102"] assert (k.layout_id, k.layout_name, k.votes, k.total, k.status) == ( "0010", "Kreisel3", 2, 2, "zugeordnet") assert (k.motor, k.protection) == ("MA0202", "FC0202") assert by_name["K0106"].layout_id == "0020" assert anlage.conveyor_for("0010-R").eplan_name == "K0102" def test_vario_antrieb_aus_dem_foerderer(): anlage = run() motor = anlage.binding("TRO03", "drive", "motor") schutz = anlage.binding("TRO03", "drive", "motorProtection") assert (motor.bmk, motor.rule, motor.constant) == ("MA0210", "mehrheit", "cOutMA0210") assert schutz.bmk == "FC0210" assert anlage.binding("TRO03", "drive", "senJam").status == "offen" def test_foerderer_mit_zu_wenig_stimmen_bleibt_offen(): hw = hardware() hw.devices = [d for d in hw.devices if d.bmk != "BG0802"] k = next(c for c in run(hw=hw).conveyors if c.eplan_name == "K0106") assert k.status == "offen" and "unsicher: 1 von 1" in k.note # --------------------------------------------------------------------------- # Handkorrekturen # --------------------------------------------------------------------------- def test_handkorrekturen(tmp_path): ini = tmp_path / "hw_overrides.ini" ini.write_text( "[x.csv.bindings]\n" "TRO06.separator1.sensor = BG0401 ; vom Nachbarn umhaengen\n" "TRO06.separator1.stopper = MB4711\n" "TRO02.separator1.stopper = -\n" "TRO99.separator1.sensor = BG0101\n" "[x.csv.conveyors]\n" "K0102 = -\n" "[andere.csv.bindings]\n" "TRO01.separator1.sensor = BG0999\n", encoding="utf-8") anlage = run(overrides=load_overrides(ini, "x.csv")) moved = anlage.binding("TRO06", "separator1", "sensor") assert (moved.bmk, moved.rule, moved.source, moved.status) == ( "BG0401", "hand", "hw_overrides.ini", "zugeordnet") # automatisch vergeben war BG0401 an TRO04 - dort jetzt offen assert anlage.binding("TRO04", "separator1", "sensor").status == "offen" unknown = anlage.binding("TRO06", "separator1", "stopper") assert (unknown.status, unknown.note) == ("widerspruch", "BMK unbekannt in der Hardwareplanung") assert anlage.binding("TRO02", "separator1", "stopper").status == "offen" assert any("TRO99.separator1" in w for w in anlage.warnings) # andere Anlage nicht angewendet assert anlage.binding("TRO01", "separator1", "sensor").bmk == "BG0101" k = next(c for c in anlage.conveyors if c.eplan_name == "K0102") assert (k.status, k.rule, k.layout_id) == ("offen", "hand", "") # ohne Foerderer kein Antrieb ueber K0102, der Vario-Antrieb bleibt assert anlage.binding("TRO03", "drive", "motor").bmk == "MA0210" def test_falscher_schluessel_in_handkorrektur(tmp_path): ini = tmp_path / "hw_overrides.ini" ini.write_text("[x.csv.bindings]\nTRO01.sensor = BG0101\n", encoding="utf-8") with pytest.raises(ValueError, match=".."): load_overrides(ini, "x.csv") # --------------------------------------------------------------------------- # Speicher: Linien-Slots # --------------------------------------------------------------------------- def test_speicher_bekommt_linien_slots_und_beide_slots_die_zuordnung(): store = _tro("TRO01", "PinStore_Auto", ["0001", "0002", "0003"], ["0016", "0017", "0018"]) assert [(s.slot, s.layout_id) for s in derive_slots(store)] == [ ("separator1", "0001"), ("line1", "0001"), ("line2", "0002"), ("line3", "0003"), ("scanner1", "")] doc = TroDocument.from_analysis(Analysis(tros=[store]), source="x.csv", parts=[_part(i, "Separator", *SEPS[i]) for i in ("0001", "0002", "0003")]) anlage = run(doc=doc) a = anlage.binding("TRO01", "separator1", "sensor") b = anlage.binding("TRO01", "line1", "sensor") assert a.bmk == b.bmk == "BG0101" and a.status == b.status == "zugeordnet" assert anlage.binding("TRO01", "line2", "stopper").bmk == "MB0202" # --------------------------------------------------------------------------- # JSON und Lauf # --------------------------------------------------------------------------- def test_ohne_parts_kein_lauf(): doc = layout() doc.parts = [] with pytest.raises(ValueError, match="parts"): run(doc=doc) def test_anlage_json_rundreise(): anlage = run() again = AnlageDocument.from_json(anlage.to_json()) assert again == anlage data = json.loads(anlage.to_json()) assert data["schema"] == "sps_skel-anlage-v1" assert data["layout"]["schema"] == "sps_skel-tro-v4" assert data["layout"]["parts"][0]["layoutId"] == "0001" def test_gespeicherter_lagebezug_gilt_nur_fuer_denselben_dateistand(tmp_path): t = Transform(dx=5.0, dy=6.0, csv_file="x.csv", dxf_file="E.json", fingerprint="aaa") save_transform(tmp_path, t) assert load_transform(tmp_path, "x.csv", "E.json", "aaa").dx == 5.0 assert load_transform(tmp_path, "x.csv", "E.json", "bbb") is None # ohne Pruefsumme (tro_annotate) wie bisher ueber den Dateinamen assert load_transform(tmp_path, "x.csv", "E.json").dx == 5.0 # alte Ablage ohne Pruefsumme gilt fuer einen Aufrufer mit Pruefsumme als veraltet save_transform(tmp_path, Transform(dx=1.0, dy=1.0, csv_file="x.csv", dxf_file="Alt.json")) assert load_transform(tmp_path, "x.csv", "Alt.json", "aaa") is None def test_cli_mit_lagebezug(tmp_path, monkeypatch, capsys): """Ende zu Ende: E-Zeichnung um (100, 50) verschoben; Lagebezug wird ermittelt, mit Pruefsumme abgelegt und beim zweiten Lauf wiederverwendet.""" results, cfg = tmp_path / "results", tmp_path / "cfg" results.mkdir() cfg.mkdir() (cfg / "e_planning.ini").write_text(RULES.read_text(encoding="utf-8"), encoding="utf-8") (results / "x_tro.json").write_text(layout().to_json(), encoding="utf-8") hw = hardware(shift=(100.0, 50.0)) (results / "X_hw.json").write_text(hw.to_json(), encoding="utf-8") monkeypatch.setenv("SKEL_RESULTS", str(results)) monkeypatch.setenv("SKEL_CFG", str(cfg)) assert merge_hw_layout.main(["--tro", "x_tro.json", "--hw", "X_hw.json"]) == 0 first = capsys.readouterr().out assert "Lagebezug neu ermittelt" in first anlage = AnlageDocument.from_json((results / "x_anlage.json").read_text(encoding="utf-8")) assert (anlage.registration.dx, anlage.registration.dy) == (100.0, 50.0) assert anlage.binding("TRO01", "separator1", "sensor").bmk == "BG0101" assert (results / "x_anlage.md").is_file() stored = json.loads((cfg / "dxf_registration.json").read_text(encoding="utf-8")) assert stored["x.csv|X_positionsconv.json"]["fingerprint"] == "abc" assert merge_hw_layout.main(["--tro", "x_tro.json", "--hw", "X_hw.json"]) == 0 assert "Lagebezug neu ermittelt" not in capsys.readouterr().out # neuer Stand der Positionsdatei, gleicher Name -> neu ermitteln hw.positions_sha256 = "neu" (results / "X_hw.json").write_text(hw.to_json(), encoding="utf-8") assert merge_hw_layout.main(["--tro", "x_tro.json", "--hw", "X_hw.json"]) == 0 assert "Lagebezug neu ermittelt" in capsys.readouterr().out def test_cli_alte_tro_json_ohne_parts(tmp_path, monkeypatch, capsys): results = tmp_path / "results" results.mkdir() doc = layout() doc.parts = [] (results / "x_tro.json").write_text(doc.to_json(), encoding="utf-8") (results / "X_hw.json").write_text(hardware().to_json(), encoding="utf-8") monkeypatch.setenv("SKEL_RESULTS", str(results)) monkeypatch.setenv("SKEL_CFG", str(Path(RULES).parent)) assert merge_hw_layout.main(["--tro", "x_tro.json", "--hw", "X_hw.json"]) == 1 assert "tro_flow --json neu erzeugen" in capsys.readouterr().err