diff --git a/README.md b/README.md
index b8209ee..b18578e 100644
--- a/README.md
+++ b/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
diff --git a/src/simaticml_decoder/fold.py b/src/simaticml_decoder/fold.py
index ac7f07c..d6fb61a 100644
--- a/src/simaticml_decoder/fold.py
+++ b/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:
+ ```` 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:
diff --git a/src/simaticml_decoder/instructions.py b/src/simaticml_decoder/instructions.py
index 382398e..fa80e1d 100644
--- a/src/simaticml_decoder/instructions.py
+++ b/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")),
}
diff --git a/tests/test_fold.py b/tests/test_fold.py
index 72e08e3..05232bc 100644
--- a/tests/test_fold.py
+++ b/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) — 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: 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():
+ # 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():
+ # 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():
+ # 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, {}, []))