Modbus-rtu/HostComputer/PythonTest/plsr_test.py

#!/usr/bin/env python3
"""PLSR Modbus RTU test client.

The ``spec`` map follows ``document/参数表-2026.xlsx``.  ``legacy`` keeps
compatibility with the first single-segment firmware image in this project.
"""

from __future__ import annotations

import argparse
import sys
import time
from dataclasses import dataclass, field
from typing import Optional, Sequence

try:
    import serial
    from serial.tools import list_ports
except ImportError:
    serial = None
    list_ports = None


SLAVE_ADDRESS = 1
DEFAULT_BAUD_RATE = 115200
MAP_SPEC = "spec"
MAP_LEGACY = "legacy"

# Requirement-map addresses.  These are holding-register offsets, not 4xxxx
# display addresses.  A 32-bit value uses high word first, low word second.
REG_CFG_PULSE_OUTPUT = 0x1000
REG_CFG_DIRECTION_OUTPUT = 0x1001
REG_CFG_WAIT_INPUT = 0x1002
REG_CFG_EXT_INPUT = 0x1003
REG_CFG_SEND_MODE = 0x1004
REG_CFG_DIRECTION_DELAY_MS = 0x1005
REG_CFG_DIRECTION_LOGIC = 0x1006
REG_CFG_ACCEL_MODE = 0x1007
REG_CFG_RUN_MODE = 0x1008
REG_CFG_SEGMENT_COUNT = 0x1009
REG_CFG_START_SEGMENT = 0x100A
REG_CFG_DEFAULT_SPEED = 0x100B
REG_CFG_START_SPEED = 0x100D
REG_CFG_END_SPEED = 0x1010
REG_CFG_ACCEL_TIME_MS = 0x1012
REG_CFG_DECEL_TIME_MS = 0x1013
REG_SEGMENT_BASE = 0x1100
REG_SEGMENT_STRIDE = 0x10
REG_MONITOR_TOTAL_PULSES = 0x2000
REG_MONITOR_FREQUENCY = 0x2002
REG_MONITOR_STATE = 0x2004
REG_MONITOR_SEGMENT = 0x2005
REG_MONITOR_ERROR = 0x2006
REG_MONITOR_STOP_REASON = 0x2007
REG_CONTROL = 0x3000

# Legacy zero-based map used by the current minimal firmware image.
REG_STATE = 0
REG_FLAGS = 1
REG_CURRENT_FREQUENCY = 3
REG_STOP_REASON = 5
REG_LAST_ERROR = 6
REG_TOTAL_PULSES_HIGH = 10
REG_TOTAL_PULSES_LOW = 11
REG_COMMAND = 100
REG_PULSES_HIGH = 102
REG_PULSES_LOW = 103
REG_FREQUENCY = 104
REG_DIRECTION = 105
REG_PULSE_OUTPUT = 106
REG_DIRECTION_OUTPUT = 107

COMMAND_START = 1
COMMAND_STOP = 2
COMMAND_RESET = 3
COMMAND_CLEAR_COUNT = 4
COMMAND_IMMEDIATE_STOP = 5
COMMAND_SAVE_CONFIG = 6
COMMAND_RESTART = 7

CONTROL_START = 1 << 0
CONTROL_STOP = 1 << 1
CONTROL_CLEAR_COUNT = 1 << 2
CONTROL_RESTART = 1 << 3
CONTROL_RESET = 1 << 4
CONTROL_IMMEDIATE_STOP = 1 << 5

STATE_NAMES = {
    0: "Idle",
    1: "Armed",
    2: "Busy",
    3: "Wait",
    4: "Decelerating",
    5: "Done",
    6: "Stopped",
    7: "Error",
}
ERROR_NAMES = {
    0: "OK",
    1: "Invalid argument",
    2: "Busy",
    3: "Resource",
    4: "State",
    5: "Wait timeout",
    6: "Path range",
    7: "Loop limit",
    8: "External stop",
    9: "Config CRC",
    10: "Communication",
    11: "Timer",
}
WAIT_CONDITION_NAMES = {
    0: "WAIT time",
    1: "WAIT signal",
    2: "ACT time",
    3: "EXT signal",
    4: "EXT signal or pulse done",
}

