scanengine-3/tests/test_helios_laser.py

"""HeliosLaser: replies that span more than one line.

Regression: every status-register query answers with the value *and* a
"Bit 15..0: ..." decode line.  The driver read one line per query and threw
the rest away with reset_input_buffer(), which at 9600 baud cannot drop
bytes that are still on the wire — so from the first LER read onward every
reply was one line behind, and the panel showed a register as
"Bit 15..0: 0000 0000 0000 0010" with the reads around it timing out.
"""
import pytest

from hardware.helios_laser import HeliosLaser, PulseMode


# What the controller actually sends back, per Section 6 of the operator's
# manual and the LER/LCE/CCE tables in Section 8.
REPLIES = {
    "LDO": ["LDO = 1"],
    "LDF": ["LDF = 20000 ns"],
    "LDS": ["LDS = 1500 mA"],
    "LDG": ["LDG = 14"],
    "LRE": ["LRE = 0"],
    "LTA": ["LTA = 25400 m°C"],
    "LTT": ["LTT = 31200 m°C"],
    "EOA": ["EOA = 40100 m°C"],
    "CSR": ["CSR = 1234567"],
    "HSR": ["HSR = 7654321"],
    # The registers are the multi-line ones.
    "LER": ["LER = 0", "Bit 15..0: 0000 0000 0000 0000"],
    "LCE": ["LCE = 2", "Bit 15..0: 0000 0000 0000 0010"],
    "CCE": ["CCE = 0", "Bit 15..0: 0000 0000 0000 0000"],
}


class FakePort:
    """Serial stand-in that answers like the Helios controller.

    ``reset_input_buffer`` is deliberately a no-op: the rest of a reply is
    still in flight when the driver has read its first line, so a flush
    cannot remove it.  The driver has to stay in step by reading what it
    asked for, not by discarding what it happens to find.
    """

    def __init__(self, replies=None, timeout=1.0, discard_writes=False):
        self.replies = dict(REPLIES) if replies is None else replies
        self.timeout = timeout
        self.is_open = True
        # "Commands or set values can be discarded by the controller
        # unintentionally" (manual, Section 6) — the case a verified write
        # exists to catch.
        self.discard_writes = discard_writes
        self.written: list[str] = []
        self._buf = bytearray()

    def _store(self, mnemonic: str, value: str):
        """Keep a written value, so a later query reads back what was set."""
        if self.discard_writes:
            return
        previous = self.replies.get(mnemonic, [f"{mnemonic} = 0"])[0]
        unit = previous.split()[3:]     # "LDS = 1500 mA" -> ["mA"]
        self.replies[mnemonic] = [" ".join([f"{mnemonic} =", value, *unit])]

    # ── the bits of pyserial.Serial the driver uses ──────────────────────────

    def write(self, data: bytes) -> int:
        text = data.decode("ascii").strip()
        self.written.append(text)
        fields = text.split()
        if len(fields) > 1:
            self._store(fields[0].upper(), fields[1])
        for line in self.replies.get(fields[0].upper() if fields else "", []):
            self._buf += line.encode("utf-8") + b"\r\n"
        return len(data)

    @property
    def in_waiting(self) -> int:
        return len(self._buf)

    def read(self, size: int = 1) -> bytes:
        chunk = bytes(self._buf[:size])
        del self._buf[:size]
        return chunk

    def read_until(self, expected: bytes = b"\n", size=None) -> bytes:
        # No terminator in the buffer models the read timing out: pyserial
        # returns whatever it has, which is b"" when nothing is pending.
        cut = self._buf.find(expected)
        cut = len(self._buf) if cut < 0 else cut + len(expected)
        chunk = bytes(self._buf[:cut])
        del self._buf[:cut]
        return chunk

    def reset_input_buffer(self):
        """No-op — see the class docstring."""

    def close(self):
        self.is_open = False


@pytest.fixture
def laser():
    """A connected driver on a fake port, with the idle wait taken out.

    TRAILING_QUIET_S covers the ~30 ms a decode line spends on the wire at
    9600 baud; the fake answers instantly, so waiting for it only slows the
    suite down.  Zeroing it also keeps the tests honest: they pass because
    the driver reads the whole reply, not because it waited long enough.
    """
    drv = HeliosLaser(timeout=1.0)
    drv.serial = FakePort()
    drv.is_connected = True
    drv.TRAILING_QUIET_S = 0.0
    drv.SET_SETTLE_S = 0.0
    return drv


@pytest.fixture
def stubborn_laser():
    """A controller that answers every query but keeps its own set values."""
    drv = HeliosLaser(timeout=1.0)
    drv.serial = FakePort(discard_writes=True)
    drv.is_connected = True
    drv.TRAILING_QUIET_S = 0.0
    drv.SET_SETTLE_S = 0.0
    return drv


def test_status_registers_are_read_in_step(laser):
    """The regression: each register gets its own value, not the previous
    register's decode line."""
    assert laser.get_status_registers() == (0, 2, 0)


