feat: add A (AND-junction) and F-system safety instructions

README.md

@@ -36,11 +36,13 @@ authoritative-looking SCL is the worst possible failure.
 
 ## Scope
 
-Covered: FC/FB blocks; LAD/FBD folding (series→AND, `O`→OR, fan-out, `Negated`→NOT,
+Covered: FC/FB blocks; LAD/FBD folding (series→AND, `O`→OR, `A`→AND, fan-out,
+`Negated`→NOT on contacts *and* on individual AND/OR/box input pins,
 daisy-chained coils, structural latch detection); SCL network reconstruction;
 ground-truth interface types; cross-reference table; the instruction set seen in
-real V21 samples (Contact, Coil/SCoil/RCoil, `O`, Move/Add/Inc, comparisons,
-Rs/Sr, P/N edges, TON family, user FC/FB calls).
+real V21 samples (Contact, Coil/SCoil/RCoil, `O`, `A`, Move/Add/Inc, comparisons,
+Rs/Sr, P/N edges, TON family, F-system safety boxes ACK_GL/ESTOP1/SFDOOR/FDBACK,
+user FC/FB calls).
 
 Deferred (parsed losslessly, rendering flagged): GRAPH/SFC and STL networks;
 absolute-addressing / array / `Operation`-template rendering. Output is

src/simaticml_decoder/fold.py

@@ -240,13 +240,18 @@ def _eval_part_out_uncached(self, uid: str, pin: str):
             return self._eval_edge(part, uid, spec)
         if cat == Category.OR_JUNCTION:
             return self._eval_or(uid)
+        if cat == Category.AND_JUNCTION:
+            return self._eval_and(uid)
         if cat == Category.FLIPFLOP:
             return self._operand_varref(uid) or _POWER
         if cat == Category.BOX:
             # Reading a box output pin (Q/ET/OUT/RET_VAL): a member of its
             # instance, or a synthetic name when there is no instance.
             label = self._box_label(part)
-            return ir.VarRef(name=f"{label}.{pin}", uid=uid)
+            value = ir.VarRef(name=f"{label}.{pin}", uid=uid)
+            if pin in part.negated_pins:
+                return ir.Not(operand=value)
+            return value
         return ir.Unhandled(part.name, uid, "unclassified category")
 
     def _eval_edge(self, part: model.Part, uid: str, spec) -> ir.Expr:
@@ -263,12 +268,41 @@ def _eval_edge(self, part: model.Part, uid: str, spec) -> ir.Expr:
         return _and([incoming, edge])
 
     def _eval_or(self, uid: str) -> ir.Expr:
-        branches = []
-        for pin in self._sorted_input_pins(uid):
-            branches.append(_materialize(self._driver_expr(uid, pin)))
+        branches = self._junction_branches(uid)
         if not branches:
             return _POWER
-        return _factor_or(branches)
+        return self._negate_out(uid, _factor_or(branches))
+
+    def _eval_and(self, uid: str) -> ir.Expr:
+        branches = self._junction_branches(uid)
+        if not branches:
+            return _POWER
+        return self._negate_out(uid, _and(branches))
+
+    def _negate_out(self, uid: str, expr: ir.Expr) -> ir.Expr:
+        """Wrap in Not when the junction's own ``out`` pin is negated (NAND/NOR:
+        ``<Negated Name="out"/>`` on the A/O Part itself, distinct from a
+        negated input branch)."""
+        part = self.net.parts.get(uid)
+        if part is not None and "out" in part.negated_pins:
+            return ir.Not(operand=expr)
+        return expr
+
+    def _junction_branches(self, uid: str) -> list[ir.Expr]:
+        """Collect and materialize an OR/AND junction's input branches.
+
+        A ``Negated Name="inN"`` on the junction Part itself (distinct from a
+        negated *Contact* feeding it) wraps that one branch in ``Not`` — TIA lets
+        you negate an individual input pin of an AND/OR box directly.
+        """
+        part = self.net.parts.get(uid)
+        branches: list[ir.Expr] = []
+        for pin in self._sorted_input_pins(uid):
+            expr = _materialize(self._driver_expr(uid, pin))
+            if part is not None and pin in part.negated_pins:
+                expr = ir.Not(operand=expr)
+            branches.append(expr)
+        return branches
 
     # -- helpers used by eval ---------------------------------------------- #
     def _driver_expr(self, uid: str, pin: str | None):
@@ -340,7 +374,7 @@ def _statement_for_part(self, uid: str, part: model.Part) -> ir.Statement | None
             return self._make_flipflop(uid, part)
         if cat == Category.BOX:
             return self._make_box(uid, part, spec)
-        # Contacts / comparisons / edges / OR are sub-expressions, not statements.
+        # Contacts / comparisons / edges / AND / OR are sub-expressions, not statements.
         return None
 
     def _make_assign(self, uid: str, part: model.Part, spec) -> ir.Statement:
@@ -371,10 +405,16 @@ def _make_flipflop(self, uid: str, part: model.Part) -> ir.Statement:
             self.warnings.append(f"Network {self.index}: {note} (UId {uid})")
             return ir.Unhandled(part_name=part.name, uid=uid, note=note)
         set_pin, reset_pin = _FLIPFLOP_PINS.get(part.name, ("s", "r"))
+        set_expr = _materialize(self._driver_expr(uid, set_pin))
+        if set_pin in part.negated_pins:
+            set_expr = ir.Not(operand=set_expr)
+        reset_expr = _materialize(self._driver_expr(uid, reset_pin))
+        if reset_pin in part.negated_pins:
+            reset_expr = ir.Not(operand=reset_expr)
         return ir.FlipFlop(
             target=target,
-            set_expr=_materialize(self._driver_expr(uid, set_pin)),
-            reset_expr=_materialize(self._driver_expr(uid, reset_pin)),
+            set_expr=set_expr,
+            reset_expr=reset_expr,
             reset_priority=(part.name == "Rs"),
             uid=uid,
         )
@@ -391,8 +431,13 @@ def _make_box(self, uid: str, part: model.Part, spec) -> ir.Statement:
             if pin == spec.power_in:  # the en pin
                 if expr is not _POWER:
                     enable = _materialize(expr)
+                    if pin in part.negated_pins:
+                        enable = ir.Not(operand=enable)
             else:
-                inputs[pin] = _materialize(expr)
+                value = _materialize(expr)
+                if pin in part.negated_pins:
+                    value = ir.Not(operand=value)
+                inputs[pin] = value
 
         for (u, pin), sinks in self.pin_out_sinks.items():
             if u != uid or pin == spec.power_out:

src/simaticml_decoder/instructions.py

@@ -22,6 +22,7 @@ class Category(str, Enum):
     POWER_FLOW = "power_flow"  # in/out, contact-like: passes or blocks power
     COIL = "coil"  # in/out, writes its operand
     OR_JUNCTION = "or_junction"  # merges parallel branches (Part Name="O")
+    AND_JUNCTION = "and_junction"  # merges parallel branches (Part Name="A")
     COMPARISON = "comparison"  # pre/out, contact-like compare
     EDGE = "edge"  # pre/out or in/out, rising/falling detection
     FLIPFLOP = "flipflop"  # named set/reset inputs -> q
@@ -63,6 +64,7 @@ def _box(name, pins, render=None):
     "SCoil": _coil("SCoil", "S"),  # set coil  ( S )
     "RCoil": _coil("RCoil", "R"),  # reset coil ( R )
     "O": Spec("O", Category.OR_JUNCTION, None, "out", "OR", ("in1", "in2", "out")),
+    "A": Spec("A", Category.AND_JUNCTION, None, "out", "AND", ("in1", "in2", "out")),
     # --- comparisons (pre/out, contact-like) ------------------------------ #
     "Lt": _cmp("Lt", "<"),
     "Le": _cmp("Le", "<="),
@@ -97,6 +99,20 @@ def _box(name, pins, render=None):
     "TP": _box("TP", ("IN", "PT", "Q", "ET")),
     # --- system FC example ------------------------------------------------- #
     "RD_LOC_T": _box("RD_LOC_T", ("en", "RET_VAL", "OUT", "eno")),
+    # --- F-system safety instructions (S7 F-blocks; en/eno boxes like any --- #
+    # other system FB, just with F-specific pin names). Pin tuples below list
+    # only the pins observed as wired in practice, not necessarily the full
+    # official interface — unlisted wired pins are still discovered dynamically
+    # from the network, since `pins` is informative only (see lookup()).
+    "ACK_GL": _box("ACK_GL", ("en", "ACK_GLOB", "eno")),
+    # ESTOP1 v1.6 confirmed against a real TIA V16 export (temp/Main_Safety_RTG1.xml):
+    # en/E_STOP/ACK_NEC/ACK/TIME_DEL in, Q/Q_DELAY/ACK_REQ/DIAG/eno out.
+    "ESTOP1": _box(
+        "ESTOP1",
+        ("en", "E_STOP", "ACK_NEC", "ACK", "TIME_DEL", "Q", "Q_DELAY", "ACK_REQ", "DIAG", "eno"),
+    ),
+    "SFDOOR": _box("SFDOOR", ("en", "IN1", "eno")),
+    "FDBACK": _box("FDBACK", ("en", "ON", "QBAD_FIO", "eno")),
 }
 
 