MANDATORY_TEST_ITEMS = (
    ("M01", "Y0..Y3 pulse output selection", "automatic register round-trip + oscilloscope"),
    ("M02", "Y12..Y15 direction output selection and logic", "automatic register round-trip + level observation"),
    ("M03", "positive/negative signed pulse count", "automatic motion/count + direction observation"),
    ("M04", "1..100000 Hz and online frequency update", "register boundary + oscilloscope/frequency counter"),
    ("M05", "complete/follow-up send mode", "state transition observation"),
    ("M06", "relative/absolute mode including equal-position no-op", "motion/count observation"),
    ("M07", "linear/S-curve/sine acceleration", "configuration round-trip + oscilloscope capture"),
    ("M08", "start/default/end speed and acceleration/deceleration time", "configuration round-trip + waveform timing"),
    ("M09", "1..10 segments and selectable start segment", "10-segment round-trip + execution sequence"),
    ("M10", "WAIT time and WAIT signal", "timing + manual X4/X5 trigger"),
    ("M11", "ACT time and EXT signal", "timing + manual X4/X5 trigger"),
    ("M12", "EXT signal or pulse completion", "manual EXT trigger and natural completion branches"),
    ("M13", "jump=0/default next/end and explicit jump", "current-segment monitor sequence"),
    ("M14", "cumulative signed count and communication clear", "automatic monitor/control check"),
    ("M15", "stop, state, current segment and error diagnostics", "automatic state/error monitor check"),
    ("M16", "power-loss save/load and power-up parameter restore", "manual power-cycle and readback"),
)


def mandatory_test_checklist() -> str:
    return "\n".join(f"{code}  {name}  [{method}]" for code, name, method in MANDATORY_TEST_ITEMS)


class PlsrError(RuntimeError):
    pass


class CommunicationError(PlsrError):
    pass


class ModbusException(PlsrError):
    def __init__(self, function: int, code: int) -> None:
        super().__init__(f"Modbus exception: function=0x{function:02X}, code=0x{code:02X}")
        self.function = function
        self.code = code


def crc16(data: bytes) -> int:
    value = 0xFFFF
    for byte in data:
        value ^= byte
        for _ in range(8):
            value = ((value >> 1) ^ 0xA001) if value & 1 else value >> 1
    return value


def with_crc(body: bytes) -> bytes:
    value = crc16(body)
    return body + bytes((value & 0xFF, value >> 8))


def read_holding_request(slave: int, start: int, count: int) -> bytes:
    return with_crc(bytes((slave, 0x03)) + start.to_bytes(2, "big") + count.to_bytes(2, "big"))


def write_register_request(slave: int, address: int, value: int) -> bytes:
    return with_crc(bytes((slave, 0x06)) + address.to_bytes(2, "big") + value.to_bytes(2, "big"))


def write_registers_request(slave: int, address: int, values: Sequence[int]) -> bytes:
    if not 1 <= len(values) <= 123:
        raise ValueError("multiple-register write count must be 1..123")
    payload = b"".join(int(v).to_bytes(2, "big") for v in values)
    body = bytes((slave, 0x10)) + address.to_bytes(2, "big") + len(values).to_bytes(2, "big")
    return with_crc(body + bytes((len(payload),)) + payload)


def _frame_length(buffer: bytearray, offset: int, expected_function: int) -> Optional[int]:
    if len(buffer) - offset < 2:
        return None
    function = buffer[offset + 1]
    if function == (expected_function | 0x80):
        return 5
    if function != expected_function:
        return -1
    if function in (0x01, 0x03):
        if len(buffer) - offset < 3:
            return None
        return 5 + buffer[offset + 2]
    if function in (0x05, 0x06, 0x0F, 0x10):
        return 8
    return -1


def extract_response(buffer: bytearray, slave: int, expected_function: int) -> Optional[bytes]:
    """Find a valid response while tolerating stale bytes and fragmentation."""
    for offset in range(max(0, len(buffer) - 260), len(buffer)):
        if buffer[offset] != slave:
            continue
        length = _frame_length(buffer, offset, expected_function)
        if length is None:
            continue
        if length < 0 or len(buffer) - offset < length:
            continue
        candidate = bytes(buffer[offset : offset + length])
        if int.from_bytes(candidate[-2:], "little") == crc16(candidate[:-2]):
            del buffer[: offset + length]
            return candidate
    if len(buffer) > 512:
        del buffer[:-260]
    return None


