Modbus-rtu/HostComputer/plsr_modbus_performance_test.py

#!/usr/bin/env python3
"""PLSR P16 Modbus 性能统计自动测试（对应规格：PLSR_MODBUS_PERFORMANCE_TEST.md）。

本脚本在四轴 100kHz 输出与 Modbus 持续轮询的并发压力下，测量并判定
规格中定义的各项性能指标，运行完成后自动读取 P16 DWT 统计并做预算检查：

- 版本号一致性（任务要求的“32/64 位版本号一致性检查”）：
  D1205~D1206 的 32 位 CPU 计时频率非零、D1207 性能统计版本为 V7、
  控制窗口协议版本 0x0100；P16 主统计块（32 位周期数与 16 位饱和字段）
  连续两次读取必须完全一致。
- 单条命令响应时间：0x03/0x10 完整往返耗时，判定预算 = 波特率折算的
  线上传输时间（8E1，每字符 11 位）+ 10ms 处理余量。
- 连续状态读取吞吐：四轴状态块（48 字×4）持续轮询速率，要求不低于
  波特率理论上限的 50%。
- 重复序号幂等：运行期重发完全相同序号的命令，固件必须只回放既有应答、
  不得重复执行（沿用 P13 控制测试的 PAUSE 重复序号用例）。
- 并发读写压力模拟：四轴 100kHz 期间持续读取轴状态，并穿插重写相同的
  S0/S1 数据（幂等写入）与读取 P16 统计块，验证并发下计数精确。
- P16 自动预算：PlsrProcess 自身 < 1ms、TIM6 控制 ISR < 0.1ms、输出/计数
  ISR < 10us，四项均须实际执行（数值不得为 0）；墙钟响应单独显示，
  超过 1ms 只给抢占提示、不判失败。

与规格/现有脚本的差异说明：
- 规格“测试方法”一节引用 P14 脚本；本脚本独立实现相同的四轴 100kHz +
  持续轮询场景（规格要求），并额外测量主机侧性能指标，满足规格
  “自动预算”与“记录到测试报告”的要求。
- 规格要求“无毛刺、无残余输出”，该两项只能由逻辑分析仪验收，本脚本
  按规格保留为人工核对提示，与 plsr_modbus_counter_stress_test.py 口径一致。
- P16 主统计块读取比 P14 脚本更严格：必须连续两次读取一致（规格要求
  统计快照一致性），三次尝试仍不一致即判定失败。
"""

from __future__ import annotations

import argparse
import time

import serial

from plsr_modbus_frequency_test import RtuClient, choose_port, signed_dword_words


CONTROL_BASE = 1200
CONTROL_WINDOW_WORDS = 338
CALL_REQUEST = CONTROL_BASE + 8
CALL_RESPONSE = CONTROL_BASE + 24
COMMAND_REQUEST = CONTROL_BASE + 40
COMMAND_RESPONSE = CONTROL_BASE + 48
AXIS_STATUS_BASE = CONTROL_BASE + 64
AXIS_STATUS_WORDS = 48
PERFORMANCE_BASE = CONTROL_BASE + 56          # D1256：P16 主统计块
STAGE_PERFORMANCE_BASE = CONTROL_BASE + 256   # D1456：P16 分阶段统计块
AB_GATE_PERFORMANCE_BASE = CONTROL_BASE + 268 # D1468：AB末周期快速门控
PERFORMANCE_VERSION = 7
STAGE_NAMES = (
    "脉冲合并/保护",
    "关键事件",
    "命令队列",
    "普通事件/方向提交",
    "HAL/路径/Profile",
    "HSD检查点",
)
STAGE_PERFORMANCE_WORDS = len(STAGE_NAMES) * 2

S0_BASES = (1600, 1800, 2000, 2200)
S1_BASES = (1700, 1900, 2100, 2300)
TEST_FREQUENCY_HZ = 100_000
TEST_PULSES = 200_000