tests/test_fold.py

@@ -6,7 +6,8 @@
 
 from __future__ import annotations
 
-from simaticml_decoder import fold, ir, model
+from simaticml_decoder import fold, instructions, ir, model
+from simaticml_decoder.instructions import Category
 from simaticml_decoder.model import Endpoint
 from simaticml_decoder.model import EndpointKind as EK
 from simaticml_decoder.model import Wire
@@ -141,6 +142,218 @@ def test_daisy_chained_coils_share_upstream_flow():
     assert by_target["#b"].value.name == "#a"
 
 
+def test_and_junction_is_nary_and():
+    parts = {
+        "1": model.Part(uid="1", name="Contact"),
+        "2": model.Part(uid="2", name="Contact"),
+        "3": model.Part(uid="3", name="A"),
+        "4": model.Part(uid="4", name="Coil"),
+    }
+    accesses = {"10": _sym("10", "a"), "11": _sym("11", "b"), "12": _sym("12", "y")}
+    wires = [
+        Wire(uid="w1", endpoints=[_pr(), _nc("1", "in")]),
+        Wire(uid="w2", endpoints=[_ic("10"), _nc("1", "operand")]),
+        Wire(uid="w3", endpoints=[_pr(), _nc("2", "in")]),
+        Wire(uid="w4", endpoints=[_ic("11"), _nc("2", "operand")]),
+        Wire(uid="w5", endpoints=[_nc("1", "out"), _nc("3", "in1")]),
+        Wire(uid="w6", endpoints=[_nc("2", "out"), _nc("3", "in2")]),
+        Wire(uid="w7", endpoints=[_nc("3", "out"), _nc("4", "in")]),
+        Wire(uid="w8", endpoints=[_ic("12"), _nc("4", "operand")]),
+    ]
+    stmt = fold.fold_network(_net(parts, accesses, wires)).statements[0]
+    assert isinstance(stmt.value, ir.And)
+    assert sorted(o.name for o in stmt.value.operands) == ["#a", "#b"]
+
+
+def test_and_junction_negated_pin_wraps_not():
+    # TIA lets you negate an individual input pin directly on an AND/OR box
+    # (distinct from a negated Contact feeding it) — <Negated Name="in2"/> on
+    # the "A" Part itself.
+    parts = {
+        "1": model.Part(uid="1", name="A", negated_pins=["in2"]),
+        "2": model.Part(uid="2", name="Coil"),
+    }
+    accesses = {"10": _sym("10", "a"), "11": _sym("11", "b"), "12": _sym("12", "y")}
+    wires = [
+        Wire(uid="w1", endpoints=[_ic("10"), _nc("1", "in1")]),
+        Wire(uid="w2", endpoints=[_ic("11"), _nc("1", "in2")]),
+        Wire(uid="w3", endpoints=[_nc("1", "out"), _nc("2", "in")]),
+        Wire(uid="w4", endpoints=[_ic("12"), _nc("2", "operand")]),
+    ]
+    stmt = fold.fold_network(_net(parts, accesses, wires)).statements[0]
+    assert isinstance(stmt.value, ir.And)
+    negated = [o for o in stmt.value.operands if isinstance(o, ir.Not)]
+    assert len(negated) == 1
+    assert negated[0].operand.name == "#b"
+
+
+def test_or_junction_negated_pin_wraps_not():
+    # Regression: the same per-pin negation applies to "O", not just "A".
+    parts = {
+        "1": model.Part(uid="1", name="O", negated_pins=["in1"]),
+        "2": model.Part(uid="2", name="Coil"),
+    }
+    accesses = {"10": _sym("10", "a"), "11": _sym("11", "b"), "12": _sym("12", "y")}
+    wires = [
+        Wire(uid="w1", endpoints=[_ic("10"), _nc("1", "in1")]),
+        Wire(uid="w2", endpoints=[_ic("11"), _nc("1", "in2")]),
+        Wire(uid="w3", endpoints=[_nc("1", "out"), _nc("2", "in")]),
+        Wire(uid="w4", endpoints=[_ic("12"), _nc("2", "operand")]),
+    ]
+    stmt = fold.fold_network(_net(parts, accesses, wires)).statements[0]
+    assert isinstance(stmt.value, ir.Or)
+    negated = [o for o in stmt.value.operands if isinstance(o, ir.Not)]
+    assert len(negated) == 1
+    assert negated[0].operand.name == "#a"
+
+
+def test_and_junction_negated_out_pin_wraps_not():
+    # NAND: <Negated Name="out"/> on the "A" Part itself negates the junction's
+    # result, not one of its input branches — distinct from an in-pin negation.
+    parts = {
+        "1": model.Part(uid="1", name="A", negated_pins=["out"]),
+        "2": model.Part(uid="2", name="Coil"),
+    }
+    accesses = {"10": _sym("10", "a"), "11": _sym("11", "b"), "12": _sym("12", "y")}
+    wires = [
+        Wire(uid="w1", endpoints=[_ic("10"), _nc("1", "in1")]),
+        Wire(uid="w2", endpoints=[_ic("11"), _nc("1", "in2")]),
+        Wire(uid="w3", endpoints=[_nc("1", "out"), _nc("2", "in")]),
+        Wire(uid="w4", endpoints=[_ic("12"), _nc("2", "operand")]),
+    ]
+    stmt = fold.fold_network(_net(parts, accesses, wires)).statements[0]
+    assert isinstance(stmt.value, ir.Not)
+    assert isinstance(stmt.value.operand, ir.And)
+    assert sorted(o.name for o in stmt.value.operand.operands) == ["#a", "#b"]
+
+
+def test_or_junction_negated_out_pin_wraps_not():
+    # NOR: same as the NAND case above, but for "O".
+    parts = {
+        "1": model.Part(uid="1", name="O", negated_pins=["out"]),
+        "2": model.Part(uid="2", name="Coil"),
+    }
+    accesses = {"10": _sym("10", "a"), "11": _sym("11", "b"), "12": _sym("12", "y")}
+    wires = [
+        Wire(uid="w1", endpoints=[_ic("10"), _nc("1", "in1")]),
+        Wire(uid="w2", endpoints=[_ic("11"), _nc("1", "in2")]),
+        Wire(uid="w3", endpoints=[_nc("1", "out"), _nc("2", "in")]),
+        Wire(uid="w4", endpoints=[_ic("12"), _nc("2", "operand")]),
+    ]
+    stmt = fold.fold_network(_net(parts, accesses, wires)).statements[0]
+    assert isinstance(stmt.value, ir.Not)
+    assert isinstance(stmt.value.operand, ir.Or)
+    assert sorted(o.name for o in stmt.value.operand.operands) == ["#a", "#b"]
+
+
+def test_flipflop_negated_set_reset_pins_wrap_not():
+    # <Negated Name="r1"/> on an "Sr" Part negates its reset input, the same way
+    # per-pin negation applies to junction and box inputs.
+    parts = {"1": model.Part(uid="1", name="Sr", negated_pins=["r1"])}
+    accesses = {
+        "10": _sym("10", "set_cond"),
+        "11": _sym("11", "reset_cond"),
+        "12": _sym("12", "q"),
+    }
+    wires = [
+        Wire(uid="w1", endpoints=[_ic("10"), _nc("1", "s")]),
+        Wire(uid="w2", endpoints=[_ic("11"), _nc("1", "r1")]),
+        Wire(uid="w3", endpoints=[_ic("12"), _nc("1", "operand")]),
+    ]
+    stmt = fold.fold_network(_net(parts, accesses, wires)).statements[0]
+    assert isinstance(stmt, ir.FlipFlop)
+    assert isinstance(stmt.set_expr, ir.VarRef) and stmt.set_expr.name == "#set_cond"
+    assert isinstance(stmt.reset_expr, ir.Not)
+    assert stmt.reset_expr.operand.name == "#reset_cond"
+
+
+def test_box_negated_en_pin_wraps_not():
+    # <Negated Name="en"/> on a box Part negates its enable condition, the same
+    # way per-pin negation applies to the box's other input pins.
+    parts = {"1": model.Part(uid="1", name="Move", negated_pins=["en"])}
+    accesses = {"10": _sym("10", "inhibit"), "11": _sym("11", "src"), "12": _sym("12", "dst")}
+    wires = [
+        Wire(uid="w1", endpoints=[_ic("10"), _nc("1", "en")]),
+        Wire(uid="w2", endpoints=[_ic("11"), _nc("1", "in")]),
+        Wire(uid="w3", endpoints=[_nc("1", "out1"), _ic("12")]),
+    ]
+    stmt = fold.fold_network(_net(parts, accesses, wires)).statements[0]
+    assert isinstance(stmt, ir.BoxCall)
+    assert isinstance(stmt.enable, ir.Not)
+    assert stmt.enable.operand.name == "#inhibit"
+
+
+def test_box_negated_output_pin_read_wraps_not():
+    # <Negated Name="Q"/> on a TON Part negates its own Q output where that
+    # output is read as a sub-expression feeding downstream logic.
+    parts = {
+        "1": model.Part(uid="1", name="TON", negated_pins=["Q"]),
+        "2": model.Part(uid="2", name="Coil"),
+    }
+    accesses = {"10": _sym("10", "y")}
+    wires = [
+        Wire(uid="w1", endpoints=[_nc("1", "Q"), _nc("2", "in")]),
+        Wire(uid="w2", endpoints=[_ic("10"), _nc("2", "operand")]),
+    ]
+    statements = fold.fold_network(_net(parts, accesses, wires)).statements
+    stmt = next(s for s in statements if isinstance(s, ir.Assign))
+    assert isinstance(stmt.value, ir.Not)
+    assert stmt.value.operand.name == "#TON_1.Q"
+
+
+def test_f_system_safety_boxes_are_catalogued():
+    # ACK_GL, ESTOP1, SFDOOR, FDBACK: S7 F-system safety instructions, boxed
+    # like any other system FB (en/eno), just with F-specific pin names.
+    for name in ("ACK_GL", "ESTOP1", "SFDOOR", "FDBACK"):
+        spec = instructions.lookup(name)
+        assert spec is not None, f"{name} missing from catalog"
+        assert spec.category == Category.BOX
+
+
+def test_estop1_pins_match_real_export():
+    # ESTOP1 v1.6, confirmed against a real TIA V16 export: en/E_STOP/ACK_NEC/
+    # ACK/TIME_DEL in, Q/Q_DELAY/ACK_REQ/DIAG/eno out. Regression for a prior
+    # placeholder entry that listed generic in1/in2 pins that don't exist on
+    # this block.
+    spec = instructions.lookup("ESTOP1")
+    assert spec is not None
+    assert set(spec.pins) == {
+        "en", "E_STOP", "ACK_NEC", "ACK", "TIME_DEL", "Q", "Q_DELAY", "ACK_REQ", "DIAG", "eno",
+    }
+
+
+def test_ack_gl_box_call_folds_with_wired_pin():
+    parts = {"1": model.Part(uid="1", name="ACK_GL")}
+    accesses = {"10": _sym("10", "reset")}
+    wires = [Wire(uid="w1", endpoints=[_ic("10"), _nc("1", "ACK_GLOB")])]
+    stmt = fold.fold_network(_net(parts, accesses, wires)).statements[0]
+    assert isinstance(stmt, ir.BoxCall)
+    assert stmt.instruction == "ACK_GL"
+    assert stmt.inputs["ACK_GLOB"].name == "#reset"
+
+
+def test_fdback_box_negated_input_wraps_not():
+    # Real-world motivation: FDBACK's ON/QBAD_FIO pins are commonly wired
+    # negated (feedback contacts are normally-closed) — box inputs must apply
+    # per-pin negation the same way junctions do.
+    parts = {
+        "1": model.Part(
+            uid="1", name="FDBACK", negated_pins=["ON", "QBAD_FIO"]
+        )
+    }
+    accesses = {"10": _sym("10", "cmd"), "11": _sym("11", "healthy")}
+    wires = [
+        Wire(uid="w1", endpoints=[_ic("10"), _nc("1", "ON")]),
+        Wire(uid="w2", endpoints=[_ic("11"), _nc("1", "QBAD_FIO")]),
+    ]
+    stmt = fold.fold_network(_net(parts, accesses, wires)).statements[0]
+    assert isinstance(stmt, ir.BoxCall)
+    assert isinstance(stmt.inputs["ON"], ir.Not)
+    assert stmt.inputs["ON"].operand.name == "#cmd"
+    assert isinstance(stmt.inputs["QBAD_FIO"], ir.Not)
+    assert stmt.inputs["QBAD_FIO"].operand.name == "#healthy"
+
+
 def test_unknown_instruction_folds_to_unhandled_and_warns():
     parts = {"1": model.Part(uid="1", name="Frobnicate")}
     logic = fold.fold_network(_net(parts, {}, []))