def encode_u32(value: int) -> tuple[int, int]:
    if not 0 <= value <= 0xFFFFFFFF:
        raise ValueError("unsigned 32-bit value must be 0..4294967295")
    return (value >> 16) & 0xFFFF, value & 0xFFFF


def encode_i32(value: int) -> tuple[int, int]:
    if not -0x80000000 <= value <= 0x7FFFFFFF:
        raise ValueError("signed 32-bit value must be -2147483648..2147483647")
    return encode_u32(value & 0xFFFFFFFF)


def decode_u32(high: int, low: int) -> int:
    return ((high & 0xFFFF) << 16) | (low & 0xFFFF)


def decode_i32(high: int, low: int) -> int:
    value = decode_u32(high, low)
    return value - 0x100000000 if value & 0x80000000 else value


def validate_frequency(value: int, *, allow_zero: bool = False) -> None:
    """Validate a frequency value used by the PLSR protocol."""
    minimum = 0 if allow_zero else 1
    if not minimum <= value <= 100_000:
        lower = "0" if allow_zero else "1"
        raise ValueError(f"frequency must be {lower}..100000 Hz")


@dataclass
class SegmentParameters:
    frequency_hz: int = 1000
    pulses: int = 0
    wait_condition: int = 0
    wait_time_ms: int = 0
    act_time_ms: int = 0
    jump_segment: int = 0

    def validate(self, index: int, total_segments: int = 10) -> None:
        validate_frequency(self.frequency_hz, allow_zero=True)
        if not -0x80000000 <= self.pulses <= 0x7FFFFFFF:
            raise ValueError(f"segment {index}: pulses must be signed 32-bit")
        if self.wait_condition not in WAIT_CONDITION_NAMES:
            raise ValueError(f"segment {index}: wait condition must be 0..4")
        if not 0 <= self.wait_time_ms <= 0xFFFF or not 0 <= self.act_time_ms <= 0xFFFF:
            raise ValueError(f"segment {index}: wait/act time must be 0..65535 ms")
        if not 0 <= self.jump_segment <= total_segments:
            raise ValueError(f"segment {index}: jump segment must be 0..{total_segments}")


@dataclass
class PlsrConfiguration:
    pulse_output: int = 0
    direction_output: int = 0
    wait_input: int = 0
    ext_input: int = 0
    send_mode: int = 0
    direction_delay_ms: int = 0
    direction_logic: int = 0
    accel_mode: int = 0
    run_mode: int = 0
    segment_count: int = 1
    start_segment: int = 1
    default_speed_hz: int = 1000
    start_speed_hz: int = 1
    end_speed_hz: int = 1
    accel_time_ms: int = 0
    decel_time_ms: int = 0
    segments: list[SegmentParameters] = field(default_factory=lambda: [SegmentParameters() for _ in range(10)])

    def validate(self) -> None:
        if not 0 <= self.pulse_output <= 3:
            raise ValueError("pulse output must be Y0..Y3 (0..3)")
        if not 0 <= self.direction_output <= 3:
            raise ValueError("direction output must be Y12..Y15 (0..3)")
        if not 0 <= self.wait_input <= 1 or not 0 <= self.ext_input <= 1:
            raise ValueError("WAIT/EXT input must be X4 or X5 (0..1)")
        if self.send_mode not in (0, 1):
            raise ValueError("send mode must be 0 (complete) or 1 (follow-up)")
        if self.direction_logic not in (0, 1):
            raise ValueError("direction logic must be 0 (positive) or 1 (negative)")
        if self.accel_mode not in (0, 1, 2):
            raise ValueError("acceleration mode must be 0, 1 or 2")
        if self.run_mode not in (0, 1):
            raise ValueError("run mode must be 0 (relative) or 1 (absolute)")
        if not 0 <= self.direction_delay_ms <= 0xFFFF:
            raise ValueError("direction delay must be 0..65535 ms")
        if not 1 <= self.segment_count <= 10:
            raise ValueError("segment count must be 1..10")
        if not 1 <= self.start_segment <= self.segment_count:
            raise ValueError("start segment must be within 1..segment count")
        validate_frequency(self.default_speed_hz)
        validate_frequency(self.start_speed_hz, allow_zero=True)
        validate_frequency(self.end_speed_hz, allow_zero=True)
        if not 0 <= self.accel_time_ms <= 0xFFFF or not 0 <= self.decel_time_ms <= 0xFFFF:
            raise ValueError("acceleration/deceleration time must be 0..65535 ms")
        if len(self.segments) < 10:
            raise ValueError("configuration must contain 10 segment slots")
        for index in range(self.segment_count):
            self.segments[index].validate(index + 1, self.segment_count)


