From 909ba7da0845609ac6c522fa74ac15416211b1e1 Mon Sep 17 00:00:00 2001 From: GruppoVeneta Date: Fri, 10 Jul 2026 12:51:53 +0200 Subject: [PATCH 1/9] fix(csv): accept decoded time drum units in round-trip --- src/laddercodec/instructions/drum.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/laddercodec/instructions/drum.py b/src/laddercodec/instructions/drum.py index f4d1cb7..a04da45 100644 --- a/src/laddercodec/instructions/drum.py +++ b/src/laddercodec/instructions/drum.py @@ -1,4 +1,4 @@ -"""Drum — event drum and time drum instructions (type marker 0x271B). +"""Drum — event drum and time drum instructions (type marker 0x271B). Binary class name: ``"Drum"``. Drum types: event_drum (event-driven), time_drum (time-driven). @@ -27,13 +27,13 @@ # Event drum: jog group (base, reset, jump, jog) _EVENT_FC_JOG = (8494, 8495, 8496, 8497) -# Time drum: unit → (base, reset) +# Time drum: unit → (base, reset) _TIME_FC: dict[str, tuple[int, int]] = { "Tms": (8455, 8456), "Td": (8484, 8485), } -# Time unit name → index string (same convention as Timer). +# Time unit name → index string (same convention as Timer). _DRUM_UNIT_TO_INDEX: dict[str, str] = { "Tms": "0", "Ts": "1", @@ -342,7 +342,10 @@ def parse_af_call(call: AfCall) -> Drum: events_or_presets = _parse_simple_list(call.kwargs.get("presets", "[]")) accumulator = call.kwargs.get("accumulator", "") unit = call.kwargs.get("unit", "") - if unit not in _TIME_FC: + # CSV parsing and round-trip validation do not build a binary blob. + # Accept every unit the decoder can emit, even when binary encoding for + # that unit is not implemented in ``_TIME_FC`` yet. + if unit not in _DRUM_UNIT_TO_INDEX: raise ValueError(f"Unsupported time drum unit: {unit!r}") return Drum( @@ -367,3 +370,4 @@ def parse_af_call(call: AfCall) -> Drum: pin_names=(".reset", ".jump", ".jog"), min_csv_rows=4, ) + From 43d7b0a88327b0f23c6fffb91723ef38ac388457 Mon Sep 17 00:00:00 2001 From: GruppoVeneta Date: Fri, 10 Jul 2026 12:52:24 +0200 Subject: [PATCH 2/9] test(csv): accept known time drum units --- tests/csv/test_converter.py | 28 +++++++++++++++++++++++++--- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/tests/csv/test_converter.py b/tests/csv/test_converter.py index 74a7ccc..081c9ae 100644 --- a/tests/csv/test_converter.py +++ b/tests/csv/test_converter.py @@ -30,6 +30,7 @@ CompareContact, Contact, Counter, + Drum, ForLoop, Next, RawInstruction, @@ -335,7 +336,8 @@ def test_timer_rejects_unknown_unit(self) -> None: with pytest.raises(ValueError, match="Unknown timer unit"): af_node_to_token(node) - def test_time_drum_rejects_unsupported_unit(self) -> None: + @pytest.mark.parametrize("unit", ["Tms", "Ts", "Tm", "Th", "Td"]) + def test_time_drum_accepts_known_unit(self, unit: str) -> None: node = AfCall( name="time_drum", args=(), @@ -343,7 +345,26 @@ def test_time_drum_rejects_unsupported_unit(self) -> None: kwargs={ "outputs": "[C211,C212]", "presets": "[100,200]", - "unit": "Ts", + "unit": unit, + "pattern": "[[1,0],[0,1]]", + "current_step": "DS186", + "accumulator": "TD5", + "completion_flag": "C213", + }, + ) + result = af_node_to_token(node) + assert isinstance(result, Drum) + assert result.unit == unit + + def test_time_drum_rejects_unknown_unit(self) -> None: + node = AfCall( + name="time_drum", + args=(), + known=True, + kwargs={ + "outputs": "[C211,C212]", + "presets": "[100,200]", + "unit": "BadUnit", "pattern": "[[1,0],[0,1]]", "current_step": "DS186", "accumulator": "TD5", @@ -705,7 +726,7 @@ def test_two_row_rung_is_noop(self) -> None: [""] * 31, ] _hydrate_wire_continuations(rows) - assert rows[1][1] == "" # unchanged — last row, not eligible + assert rows[1][1] == "" # unchanged — last row, not eligible def test_existing_pipe_also_propagates(self) -> None: """| in the source row also propagates downward, not just T.""" @@ -740,3 +761,4 @@ def test_timer_autopad_hydrates_T_wire(self, tmp_path: Path) -> None: # T in col B (index 1) on row 0 should hydrate | on row 1 assert conds[0][1] == "T" assert conds[1][1] == "|" + From 852ff20971e6c8295c096ce0803893821b2439f2 Mon Sep 17 00:00:00 2001 From: GruppoVeneta Date: Fri, 10 Jul 2026 12:52:26 +0200 Subject: [PATCH 3/9] test(csv): cover time drum CSV round-trip units --- tests/csv/test_writer.py | 31 ++++++++++++++++++++++++++++--- 1 file changed, 28 insertions(+), 3 deletions(-) diff --git a/tests/csv/test_writer.py b/tests/csv/test_writer.py index 6cec404..3ad708c 100644 --- a/tests/csv/test_writer.py +++ b/tests/csv/test_writer.py @@ -1,4 +1,4 @@ -"""Tests for laddercodec.csv.writer — Rung → CSV round-trip.""" +"""Tests for laddercodec.csv.writer — Rung → CSV round-trip.""" from __future__ import annotations @@ -128,7 +128,7 @@ def _event_drum() -> Drum: # --------------------------------------------------------------------------- -# Round-trip: golden.bin → decode → CSV → read_csv → compare +# Round-trip: golden.bin → decode → CSV → read_csv → compare # --------------------------------------------------------------------------- @@ -173,7 +173,7 @@ def test_round_trip(self, golden_stem: str, tmp_path: Path) -> None: class TestDecodedRungToRows: def test_simple_contact_coil(self) -> None: - """Single row: contact → wire → coil.""" + """Single row: contact → wire → coil.""" rung = Rung( logical_rows=1, conditions=[[Contact(InstructionType.CONTACT_NO, "X001")] + ["-"] * 30], @@ -241,6 +241,30 @@ def test_timer_retained_emits_reset_pin(self) -> None: assert rows[1][0] == "" assert rows[1][32] == ".reset()" + @pytest.mark.parametrize("unit", ["Ts", "Tm", "Th"]) + def test_time_drum_known_unit_roundtrips_to_csv(self, unit: str) -> None: + drum = Drum( + drum_kind="time", + outputs=["C206", "C207"], + events_or_presets=["100", "200"], + pattern=[[1, 0], [0, 1]], + current_step="DS186", + completion_flag="C213", + accumulator="TD5", + unit=unit, + ) + rung = Rung( + logical_rows=4, + conditions=[_contact_row("C210"), *([_blank_conditions()] * 3)], + instructions=[drum, "", "", ""], + comment_rtf=None, + comment=None, + ) + + rows = decoded_rung_to_rows(rung) + + assert f"unit={unit}" in rows[0][32] + def test_nop(self) -> None: """NOP token serializes correctly.""" rung = Rung( @@ -929,3 +953,4 @@ def test_raises_on_missing_drum_jump_pin(self) -> None: with pytest.raises(WriterError, match=r"AF mismatch at row 1"): _validate_roundtrip(rung, rows) + From 336915fa6d8f542731c5785919cbdf06c91b1ab0 Mon Sep 17 00:00:00 2001 From: GruppoVeneta Date: Fri, 10 Jul 2026 12:54:47 +0200 Subject: [PATCH 4/9] fix: restore UTF-8 source content --- src/laddercodec/instructions/drum.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/src/laddercodec/instructions/drum.py b/src/laddercodec/instructions/drum.py index a04da45..65e9608 100644 --- a/src/laddercodec/instructions/drum.py +++ b/src/laddercodec/instructions/drum.py @@ -1,4 +1,4 @@ -"""Drum — event drum and time drum instructions (type marker 0x271B). +"""Drum — event drum and time drum instructions (type marker 0x271B). Binary class name: ``"Drum"``. Drum types: event_drum (event-driven), time_drum (time-driven). @@ -27,13 +27,13 @@ # Event drum: jog group (base, reset, jump, jog) _EVENT_FC_JOG = (8494, 8495, 8496, 8497) -# Time drum: unit → (base, reset) +# Time drum: unit → (base, reset) _TIME_FC: dict[str, tuple[int, int]] = { "Tms": (8455, 8456), "Td": (8484, 8485), } -# Time unit name → index string (same convention as Timer). +# Time unit name → index string (same convention as Timer). _DRUM_UNIT_TO_INDEX: dict[str, str] = { "Tms": "0", "Ts": "1", @@ -370,4 +370,3 @@ def parse_af_call(call: AfCall) -> Drum: pin_names=(".reset", ".jump", ".jog"), min_csv_rows=4, ) - From f209ac88d76dca375b52bff798777b4c0ba2678f Mon Sep 17 00:00:00 2001 From: GruppoVeneta Date: Fri, 10 Jul 2026 12:55:06 +0200 Subject: [PATCH 5/9] fix: restore UTF-8 source content From be60ebf7ef9a0aa08129ef70a91cf02e16196cfb Mon Sep 17 00:00:00 2001 From: GruppoVeneta Date: Fri, 10 Jul 2026 12:55:19 +0200 Subject: [PATCH 6/9] fix: restore UTF-8 source content --- tests/csv/test_converter.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tests/csv/test_converter.py b/tests/csv/test_converter.py index 081c9ae..5378852 100644 --- a/tests/csv/test_converter.py +++ b/tests/csv/test_converter.py @@ -726,7 +726,7 @@ def test_two_row_rung_is_noop(self) -> None: [""] * 31, ] _hydrate_wire_continuations(rows) - assert rows[1][1] == "" # unchanged — last row, not eligible + assert rows[1][1] == "" # unchanged — last row, not eligible def test_existing_pipe_also_propagates(self) -> None: """| in the source row also propagates downward, not just T.""" @@ -761,4 +761,3 @@ def test_timer_autopad_hydrates_T_wire(self, tmp_path: Path) -> None: # T in col B (index 1) on row 0 should hydrate | on row 1 assert conds[0][1] == "T" assert conds[1][1] == "|" - From 187b12ca207052b4d8aae24c03c80aaf480bb20e Mon Sep 17 00:00:00 2001 From: GruppoVeneta Date: Fri, 10 Jul 2026 12:55:21 +0200 Subject: [PATCH 7/9] fix: restore UTF-8 source content --- tests/csv/test_writer.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/tests/csv/test_writer.py b/tests/csv/test_writer.py index 3ad708c..83c9c08 100644 --- a/tests/csv/test_writer.py +++ b/tests/csv/test_writer.py @@ -1,4 +1,4 @@ -"""Tests for laddercodec.csv.writer — Rung → CSV round-trip.""" +"""Tests for laddercodec.csv.writer — Rung → CSV round-trip.""" from __future__ import annotations @@ -128,7 +128,7 @@ def _event_drum() -> Drum: # --------------------------------------------------------------------------- -# Round-trip: golden.bin → decode → CSV → read_csv → compare +# Round-trip: golden.bin → decode → CSV → read_csv → compare # --------------------------------------------------------------------------- @@ -173,7 +173,7 @@ def test_round_trip(self, golden_stem: str, tmp_path: Path) -> None: class TestDecodedRungToRows: def test_simple_contact_coil(self) -> None: - """Single row: contact → wire → coil.""" + """Single row: contact → wire → coil.""" rung = Rung( logical_rows=1, conditions=[[Contact(InstructionType.CONTACT_NO, "X001")] + ["-"] * 30], @@ -953,4 +953,3 @@ def test_raises_on_missing_drum_jump_pin(self) -> None: with pytest.raises(WriterError, match=r"AF mismatch at row 1"): _validate_roundtrip(rung, rows) - From 271f53990d8cef6871a87ae7a4dbb6b89bd018ea Mon Sep 17 00:00:00 2001 From: GruppoVeneta Date: Fri, 10 Jul 2026 14:00:52 +0200 Subject: [PATCH 8/9] fix(csv): preserve positional drum continuation rows --- src/laddercodec/csv/writer.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/laddercodec/csv/writer.py b/src/laddercodec/csv/writer.py index 568cc2a..66aa25d 100644 --- a/src/laddercodec/csv/writer.py +++ b/src/laddercodec/csv/writer.py @@ -280,11 +280,17 @@ def _emit_drum_block( f".jump({drum.jump_target})", ) else: - data_row_count = _emit_blank_continuation(rows, data_row_count, condition_rows[start + 2]) + # Drum continuation rows are positional. Omitting an empty jump row + # lets a later blank-AF row slide into the jump/jog slot on reparse. + data_row_count = _append_data_row( + rows, data_row_count, condition_rows[start + 2], "" + ) if drum.jog_enabled: data_row_count = _append_data_row(rows, data_row_count, condition_rows[start + 3], ".jog()") else: - data_row_count = _emit_blank_continuation(rows, data_row_count, condition_rows[start + 3]) + data_row_count = _append_data_row( + rows, data_row_count, condition_rows[start + 3], "" + ) return data_row_count, 4 From f6a3ff56a9ef7b3c77ff26701ce5bc15106bcc20 Mon Sep 17 00:00:00 2001 From: GruppoVeneta Date: Fri, 10 Jul 2026 14:01:08 +0200 Subject: [PATCH 9/9] test(csv): cover drum rows followed by ladder logic --- tests/csv/test_writer.py | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/tests/csv/test_writer.py b/tests/csv/test_writer.py index 83c9c08..866ce0b 100644 --- a/tests/csv/test_writer.py +++ b/tests/csv/test_writer.py @@ -23,6 +23,7 @@ ) from laddercodec.csv.writer import ( WriterError, + _rebuild_rung_from_rows, _validate_roundtrip, decoded_rung_to_rows, ) @@ -265,6 +266,37 @@ def test_time_drum_known_unit_roundtrips_to_csv(self, unit: str) -> None: assert f"unit={unit}" in rows[0][32] + def test_time_drum_preserves_blank_pin_slots_before_following_rows(self) -> None: + drum = Drum( + drum_kind="time", + outputs=["C1056", "C1057"], + events_or_presets=["10", "600"], + pattern=[[1, 0], [0, 1]], + current_step="DS6", + completion_flag="C109", + accumulator="TD4", + unit="Ts", + ) + rung = Rung( + logical_rows=5, + conditions=[ + _contact_row("C108"), + _contact_row("C109"), + _blank_conditions(), + _contact_row("C110"), + _contact_row("C111"), + ], + instructions=[drum, "", "", "", ""], + comment_rtf=None, + comment=None, + ) + + rows = decoded_rung_to_rows(rung) + rebuilt = _rebuild_rung_from_rows(rows) + + assert rebuilt.logical_rows == 5 + assert rebuilt.conditions == rung.conditions + def test_nop(self) -> None: """NOP token serializes correctly.""" rung = Rung(