Modbus-rtu/HostComputer/plsr_modbus_bit_input_test.py

#!/usr/bin/env python3
"""Validate the production Modbus M -> PLSR WAIT bit-data path.

X is intentionally read-only through function 0x02.  A real X/EXT/hard-limit
test is only possible after the board-specific GPIO-to-X mapping is supplied.
"""

from __future__ import annotations

import argparse
import struct
import time

import serial

from plsr_modbus_control_test import (
    CALL_COMMIT,
    CALL_START,
    CONTROL_BASE,
    CONTROL_WINDOW_WORDS,
    RESULT_OK,
    RESULT_QUEUED,
    S0_BASE,
    S1_BASE,
    check_result,
    read_axis_status,
    send_call,
)
from plsr_modbus_frequency_test import RtuClient, choose_port, signed_dword_words


STATE_WAIT = 5
STATE_COMPLETED = 7
M_TEST_POINT = 123


def write_coil(client: RtuClient, address: int, value: bool) -> None:
    encoded = 0xFF00 if value else 0x0000
    pdu = bytes((0x05,)) + struct.pack(">HH", address, encoded)
    response = client.exchange(pdu, 8)
    expected = bytes((client.slave, 0x05)) + struct.pack(">HH", address, encoded)
    if response[:6] != expected:
        raise RuntimeError(f"FC05 回显错误: {response.hex(' ')}")


def read_bits(client: RtuClient, function: int, address: int, count: int) -> list[int]:
    byte_count = (count + 7) // 8
    pdu = bytes((function,)) + struct.pack(">HH", address, count)
    response = client.exchange(pdu, 5 + byte_count)
    if response[1] != function or response[2] != byte_count:
        raise RuntimeError(f"FC{function:02X} 响应格式错误: {response.hex(' ')}")
    return [
        (response[3 + (index // 8)] >> (index % 8)) & 1
        for index in range(count)
    ]


def wait_for_state(client: RtuClient, expected: int, timeout: float) -> dict[str, int]:
    deadline = time.monotonic() + timeout
    latest: dict[str, int] | None = None
    while time.monotonic() < deadline:
        latest = read_axis_status(client)
        if latest["state"] == expected:
            return latest
    raise RuntimeError(f"等待状态 {expected} 超时，最后状态: {latest}")


def prepare_wait_job(client: RtuClient) -> None:
    words = [0] * 20
    words[0:2] = signed_dword_words(1)       # one segment
    words[10:12] = signed_dword_words(1000) # 1000 Hz
    words[12:14] = signed_dword_words(500)  # +500 pulses
    words[14] = (2 << 8) | 5                # WAIT_SIGNAL, source M
    words[15:17] = signed_dword_words(M_TEST_POINT)
    words[17] = 0                           # constant fall-through jump
    words[18:20] = signed_dword_words(0)
    client.write_multiple(S0_BASE, words)
    client.write_multiple(S1_BASE, [0, 0, 0, 0])


def main() -> int:
    parser = argparse.ArgumentParser(
        description="PLSR 真实 Modbus M 位源、WAIT 与 FC02 X 只读视图测试"
    )
    parser.add_argument("--port", default="COM5")
    parser.add_argument("--baud", type=int, default=9600)
    parser.add_argument("--slave", type=int, default=1)
    args = parser.parse_args()

    with serial.Serial(
        port=choose_port(args.port),
        baudrate=args.baud,
        bytesize=serial.EIGHTBITS,
        parity=serial.PARITY_EVEN,
        stopbits=serial.STOPBITS_ONE,
        timeout=1.0,
        write_timeout=1.0,
    ) as uart:
        client = RtuClient(uart, args.slave)
        header = client.read_holding(CONTROL_BASE, 8)
        if header[3] != CONTROL_WINDOW_WORDS:
            raise RuntimeError(
                f"控制窗口版本不匹配: firmware={header[3]}, script={CONTROL_WINDOW_WORDS}"
            )

        # FC02 must exist and must not alias writable M coils.
        x_before = read_bits(client, 0x02, M_TEST_POINT, 1)[0]
        write_coil(client, M_TEST_POINT, False)
        if read_bits(client, 0x01, M_TEST_POINT, 1) != [0]:
            raise RuntimeError("FC05 写 M=0 后 FC01 回读不一致")
        if read_bits(client, 0x02, M_TEST_POINT, 1)[0] != x_before:
            raise RuntimeError("X 与 M 发生别名：写 M 意外改变了 FC02 X")

        prepare_wait_job(client)
        sequence = int(time.time()) & 0x7FFFFFFF
        response = send_call(client, sequence, CALL_COMMIT)
        check_result(response, RESULT_OK, "COMMIT")
        response = send_call(client, sequence + 1, CALL_START)
        # START queues the immutable snapshot; PlsrTask applies it
        # asynchronously, so QUEUED is the successful protocol reply.
        check_result(response, RESULT_QUEUED, "START")

        waiting = wait_for_state(client, STATE_WAIT, 5.0)
        if waiting["task_pulses"] != 500:
            raise RuntimeError(f"进入 WAIT 时任务脉冲不是 500: {waiting}")
        print("M123=0：500 脉冲完成后稳定进入 WAIT")

        write_coil(client, M_TEST_POINT, True)
        if read_bits(client, 0x01, M_TEST_POINT, 1) != [1]:
            raise RuntimeError("FC05 写 M=1 后 FC01 回读不一致")
        completed = wait_for_state(client, STATE_COMPLETED, 3.0)
        if completed["task_pulses"] != 500:
            raise RuntimeError(f"WAIT 释放后任务计数异常: {completed}")
        print("M123 0->1：PLSR WAIT 已释放并正常 COMPLETED")
        print(f"FC02 X123 只读值={x_before}；写 M 不会改变 X")
        print("PASS：Modbus FC05/FC01 -> M image -> PLSR readBit/WAIT 生产链通过")
        print("待硬件映射后再测：实际 X 输入、EXT 上升沿及正/负硬限位。")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (RuntimeError, serial.SerialException) as error:
        print(f"测试失败：{error}")
        raise SystemExit(1)