@dataclass(frozen=True)
class Snapshot:
    # First eight fields retain the legacy GUI/test API.
    state: int
    flags: int
    current_frequency: int
    stop_reason: int
    last_error: int
    total_pulses: int
    pulse_output: int
    direction_output: int
    current_segment: int = 0
    protocol: str = MAP_LEGACY

    @property
    def state_name(self) -> str:
        return STATE_NAMES.get(self.state, f"Unknown({self.state})")

    @property
    def error_name(self) -> str:
        return ERROR_NAMES.get(self.last_error, f"Unknown({self.last_error})")


class ModbusRtuClient:
    def __init__(self, port: str, baud_rate: int = DEFAULT_BAUD_RATE, slave: int = SLAVE_ADDRESS,
                 timeout: float = 0.6, retries: int = 2, verbose: bool = False,
                 protocol: str = MAP_LEGACY) -> None:
        if protocol not in (MAP_SPEC, MAP_LEGACY):
            raise ValueError("protocol must be 'spec' or 'legacy'")
        self.port, self.baud_rate, self.slave = port, baud_rate, slave
        self.timeout, self.retries, self.verbose, self.protocol = timeout, retries, verbose, protocol
        self._serial = None

    def __enter__(self) -> "ModbusRtuClient":
        self.open()
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        self.close()

    def open(self) -> None:
        if serial is None:
            raise PlsrError("pyserial is not installed; run: python -m pip install -r requirements.txt")
        self._serial = serial.Serial(port=self.port, baudrate=self.baud_rate, bytesize=serial.EIGHTBITS,
                                     parity=serial.PARITY_EVEN, stopbits=serial.STOPBITS_ONE,
                                     timeout=0.05, write_timeout=self.timeout)
        self._serial.reset_input_buffer()
        self._serial.reset_output_buffer()

    def close(self) -> None:
        if self._serial is not None:
            self._serial.close()
            self._serial = None

    def _log_frame(self, prefix: str, frame: bytes) -> None:
        if self.verbose:
            print(f"{time.strftime('%H:%M:%S')} {prefix} {frame.hex(' ').upper()}")

    def _receive(self, expected_function: int) -> bytes:
        if self._serial is None:
            raise CommunicationError("serial port is not open")
        deadline, buffer = time.monotonic() + self.timeout, bytearray()
        while time.monotonic() < deadline:
            waiting = self._serial.in_waiting
            chunk = self._serial.read(waiting if waiting > 0 else 1)
            if chunk:
                buffer.extend(chunk)
                response = extract_response(buffer, self.slave, expected_function)
                if response is not None:
                    self._log_frame("RX", response)
                    if response[1] == (expected_function | 0x80):
                        raise ModbusException(expected_function, response[2])
                    return response
        tail = bytes(buffer[-32:]).hex(" ").upper() if buffer else "<none>"
        raise CommunicationError(f"response timeout; received tail: {tail}")

    def transaction(self, request: bytes, expected_function: int) -> bytes:
        if self._serial is None:
            raise CommunicationError("serial port is not open")
        last_error: Optional[Exception] = None
        for attempt in range(self.retries + 1):
            if attempt:
                time.sleep(0.1)
            try:
                if self._serial is None:
                    self.open()
                self._serial.reset_input_buffer()
                self._log_frame("TX", request)
                written = self._serial.write(request)
                self._serial.flush()
                if written != len(request):
                    last_error = CommunicationError(
                        f"short serial write: {written}/{len(request)}"
                    )
                    continue
                return self._receive(expected_function)
            except ModbusException:
                raise
            except CommunicationError as exc:
                last_error = exc
                if (attempt > 0) and (attempt < self.retries):
                    try:
                        self.close()
                        self.open()
                    except (OSError, PlsrError) as reopen_error:
                        last_error = reopen_error
            except OSError as exc:
                last_error = exc
                if attempt < self.retries:
                    try:
                        self.close()
                        self.open()
                    except (OSError, PlsrError) as reopen_error:
                        last_error = reopen_error
        raise CommunicationError(f"transaction failed after {self.retries + 1} attempts: {last_error}")

    def read_holding(self, start: int, count: int) -> list[int]:
        if not 0 <= start <= 0xFFFF or not 1 <= count <= 125:
            raise ValueError("invalid holding-register range")
        response = self.transaction(read_holding_request(self.slave, start, count), 0x03)
        if response[2] != count * 2:
            raise CommunicationError(f"unexpected byte count: {response[2]}, expected {count * 2}")
        payload = response[3:-2]
        return [int.from_bytes(payload[i:i + 2], "big") for i in range(0, len(payload), 2)]

    def write_register(self, address: int, value: int) -> None:
        if not 0 <= address <= 0xFFFF or not 0 <= value <= 0xFFFF:
            raise ValueError("register address/value out of range")
        request = write_register_request(self.slave, address, value)
        if self.transaction(request, 0x06) != request:
            raise CommunicationError("write response does not echo the request")

    def write_registers(self, address: int, values: Sequence[int]) -> None:
        request = write_registers_request(self.slave, address, values)
        response = self.transaction(request, 0x10)
        if response[:6] != request[:6] or int.from_bytes(response[-2:], "little") != crc16(response[:-2]):
            raise CommunicationError("multiple-write response does not echo address/count")

    def read_u32(self, address: int) -> int:
        values = self.read_holding(address, 2)
        return decode_u32(values[0], values[1])

    def read_i32(self, address: int) -> int:
        values = self.read_holding(address, 2)
        return decode_i32(values[0], values[1])

    def write_u32(self, address: int, value: int) -> None:
        self.write_registers(address, encode_u32(value))

    def write_i32(self, address: int, value: int) -> None:
        self.write_registers(address, encode_i32(value))

    def set_running_frequency(self, frequency: int) -> Snapshot:
        """修改当前运行段频率，不重新启动脉冲任务。"""
        validate_frequency(frequency)
        if self.protocol == MAP_LEGACY:
            self.write_register(REG_FREQUENCY, frequency)
            return self.read_snapshot()

        snapshot = self.read_snapshot()
        if snapshot.state not in (1, 2, 3, 4):
            raise PlsrError("motion is not running")
        if not 1 <= snapshot.current_segment <= 10:
            raise PlsrError("current segment is invalid")
        address = REG_SEGMENT_BASE + (snapshot.current_segment - 1) * REG_SEGMENT_STRIDE
        self.write_u32(address, frequency)
        return self.read_snapshot()

    def command(self, value: int) -> None:
        if self.protocol == MAP_LEGACY:
            self.write_register(REG_COMMAND, value)
            return
        if value == COMMAND_START:
            self.write_register(REG_CONTROL, CONTROL_START)
        elif value == COMMAND_STOP:
            self.write_register(REG_CONTROL, CONTROL_STOP)
        elif value == COMMAND_IMMEDIATE_STOP:
            self.write_register(REG_CONTROL, CONTROL_IMMEDIATE_STOP)
        elif value == COMMAND_CLEAR_COUNT:
            self.write_register(REG_CONTROL, CONTROL_CLEAR_COUNT)
        elif value == COMMAND_RESET:
            self.write_register(REG_CONTROL, CONTROL_RESET)
        elif value == COMMAND_RESTART:
            self.write_register(REG_CONTROL, CONTROL_RESTART)
        elif value == COMMAND_SAVE_CONFIG:
            raise PlsrError("the requirement map has no save bit; configuration persistence is firmware-managed")
        else:
            raise ValueError(f"unsupported command: {value}")

    def read_snapshot(self) -> Snapshot:
        if self.protocol == MAP_LEGACY:
            status = self.read_holding(0, 32)
            outputs = self.read_holding(REG_PULSE_OUTPUT, 2)
            return Snapshot(status[REG_STATE], status[REG_FLAGS], status[REG_CURRENT_FREQUENCY],
                            status[REG_STOP_REASON], status[REG_LAST_ERROR],
                            decode_u32(status[REG_TOTAL_PULSES_HIGH], status[REG_TOTAL_PULSES_LOW]),
                            outputs[0], outputs[1])
        monitor = self.read_holding(REG_MONITOR_TOTAL_PULSES, 8)
        config = self.read_holding(REG_CFG_PULSE_OUTPUT, 2)
        state = monitor[4]
        flags = 1 if state in (1, 2, 3, 4) else (2 if state == 5 else (4 if state == 7 else 0))
        return Snapshot(state, flags, decode_u32(monitor[2], monitor[3]), monitor[7], monitor[6],
                        decode_i32(monitor[0], monitor[1]), config[0], config[1], monitor[5], MAP_SPEC)

    def read_configuration(self) -> PlsrConfiguration:
        if self.protocol != MAP_SPEC:
            raise PlsrError("configuration object is only available in spec protocol")
        values = self.read_holding(REG_CFG_PULSE_OUTPUT, REG_CFG_DECEL_TIME_MS - REG_CFG_PULSE_OUTPUT + 1)
        cfg = PlsrConfiguration(
            pulse_output=values[0], direction_output=values[1], wait_input=values[2], ext_input=values[3],
            send_mode=values[4], direction_delay_ms=values[5], direction_logic=values[6], accel_mode=values[7],
            run_mode=values[8], segment_count=values[9], start_segment=values[10],
            default_speed_hz=decode_u32(values[11], values[12]), start_speed_hz=decode_u32(values[13], values[14]),
            end_speed_hz=decode_u32(values[16], values[17]), accel_time_ms=values[18], decel_time_ms=values[19],
        )
        cfg.segments = [self.read_segment(index + 1) for index in range(10)]
        return cfg

    def write_configuration(self, config: PlsrConfiguration, save: bool = False) -> None:
        if self.protocol != MAP_SPEC:
            raise PlsrError("configuration object is only available in spec protocol")
        config.validate()
        values = [config.pulse_output, config.direction_output, config.wait_input, config.ext_input,
                  config.send_mode, config.direction_delay_ms, config.direction_logic, config.accel_mode,
                  config.run_mode, config.segment_count, config.start_segment]
        values.extend(encode_u32(config.default_speed_hz))
        values.extend(encode_u32(config.start_speed_hz))
        values.append(0)  # 0x100F reserved
        values.extend(encode_u32(config.end_speed_hz))
        values.extend((config.accel_time_ms, config.decel_time_ms))
        self.write_registers(REG_CFG_PULSE_OUTPUT, values)
        for index in range(1, config.segment_count + 1):
            self.write_segment(index, config.segments[index - 1])
        if save:
            # There is no save register in the revised parameter table. Keep
            # the argument for GUI compatibility and make the limitation clear.
            raise PlsrError("configuration written to RAM; add the firmware save command before requesting Flash save")

    def read_segment(self, index: int) -> SegmentParameters:
        if self.protocol != MAP_SPEC or not 1 <= index <= 10:
            raise ValueError("segment index must be 1..10")
        base = REG_SEGMENT_BASE + (index - 1) * REG_SEGMENT_STRIDE
        values = self.read_holding(base, 8)
        return SegmentParameters(decode_u32(values[0], values[1]), decode_i32(values[2], values[3]),
                                 values[4], values[5], values[6], values[7])

    def write_segment(self, index: int, segment: SegmentParameters) -> None:
        if self.protocol != MAP_SPEC or not 1 <= index <= 10:
            raise ValueError("segment index must be 1..10")
        segment.validate(index)
        base = REG_SEGMENT_BASE + (index - 1) * REG_SEGMENT_STRIDE
        self.write_registers(base, [*encode_u32(segment.frequency_hz), *encode_i32(segment.pulses),
                                    segment.wait_condition, segment.wait_time_ms, segment.act_time_ms,
                                    segment.jump_segment])

    def configure_outputs(self, pulse_output: int, direction_output: int, save: bool = False) -> Snapshot:
        if self.protocol == MAP_LEGACY:
            if not 0 <= pulse_output <= 7 or not 0 <= direction_output <= 7 or pulse_output == direction_output:
                raise ValueError("legacy output points must be distinct Q0..Q7")
            snapshot = self.read_snapshot()
            if snapshot.state == 2:
                raise PlsrError("configuration is blocked while motion is Busy")
            self.write_register(REG_PULSE_OUTPUT, pulse_output)
            self.write_register(REG_DIRECTION_OUTPUT, direction_output)
            if save:
                self.command(COMMAND_SAVE_CONFIG)
            return self.read_snapshot()
        if not 0 <= pulse_output <= 3 or not 0 <= direction_output <= 3:
            raise ValueError("outputs must be Y0..Y3 and Y12..Y15")
        if self.read_snapshot().state in (1, 2, 3, 4):
            raise PlsrError("configuration is blocked while motion is active")
        self.write_register(REG_CFG_PULSE_OUTPUT, pulse_output)
        self.write_register(REG_CFG_DIRECTION_OUTPUT, direction_output)
        return self.read_snapshot()

    def start_motion(self, pulses: int, frequency: int, direction: int = 0,
                     restart: bool = False) -> None:
        if self.protocol == MAP_LEGACY:
            if not 1 <= pulses <= 0xFFFFFFFF:
                raise ValueError("legacy pulses must be 1..4294967295")
            if not 1 <= frequency <= 1000:
                raise ValueError("legacy firmware test module supports 1..1000 Hz")
            if direction not in (0, 1):
                raise ValueError("direction must be 0 or 1")
            self.write_register(REG_PULSES_HIGH, (pulses >> 16) & 0xFFFF)
            self.write_register(REG_PULSES_LOW, pulses & 0xFFFF)
            self.write_register(REG_FREQUENCY, frequency)
            self.write_register(REG_DIRECTION, direction)
            self.command(COMMAND_RESTART if restart else COMMAND_START)
            return
        validate_frequency(frequency)
        if pulses == 0 or not -0x80000000 <= pulses <= 0x7FFFFFFF:
            raise ValueError("spec pulse count must be non-zero signed 32-bit")
        if direction not in (0, 1):
            raise ValueError("direction must be 0 or 1")
        if direction == 1 and pulses > 0:
            pulses = -pulses
        cfg = self.read_configuration()
        cfg.segments[cfg.start_segment - 1] = SegmentParameters(frequency, pulses)
        cfg.segment_count = max(cfg.segment_count, cfg.start_segment)
        cfg.validate()
        self.write_segment(cfg.start_segment, cfg.segments[cfg.start_segment - 1])
        self.command(COMMAND_RESTART if restart else COMMAND_START)