RESULT_OK = 0
RESULT_QUEUED = 1
STATE_ACCEL = 2
STATE_RUN = 3
STATE_PAUSED = 6
STATE_COMPLETED = 7

CALL_COMMIT = 1
CALL_START = 2
CMD_PAUSE = 3
CMD_RESUME = 4
CMD_SET_POSITION = 5

BITS_PER_CHARACTER = 11        # Modbus RTU 8E1：每字符 11 位
PROCESSING_BUDGET_MS = 10.0    # 单条命令响应预算中的处理余量（主机+从站开销）
THROUGHPUT_MIN_RATIO = 0.5     # 实测吞吐不低于波特率理论上限的比例
LATENCY_SAMPLES = 20           # 每种命令的响应时间采样次数
RUN_DEADLINE_SECONDS = 15.0    # 四轴运行完成的等待上限

METRICS: list[tuple[str, bool, str]] = []


def put_u32(words: list[int], offset: int, value: int) -> None:
    words[offset : offset + 2] = signed_dword_words(value)


def put_u64(words: list[int], offset: int, value: int) -> None:
    raw = value & 0xFFFFFFFFFFFFFFFF
    words[offset : offset + 4] = [
        (raw >> shift) & 0xFFFF for shift in (0, 16, 32, 48)
    ]


def get_u32(words: list[int], offset: int) -> int:
    return words[offset] | (words[offset + 1] << 16)


def get_u64(words: list[int], offset: int, signed: bool = False) -> int:
    raw = sum(words[offset + index] << (16 * index) for index in range(4))
    if signed and raw & (1 << 63):
        return raw - (1 << 64)
    return raw


def wait_response(
    client: RtuClient, address: int, words: int, sequence: int, timeout: float = 2.0
) -> list[int]:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        response = client.read_holding(address, words)
        if get_u32(response, 0) == sequence:
            return response
    raise RuntimeError(f"等待序号 {sequence} 的应答超时")


def send_command(
    client: RtuClient, sequence: int, axis: int, opcode: int, argument: int = 0
) -> list[int]:
    request = [0] * 8
    put_u32(request, 0, sequence)
    request[2] = opcode
    request[3] = axis
    put_u64(request, 4, argument)
    client.write_multiple(COMMAND_REQUEST, request)
    return wait_response(client, COMMAND_RESPONSE, 8, sequence)


def send_call(
    client: RtuClient,
    sequence: int,
    axis: int,
    operation: int,
    s2_set: int = 1,
) -> list[int]:
    request = [0] * 16
    put_u32(request, 0, sequence)
    request[2] = 0  # S0 device D
    put_u32(request, 3, S0_BASES[axis])
    request[5] = 0  # S1 device D
    put_u32(request, 6, S1_BASES[axis])
    request[8] = 0  # S2 constant
    put_u32(request, 10, s2_set)
    request[12] = axis
    request[13] = 0  # PULSE/DIR
    request[14] = operation
    client.write_multiple(CALL_REQUEST, request)
    return wait_response(client, CALL_RESPONSE, 12, sequence)


def check_result(response: list[int], offset: int, expected: int, label: str) -> None:
    if response[offset] != expected:
        raise RuntimeError(
            f"{label} 返回 {response[offset]}，期望 {expected}；应答={response}"
        )


def read_axis_status(client: RtuClient, axis: int) -> dict[str, int]:
    address = AXIS_STATUS_BASE + axis * AXIS_STATUS_WORDS
    words = client.read_holding(address, AXIS_STATUS_WORDS)
    generation_begin = get_u32(words, 0)
    generation_end = get_u32(words, 46)
    if generation_begin != generation_end or generation_begin & 1:
        raise RuntimeError(
            f"轴{axis}状态快照不一致：begin={generation_begin}, end={generation_end}"
        )
    return {
        "state": words[2],
        "flags": get_u32(words, 3),
        "error": words[6],
        "stop_reason": words[7],
        "last_result": words[8],
        "last_sequence": get_u32(words, 10),
        "logical_position": get_u64(words, 16, signed=True),
        "task_pulses": get_u64(words, 20, signed=True),
        "physical_pulses": get_u64(words, 28),
        "counter_mode": words[37],
        "current_frequency": get_u32(words, 38),
        "target_frequency": get_u32(words, 40),
    }