def test_reads_after_a_register_are_not_a_line_behind(laser):
    """A whole status poll, in the order HeliosWorker._poll_once issues it."""
    assert laser.get_status_registers() == (0, 2, 0)
    assert laser.is_laser_enabled() is True
    assert laser.get_current_ma() == 1500
    assert laser.get_diode_temp_c() == pytest.approx(25.4)
    assert laser.get_power_stage_temp_c() == pytest.approx(31.2)
    assert laser.get_qswitch_temp_c() == pytest.approx(40.1)


def test_decode_line_is_consumed_not_left_behind(laser):
    laser.get_status_registers()
    assert laser.serial.in_waiting == 0


def test_unit_suffix_is_stripped(laser):
    assert laser.get_frequency_hz() == 50000       # "LDF = 20000 ns"
    assert laser.get_current_ma() == 1500          # "LDS = 1500 mA"


def test_a_reply_from_an_earlier_command_is_skipped(laser):
    """An answer already in the buffer when the query goes out belongs to
    whoever asked for it, and must not be returned as this query's value."""
    laser.serial._buf += b"LER = 8\r\nBit 15..0: 0000 0000 0000 1000\r\n"
    assert laser.get_current_ma() == 1500


def test_reply_without_a_mnemonic_is_taken_as_the_value(laser):
    """The serial numbers come back as a bare string on some firmware."""
    laser.serial.replies = {"CSR": ["A1B2C3D4"]}
    assert laser.get_controller_serial() == "A1B2C3D4"


def test_silent_device_reports_a_timeout(laser):
    laser.serial.replies = {}
    assert laser.get_current_ma() is None
    assert laser._query("LTA") is None


def test_set_commands_clear_their_acknowledgement(laser):
    """A setter that leaves the controller's echo in the buffer desynchronises
    the next query just as a decode line does."""
    assert laser.set_laser_enable(True) is True
    assert laser.serial.in_waiting == 0
    assert laser.set_pulse_mode(PulseMode.CONTINUOUS_PULSING) is True
    assert laser.get_current_ma() == 1500


def test_raw_command_returns_every_line(laser):
    """The diagnostics console is where a multi-line reply should be visible."""
    assert laser.send_raw_command("LCE") == (
        "LCE = 2\nBit 15..0: 0000 0000 0000 0010"
    )


# ── Set values the controller may discard ────────────────────────────────────
#
# Section 6 of the manual: "Commands or set values can be discarded by the
# controller unintentionally.  It is recommended to query the set value
# after the command is entered to confirm the actual value."  A write that
# reports success without reading back leaves the panel showing a setpoint
# the laser never took, until the next poll replaces it with the old value.

def test_a_set_current_is_read_back(laser):
    assert laser.set_current_ma(900) is True
    assert laser.get_current_ma() == 900
    assert "LDS 900" in laser.serial.written


def test_a_discarded_set_current_is_retried_then_reported(stubborn_laser):
    assert stubborn_laser.set_current_ma(900) is False
    # Retried, not given up on after one write.
    assert stubborn_laser.serial.written.count("LDS 900") == \
        stubborn_laser.SET_RETRIES
    # And the controller's own value is what it still holds.
    assert stubborn_laser.get_current_ma() == 1500


def test_a_set_that_takes_on_a_retry_succeeds(laser):
    """One dropped write, then the controller accepts — still a success."""
    real_write = laser.serial.write
    state = {"drops": 1}

    def flaky(data: bytes) -> int:
        if data.decode("ascii").strip().startswith("LDS ") and state["drops"]:
            state["drops"] -= 1
            laser.serial.written.append(data.decode("ascii").strip())
            return len(data)        # swallowed: nothing stored, no reply
        return real_write(data)

    laser.serial.write = flaky
    assert laser.set_current_ma(900) is True
    assert laser.get_current_ma() == 900


def test_a_set_frequency_is_read_back(laser):
    assert laser.set_frequency_hz(25000) is True    # 40000 ns
    assert laser.get_frequency_hz() == 25000
    assert "LDF 40000" in laser.serial.written


def test_an_out_of_range_current_is_not_sent(laser):
    assert laser.set_current_ma(9000) is False
    assert laser.serial.written == []


def test_an_unlabelled_number_is_not_taken_as_a_register_value(laser):
    """A bare number answers nothing in particular.

    The controller's status registers read 32 when bit 5 is set (LER "Over
    voltage laser diode", LCE "Door switch open"), and a diode current of
    32 mA is a perfectly ordinary-looking value — so a stray "32" must not
    be allowed to pass for the answer to LDS.
    """
    laser.serial.replies = {"LDS": ["32"]}
    assert laser.get_current_ma() is None


def test_a_serial_number_still_comes_back_bare(laser):
    """The one reply that legitimately names nothing."""
    laser.serial.replies = {"CSR": ["A1B2C3D4"], "HSR": ["7654321"]}
    assert laser.get_controller_serial() == "A1B2C3D4"
    assert laser.get_head_serial() == "7654321"