def print_snapshot(snapshot: Snapshot) -> None:
    print(f"state={snapshot.state_name:<14} segment={snapshot.current_segment:<2} "
          f"frequency={snapshot.current_frequency}Hz total_pulses={snapshot.total_pulses} "
          f"error={snapshot.last_error}:{snapshot.error_name} outputs={snapshot.pulse_output}/{snapshot.direction_output}")


def wait_for_terminal(client: ModbusRtuClient, timeout: float, interval: float = 0.25) -> Snapshot:
    deadline, previous = time.monotonic() + timeout, None
    while time.monotonic() < deadline:
        snapshot = client.read_snapshot()
        current = (snapshot.state, snapshot.total_pulses, snapshot.last_error)
        if current != previous:
            print_snapshot(snapshot)
            previous = current
        if snapshot.state in (5, 6, 7):
            return snapshot
        time.sleep(interval)
    try:
        client.command(COMMAND_STOP)
    except PlsrError:
        pass
    raise PlsrError("motion timeout; stop command was attempted")


def require_safe_confirmation(args: argparse.Namespace) -> None:
    if not args.confirm_safe:
        raise PlsrError("motion output is blocked; disconnect the mechanism and add --confirm-safe")


def offline_self_test() -> None:
    assert crc16(bytes.fromhex("01 03 00 00 00 0A")) == 0xCDC5
    assert read_holding_request(1, 0, 32).hex(" ") == "01 03 00 00 00 20 44 12"
    assert encode_i32(-1) == (0xFFFF, 0xFFFF)
    assert decode_i32(0x8000, 0) == -2147483648
    request = write_registers_request(1, REG_SEGMENT_BASE, [100, 0, 0xFFFF, 0xFFFF, 0, 0, 0, 0])
    assert request[1] == 0x10 and request[6] == 16
    response = with_crc(bytes.fromhex("01 03 04 00 05 00 06"))
    noisy = bytearray(bytes.fromhex("40 00 00") + response)
    assert extract_response(noisy, 1, 0x03) == response
    print("PASS CRC16, signed double-word, FC16 and response resynchronization")