def wait_status(
    client: RtuClient,
    axis: int,
    states: set[int],
    sequence: int | None = None,
    timeout: float = 5.0,
) -> dict[str, int]:
    deadline = time.monotonic() + timeout
    latest: dict[str, int] | None = None
    while time.monotonic() < deadline:
        latest = read_axis_status(client, axis)
        sequence_ok = sequence is None or latest["last_sequence"] == sequence
        if latest["state"] in states and sequence_ok:
            return latest
    raise RuntimeError(f"等待轴{axis}状态 {sorted(states)} 超时，最后状态：{latest}")


def wait_running_output(
    client: RtuClient, axis: int, sequence: int, timeout: float = 5.0
) -> dict[str, int]:
    """等待真实脉冲恢复，不能只依据 ACCEL/RUN 状态标签。"""
    deadline = time.monotonic() + timeout
    latest: dict[str, int] | None = None
    while time.monotonic() < deadline:
        latest = read_axis_status(client, axis)
        pulse_active = (latest["flags"] & (1 << 1)) != 0
        if (
            latest["last_sequence"] == sequence
            and latest["state"] in {STATE_ACCEL, STATE_RUN}
            and pulse_active
            and latest["current_frequency"] > 0
        ):
            return latest
    raise RuntimeError(f"等待轴{axis}实际脉冲恢复超时，最后状态：{latest}")


def print_process_checkpoint(client: RtuClient, label: str) -> None:
    words = client.read_holding(PERFORMANCE_BASE, 4)
    stage_words = client.read_holding(
        STAGE_PERFORMANCE_BASE, STAGE_PERFORMANCE_WORDS
    )
    stages = ", ".join(
        f"{name}={get_u32(stage_words, index * 2)}"
        for index, name in enumerate(STAGE_NAMES)
    )
    print(
        f"P16阶段[{label}]：自身最大={get_u32(words, 0)} cycles，"
        f"响应最大={get_u32(words, 2)} cycles"
    )
    print(f"  分段最大：{stages}")


def read_performance_block(client: RtuClient) -> list[int] | None:
    """读取 P16 主统计块；连续两次读取一致才算有效（快照一致性）。"""
    for _ in range(3):
        first = client.read_holding(PERFORMANCE_BASE, 8)
        second = client.read_holding(PERFORMANCE_BASE, 8)
        if first == second:
            return first
    return None


def wire_time_ms(baud: int, request_bytes: int, response_bytes: int) -> float:
    """RTU 8E1 线上传输时间：每字符 11 位。"""
    return (request_bytes + response_bytes) * BITS_PER_CHARACTER * 1000.0 / baud


def throughput_theory(baud: int) -> float:
    """四轴状态轮询（4×48字，请求 8B + 响应 101B）的波特率理论速率（轮/s）。"""
    bytes_per_poll = 4 * (8 + 101)
    return baud / (bytes_per_poll * BITS_PER_CHARACTER)


