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 diff --git a/src/laddercodec/instructions/drum.py b/src/laddercodec/instructions/drum.py index f4d1cb7..65e9608 100644 --- a/src/laddercodec/instructions/drum.py +++ b/src/laddercodec/instructions/drum.py @@ -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( diff --git a/tests/csv/test_converter.py b/tests/csv/test_converter.py index 74a7ccc..5378852 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", diff --git a/tests/csv/test_writer.py b/tests/csv/test_writer.py index 6cec404..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, ) @@ -241,6 +242,61 @@ 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_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(