def add_connection_options(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--port", help="serial port, for example COM9")
    parser.add_argument("--baud", type=int, default=DEFAULT_BAUD_RATE)
    parser.add_argument("--slave", type=int, default=SLAVE_ADDRESS)
    parser.add_argument("--timeout", type=float, default=0.6)
    parser.add_argument("--retries", type=int, default=2)
    parser.add_argument("--map", choices=(MAP_SPEC, MAP_LEGACY), default=MAP_SPEC,
                        help="spec follows 参数表-2026.xlsx; legacy follows the first firmware image")
    parser.add_argument("-v", "--verbose", action="store_true")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="PLSR firmware Modbus RTU test tool")
    add_connection_options(parser)
    subs = parser.add_subparsers(dest="command_name", required=True)
    subs.add_parser("ports")
    subs.add_parser("self-test")
    subs.add_parser("mandatory-list")
    subs.add_parser("status")
    monitor = subs.add_parser("monitor"); monitor.add_argument("--interval", type=float, default=0.5)
    run = subs.add_parser("run"); run.add_argument("--pulses", type=int, default=10); run.add_argument("--frequency", type=int, default=2)
    run.add_argument("--direction", choices=("forward", "reverse"), default="forward"); run.add_argument("--wait-timeout", type=float, default=30.0); run.add_argument("--no-wait", action="store_true"); run.add_argument("--confirm-safe", action="store_true")
    subs.add_parser("stop"); subs.add_parser("immediate-stop"); subs.add_parser("reset"); subs.add_parser("clear")
    config = subs.add_parser("configure"); config.add_argument("--pulse-output", type=int, required=True); config.add_argument("--direction-output", type=int, required=True); config.add_argument("--save", action="store_true")
    auto = subs.add_parser("auto-test"); auto.add_argument("--pulses", type=int, default=6); auto.add_argument("--frequency", type=int, default=2); auto.add_argument("--confirm-safe", action="store_true"); auto.add_argument("--wait-timeout", type=float, default=30.0)
    return parser