def sample_round_trips(
    client: RtuClient,
    baud: int,
    actions: list[tuple[str, int, int, object]],
    samples: int,
) -> dict[str, tuple[float, float, float, float]]:
    """对每种命令采样往返耗时，返回 {名称: (最小, 中位, 最大, 预算) ms}。"""
    results = {}
    for name, request_bytes, response_bytes, action in actions:
        collected = []
        for _ in range(samples):
            started = time.perf_counter()
            action()
            collected.append((time.perf_counter() - started) * 1000.0)
        ordered = sorted(collected)
        results[name] = (
            ordered[0],
            ordered[len(ordered) // 2],
            ordered[-1],
            wire_time_ms(baud, request_bytes, response_bytes) + PROCESSING_BUDGET_MS,
        )
    return results


def build_s0_job() -> list[int]:
    s0 = [0] * 20
    put_u32(s0, 0, 1)
    put_u32(s0, 10, TEST_FREQUENCY_HZ)
    put_u32(s0, 12, TEST_PULSES)
    return s0


def prepare_jobs(client: RtuClient) -> None:
    for axis in range(4):
        client.write_multiple(S0_BASES[axis], build_s0_job())
        client.write_multiple(S1_BASES[axis], [0] * 4)


def record_metric(name: str, passed: bool, detail: str) -> None:
    """记录一项指标并实时打印 通过/失败 + 实测值。"""
    METRICS.append((name, passed, detail))
    print(f"  [{'通过' if passed else '失败'}] {name}：{detail}")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="PLSR P16 Modbus 性能统计自动测试"
        "（PLSR_MODBUS_PERFORMANCE_TEST.md）"
    )
    parser.add_argument("--port", help="串口，例如 COM5；只有一个串口时可省略")
    parser.add_argument("--baud", type=int, default=9600)
    parser.add_argument("--slave", type=int, default=1)
    parser.add_argument(
        "--polls",
        type=int,
        default=10,
        help="静态吞吐测试的完整四轴轮询轮数（默认 10）",
    )
    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[:5] != [
            0x504C,
            0x5352,
            0x0100,
            CONTROL_WINDOW_WORDS,
            0x0007,
        ]:
            raise RuntimeError(
                f"P16 控制窗口未就绪：{header}；请烧录当前固件并复位"
            )
        print("P16 控制窗口就绪：D1200~D1537，协议 V1.0")

        # ---- 指标1：版本号一致性（32位CPU时钟 + 16位性能版本 + 统计块快照） ----
        core_clock_hz = get_u32(header, 5)
        performance_version = header[7]
        baseline_block = read_performance_block(client)
        version_ok = (
            core_clock_hz != 0
            and performance_version == PERFORMANCE_VERSION
            and baseline_block is not None
        )
        consistency_text = (
            "一致" if baseline_block is not None else "连续三次读取不一致"
        )
        record_metric(
            "版本号一致性",
            version_ok,
            f"协议版本={header[2]:#06x}，32位CPU计时频率(D1205~D1206)="
            f"{core_clock_hz}Hz，性能统计版本(D1207)=V{performance_version}，"
            f"P16统计块两次读取{consistency_text}",
        )
        if not version_ok:
            raise RuntimeError("版本号一致性检查失败，后续指标失去判定基准")
        print_process_checkpoint(client, "复位后基线")
        print("P16 已就绪：四轴 PULSE/DIR，100kHz，200000脉冲/轴")

        # ---- 指标2：单条命令响应时间 ----
        s0_job = build_s0_job()
        actions = [
            (
                "0x03读取控制头(8字)",
                8,
                21,
                lambda: client.read_holding(CONTROL_BASE, 8),
            ),
            (
                "0x10写入S0(20字)",
                49,
                8,
                lambda: client.write_multiple(S0_BASES[0], s0_job),
            ),
            (
                "0x10写入S1(4字)",
                17,
                8,
                lambda: client.write_multiple(S1_BASES[0], [0] * 4),
            ),
        ]
        latency = sample_round_trips(client, args.baud, actions, LATENCY_SAMPLES)
        latency_ok = True
        latency_parts = []
        for name, (minimum, median, maximum, budget) in latency.items():
            latency_ok = latency_ok and median <= budget
            latency_parts.append(
                f"{name} 中位{median:.1f}ms"
                f"（最小{minimum:.1f}/最大{maximum:.1f}，预算{budget:.1f}ms）"
            )
        record_metric("单条命令响应时间", latency_ok, "；".join(latency_parts))

        # ---- 指标3：连续状态读取吞吐（静态轮询，无运行压力） ----
        static_started = time.monotonic()
        for _ in range(args.polls):
            for axis in range(4):
                read_axis_status(client, axis)
        static_elapsed = time.monotonic() - static_started
        static_polls = args.polls * 4
        static_rate = static_polls / static_elapsed
        theory = throughput_theory(args.baud)
        ratio = static_rate / theory
        record_metric(
            "连续状态读取吞吐",
            ratio >= THROUGHPUT_MIN_RATIO,
            f"静态四轴轮询 {static_polls} 次耗时 {static_elapsed:.3f}s，"
            f"速率 {static_rate:.2f} 轮/s，波特率理论上限 {theory:.2f} 轮/s"
            f"（占比 {ratio:.0%}）",
        )

        prepare_jobs(client)
        sequence = 100
        for axis in range(4):
            response = send_command(
                client, sequence, axis, CMD_SET_POSITION, argument=0
            )
            check_result(response, 4, RESULT_QUEUED, f"轴{axis} SET_POSITION")
            sequence += 1
        print("四轴位置已清零，S0/S1 已用 0x10 原子写入")
        print_process_checkpoint(client, "位置清零")

        for axis in range(4):
            response = send_call(client, sequence, axis, CALL_COMMIT)
            check_result(response, 3, RESULT_OK, f"轴{axis} COMMIT")
            if response[11] != 1:
                raise RuntimeError(f"轴{axis} COMMIT 未建立有效快照")
            sequence += 1
        print("四轴 COMMIT 校验通过")
        print_process_checkpoint(client, "COMMIT")

        physical_baseline = [
            read_axis_status(client, axis)["physical_pulses"]
            for axis in range(4)
        ]

        started = time.monotonic()
        for axis in range(4):
            response = send_call(client, sequence, axis, CALL_START)
            check_result(response, 3, RESULT_QUEUED, f"轴{axis} START")
            sequence += 1
        print("四轴 START 已排队；持续读取状态以施加 Modbus/任务并发压力")
        print_process_checkpoint(client, "START")

        running_status = [read_axis_status(client, axis) for axis in range(4)]
        for axis, status in enumerate(running_status):
            if status["state"] not in {STATE_ACCEL, STATE_RUN}:
                raise RuntimeError(f"轴{axis} 未进入运行态：{status}")
            expected_mode = 1 if axis < 2 else 0
            if status["counter_mode"] != expected_mode:
                raise RuntimeError(
                    f"轴{axis}计数模式={status['counter_mode']}，期望={expected_mode}"
                )
            if status["current_frequency"] != TEST_FREQUENCY_HZ:
                raise RuntimeError(f"轴{axis}频率不正确：{status}")
        print("运行期计数租约正确：Q0/Q1=硬件，Q2/Q3=软件回退")
        print_process_checkpoint(client, "进入运行态")

        # ---- 指标4：重复序号幂等（轴0，运行期命令处理） ----
        pause_sequence = sequence
        response = send_command(client, pause_sequence, 0, CMD_PAUSE)
        check_result(response, 4, RESULT_QUEUED, f"轴0 PAUSE#{pause_sequence}")
        status = wait_status(client, 0, {STATE_PAUSED}, sequence=pause_sequence)
        # 重发完全相同的请求：必须只回放既有应答、不得重复执行
        replay = send_command(client, pause_sequence, 0, CMD_PAUSE)
        status_again = read_axis_status(client, 0)
        idem_ok = (
            replay == response
            and status_again["state"] == STATE_PAUSED
            and status_again["last_sequence"] == pause_sequence
        )
        record_metric(
            "重复序号幂等",
            idem_ok,
            f"重发 PAUSE#{pause_sequence}：应答与首次完全一致={replay == response}，"
            f"状态保持暂停，last_sequence={status_again['last_sequence']}"
            f"（期望 {pause_sequence}）",
        )
        sequence += 1
        response = send_command(client, sequence, 0, CMD_RESUME)
        check_result(response, 4, RESULT_QUEUED, f"轴0 RESUME#{sequence}")
        status = wait_running_output(client, 0, sequence=sequence)
        resumed_pulses = status["task_pulses"]
        time.sleep(0.5)
        status = read_axis_status(client, 0)
        if status["task_pulses"] <= resumed_pulses:
            raise RuntimeError(f"RESUME#{sequence} 后轴0脉冲计数未增长：{status}")
        print(
            f"轴0 PAUSE/RESUME 完成：恢复输出，当前 {status['current_frequency']}Hz"
        )
        print_process_checkpoint(client, "暂停/恢复")

        # ---- 指标5：并发读写压力模拟（运行期持续轮询 + 幂等写入） ----
        loop_started = time.monotonic()
        deadline = loop_started + RUN_DEADLINE_SECONDS
        polls = 0
        writes = 0
        final_status: list[dict[str, int]] = running_status
        while time.monotonic() < deadline:
            final_status = [read_axis_status(client, axis) for axis in range(4)]
            polls += 4
            # 写压力：重写与 prepare_jobs 完全相同的 S0/S1（幂等，
            # 不影响 COMMIT 后已建立的运行快照）
            client.write_multiple(S0_BASES[0], s0_job)
            client.write_multiple(S1_BASES[0], [0] * 4)
            # 并发下验证 P16 统计块可正常读取
            client.read_holding(PERFORMANCE_BASE, 8)
            writes += 2
            if all(item["state"] == STATE_COMPLETED for item in final_status):
                break
        else:
            raise RuntimeError(f"等待四轴完成超时，最后状态：{final_status}")
        run_elapsed = time.monotonic() - loop_started

        # 并发压力下四轴计数精确性核对
        count_errors: list[str] = []
        for axis, status in enumerate(final_status):
            expected = {
                "logical_position": TEST_PULSES,
                "task_pulses": TEST_PULSES,
                "error": 0,
                "last_result": RESULT_OK,
            }
            bad = {
                key: (status[key], value)
                for key, value in expected.items()
                if status[key] != value
            }
            physical_delta = status["physical_pulses"] - physical_baseline[axis]
            if physical_delta != TEST_PULSES:
                bad["physical_pulses_delta"] = (physical_delta, TEST_PULSES)
            if bad:
                count_errors.append(f"轴{axis}: {bad}")
        record_metric(
            "并发读写压力",
            not count_errors,
            f"四轴100kHz运行期间读取状态 {polls} 次、穿插幂等写入 {writes} 次，"
            f"轮询吞吐 {polls / run_elapsed:.2f} 次/s；四轴计数均精确为 "
            f"{TEST_PULSES}、error=0"
            + (f"；异常：{'；'.join(count_errors)}" if count_errors else ""),
        )

        elapsed = time.monotonic() - started
        print_process_checkpoint(client, "运行完成")

        # ---- 指标6：P16 统计块快照一致性（运行后，含基线单调性） ----
        block = read_performance_block(client)
        if block is None:
            record_metric(
                "P16统计块快照一致性", False, "三次读取未得到连续一致结果"
            )
        else:
            regressions = [
                f"D{PERFORMANCE_BASE + offset}: {baseline_block[offset]}→{block[offset]}"
                for offset in range(len(block))
                if block[offset] < baseline_block[offset]
            ]
            detail = "连续两次读取完全一致（32位周期数与16位饱和字段）"
            if regressions:
                detail += "；但相对复位后基线发生回退：" + "、".join(regressions)
            record_metric("P16统计块快照一致性", not regressions, detail)

        # ---- 指标7~10：P16 自动预算 ----
        stage_performance = client.read_holding(
            STAGE_PERFORMANCE_BASE, STAGE_PERFORMANCE_WORDS
        )
        ab_gate_cycles = get_u32(
            client.read_holding(AB_GATE_PERFORMANCE_BASE, 2), 0
        )
        cycle_values: dict[str, int] = {}
        if block is not None:
            cycle_values = {
                "PlsrProcess自身": get_u32(block, 0),
                "PlsrProcess响应": get_u32(block, 2),
                "TIM6控制ISR": get_u32(block, 4),
                "输出定时器ISR": block[6],
                "TIM9/12计数ISR": block[7],
            }
            print("P16 DWT最坏执行时间：")
            for name, cycles in cycle_values.items():
                microseconds = cycles * 1_000_000.0 / core_clock_hz
                print(f"  {name:<16} {cycles:8d} cycles  {microseconds:8.3f}us")
        print("P16 PlsrProcess分段最大执行时间（各段独立峰值，来自不同轮次不能相加）：")
        for index, name in enumerate(STAGE_NAMES):
            cycles = get_u32(stage_performance, index * 2)
            microseconds = cycles * 1_000_000.0 / core_clock_hz
            print(f"  {name:<18} {cycles:8d} cycles  {microseconds:8.3f}us")
        print(
            "  AB末周期快速门控       "
            f"{ab_gate_cycles:8d} cycles  "
            f"{ab_gate_cycles * 1_000_000.0 / core_clock_hz:8.3f}us  "
            "（本PULSE/DIR用例未执行时允许为0）"
        )

        budgets = {
            "PlsrProcess自身": (core_clock_hz // 1_000, "1ms"),
            "TIM6控制ISR": (core_clock_hz // 10_000, "0.1ms"),
            "输出定时器ISR": (core_clock_hz // TEST_FREQUENCY_HZ, "10us"),
            "TIM9/12计数ISR": (core_clock_hz // TEST_FREQUENCY_HZ, "10us"),
        }
        for name, (budget, budget_text) in budgets.items():
            if block is None:
                record_metric(f"P16 {name}预算", False, "统计块不可用，无法判定")
                continue
            cycles = cycle_values[name]
            microseconds = cycles * 1_000_000.0 / core_clock_hz
            passed = cycles != 0 and cycles < budget
            detail = (
                f"实测 {cycles} cycles = {microseconds:.3f}us，"
                f"预算 < {budget} cycles（{budget_text}）"
            )
            if cycles == 0:
                detail += "；数值为 0，路径未被实际执行"
            record_metric(f"P16 {name}预算", passed, detail)

        # 墙钟响应与阶段峰值超过 1ms 时仅给提示（规格：不与自身 CPU 混算）
        if block is not None:
            response_cycles = cycle_values["PlsrProcess响应"]
            if response_cycles >= core_clock_hz // 1_000:
                print(
                    "提示：PlsrProcess墙钟响应超过1ms，但自身CPU执行时间达标；"
                    "差值来自高优先级PLSR定时器中断抢占。"
                )
            for index, name in enumerate(STAGE_NAMES):
                cycles = get_u32(stage_performance, index * 2)
                if cycles >= core_clock_hz // 1_000:
                    print(
                        f"提示：阶段“{name}”峰值超过1ms，需定位峰值路径；"
                        "各段最大值来自不同轮次，不能直接相加。"
                    )

        # ---- 汇总 ----
        print("\n性能测试指标汇总：")
        passed_count = sum(1 for _, passed, _ in METRICS if passed)
        for name, passed, detail in METRICS:
            print(f"  [{'通过' if passed else '失败'}] {name}：{detail}")
        print(f"最终判定：{passed_count}/{len(METRICS)} 项通过")
        failed = [name for name, passed, _ in METRICS if not passed]
        if failed:
            raise RuntimeError("未通过指标：" + "、".join(failed))
        print(
            f"全部 PASS：四轴均为 {TEST_PULSES} 脉冲，耗时 {elapsed:.3f}s，"
            f"运行期状态读取 {polls} 次"
        )
        print("请再核对逻辑分析仪：Q0~Q3 各200000个上升沿、100kHz、无窄脉冲。")
    return 0


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