def run_connected_command(args: argparse.Namespace) -> None:
    if not args.port:
        raise PlsrError("--port is required for this command")
    with ModbusRtuClient(args.port, args.baud, args.slave, args.timeout, args.retries, args.verbose, args.map) as client:
        if args.command_name == "status": print_snapshot(client.read_snapshot())
        elif args.command_name == "monitor":
            while True: print_snapshot(client.read_snapshot()); time.sleep(max(0.1, args.interval))
        elif args.command_name == "run":
            require_safe_confirmation(args); client.start_motion(args.pulses, args.frequency, int(args.direction == "reverse"))
            if not args.no_wait and wait_for_terminal(client, args.wait_timeout).state != 5: raise PlsrError("motion did not finish normally")
        elif args.command_name == "configure": print_snapshot(client.configure_outputs(args.pulse_output, args.direction_output, args.save))
        elif args.command_name == "stop": client.command(COMMAND_STOP); print_snapshot(client.read_snapshot())
        elif args.command_name == "immediate-stop": client.command(COMMAND_IMMEDIATE_STOP); print_snapshot(client.read_snapshot())
        elif args.command_name == "reset": client.command(COMMAND_RESET); print_snapshot(client.read_snapshot())
        elif args.command_name == "clear": client.command(COMMAND_CLEAR_COUNT); print_snapshot(client.read_snapshot())
        elif args.command_name == "auto-test":
            require_safe_confirmation(args); client.command(COMMAND_CLEAR_COUNT); client.start_motion(args.pulses, args.frequency, 0); first = wait_for_terminal(client, args.wait_timeout)
            if first.state != 5: raise PlsrError("forward test failed")
            client.start_motion(args.pulses, args.frequency, 1); second = wait_for_terminal(client, args.wait_timeout)
            if second.state != 5: raise PlsrError("reverse test failed")
            print("PASS forward/reverse mandatory smoke test")


def main(argv: Optional[Sequence[str]] = None) -> int:
    args = build_parser().parse_args(argv)
    try:
        if args.command_name == "self-test": offline_self_test()
        elif args.command_name == "mandatory-list": print(mandatory_test_checklist())
        elif args.command_name == "ports":
            if list_ports is None: raise PlsrError("pyserial is not installed")
            for item in sorted(list_ports.comports(), key=lambda p: p.device): print(f"{item.device:<8} {item.description}")
        else: run_connected_command(args)
        return 0
    except KeyboardInterrupt: return 130
    except (OSError, ValueError, PlsrError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr); return 1


if __name__ == "__main__":
    raise SystemExit(main())