Modbus-rtu/HostComputer/PLSR_Python_Tester/plsr_protocol.py

"""PLSR 参数表-2026 的 Modbus RTU 地址、枚举和编解码工具。"""

from enum import IntEnum
from typing import Dict, List, Sequence, Tuple

DEFAULT_SLAVE_ADDRESS = 1
DEFAULT_BAUD_RATE = 115200
MIN_FREQUENCY_HZ = 1
MAX_FREQUENCY_HZ = 100000
MAX_RAMP_TIME_MS = 0xFFFF


class Register(IntEnum):
    PULSE_POINT = 0x1000
    DIRECTION_POINT = 0x1001
    WAIT_SIGNAL = 0x1002
    EXT_SIGNAL = 0x1003
    SEND_MODE = 0x1004
    DIRECTION_DELAY_MS = 0x1005
    DIRECTION_LOGIC = 0x1006
    PROFILE_TYPE = 0x1007
    RUN_MODE = 0x1008
    TOTAL_SEGMENTS = 0x1009
    START_SEGMENT = 0x100A
    TARGET_FREQ_LO = 0x100B
    TARGET_FREQ_HI = 0x100C
    START_FREQ_LO = 0x100D
    START_FREQ_HI = 0x100E
    OUTPUT_MODE = 0x100F
    END_FREQ_LO = 0x1010
    END_FREQ_HI = 0x1011
    ACCEL_TIME = 0x1012
    DECEL_TIME = 0x1013
    MAX_FREQ_LO = 0x1014
    MAX_FREQ_HI = 0x1015
    # 瞬时运行参数，不属于公共参数，也不保存到Flash。
    RUNTIME_FREQ_LO = 0x1016
    RUNTIME_FREQ_HI = 0x1017

    SEGMENT_TABLE_BASE = 0x1100

    CUMULATIVE_POSITION_LO = 0x2000
    CUMULATIVE_POSITION_HI = 0x2001
    CURRENT_FREQ_LO = 0x2002
    CURRENT_FREQ_HI = 0x2003
    STATE = 0x2004
    CURRENT_SEGMENT = 0x2005
    ERROR = 0x2006
    CURRENT_PULSES_LO = 0x2007
    CURRENT_PULSES_HI = 0x2008
    TOTAL_PULSES_LO = 0x2009
    TOTAL_PULSES_HI = 0x200A
    DIAG_STATUS = 0x200B
    DIAG_PROFILE_TYPE = 0x200C
    DIAG_PHASE = 0x200D
    DIAG_EXPECTED_PULSES_LO = 0x200E
    DIAG_EXPECTED_PULSES_HI = 0x200F
    DIAG_OBSERVED_PULSES_LO = 0x2010
    DIAG_OBSERVED_PULSES_HI = 0x2011
    DIAG_EXPECTED_FREQ_LO = 0x2012
    DIAG_EXPECTED_FREQ_HI = 0x2013
    DIAG_ACTUAL_FREQ_LO = 0x2014
    DIAG_ACTUAL_FREQ_HI = 0x2015
    DIAG_MAX_FREQ_ERROR_LO = 0x2016
    DIAG_MAX_FREQ_ERROR_HI = 0x2017
    DIAG_COUNT_ERRORS = 0x2018
    DIAG_FREQ_ERRORS = 0x2019
    DIAG_PROFILE_ERRORS = 0x201A
    CONTROL = 0x3000

    # 旧界面源码的兼容别名；实际通讯流程只使用上面的参数表地址。
    COMMAND = CONTROL
    DIRECTION = DIRECTION_LOGIC
    SEGMENT_META_BASE = SEGMENT_TABLE_BASE
    SEGMENT_PULSES_DONE_HI = CURRENT_FREQ_LO
    SEGMENT_PULSES_DONE_LO = CURRENT_FREQ_HI
    ACTIVE_WAIT = STATE
    WAIT_ELAPSED_HI = CURRENT_SEGMENT
    WAIT_ELAPSED_LO = ERROR


SEGMENT_STRIDE = 0x10
SEGMENT_USED_REGISTERS = 8


def segment_base(index: int) -> int:
    if not 0 <= index < 10:
        raise ValueError("段索引必须为0~9")
    return int(Register.SEGMENT_TABLE_BASE) + index * SEGMENT_STRIDE


def register_read_count(_first: Register = Register.PULSE_POINT) -> int:
    """全部公共参数连续块0x1000~0x1015的长度。"""
    return int(Register.MAX_FREQ_HI) - int(Register.PULSE_POINT) + 1


class Command(IntEnum):
    START = 0x0001
    STOP = 0x0002
    CLEAR_POSITION = 0x0004
    RESUME = 0x0001  # 暂停状态下再次置START即继续
    # 兼容旧界面分支；新参数表没有这些独立命令码。
    CLEAR_TOTAL = 0x0008
    SAVE_PUBLIC = 0x0010
    SAVE_OUTPUTS = SAVE_PUBLIC  # 兼容旧名称；设备端现在保存全部公共参数
    CLEAR_DIAGNOSTIC = 0x0020
    UPDATE_FREQUENCY = 0x0040
    START_SEGMENTS = 0x0001


STATUS_REGISTER_COUNT = (
    int(Register.DIAG_PROFILE_ERRORS) - int(Register.CUMULATIVE_POSITION_LO) + 1
)


class DiagnosticStatus(IntEnum):
    ACTIVE = 0x0001
    COMPLETE = 0x0002
    COUNT_PASS = 0x0004
    FREQUENCY_PASS = 0x0008
    PROFILE_PASS = 0x0010
    COUNT_FAIL = 0x0100
    FREQUENCY_FAIL = 0x0200
    PROFILE_FAIL = 0x0400


DIAGNOSTIC_PHASE_TEXT = {
    0: "空闲",
    1: "加速段",
    2: "匀速段",
    3: "减速段",
}


class State(IntEnum):
    IDLE = 0
    RUNNING = 1
    COMPLETE = 2
    ERROR = 3
    PAUSED = 4
    WAITING = 5


class WaitCondition(IntEnum):
    WAIT_TIME = 0
    WAIT_SIGNAL = 1
    ACT_TIME = 2
    EXT_SIGNAL = 3
    EXT_OR_COMPLETE = 4


class RunMode(IntEnum):
    RELATIVE = 0
    ABSOLUTE = 1


class SendMode(IntEnum):
    COMPLETE = 0
    FOLLOW = 1


class DirectionLogic(IntEnum):
    POSITIVE = 0
    NEGATIVE = 1


class ProfileType(IntEnum):
    LINEAR = 0
    S_CURVE = 1
    SINE = 2


class OutputMode(IntEnum):
    PULSE_DIRECTION = 0
    AB_QUADRATURE = 1


class Error(IntEnum):
    NONE = 0
    BUSY = 1
    INVALID_PULSE_COUNT = 2
    INVALID_FREQUENCY = 3
    INVALID_DIRECTION = 4
    TIMER_RANGE = 5
    INVALID_COMMAND = 6
    NOT_PAUSED = 7
    NOT_ACTIVE = 8
    INVALID_PULSE_POINT = 9
    INVALID_DIRECTION_POINT = 10
    SAME_OUTPUT_POINT = 11
    FLASH_SAVE = 12
    INVALID_MODE = 13
    INVALID_PROFILE = 14
    INVALID_RAMP_TIME = 15
    INVALID_SPEED = 16
    INVALID_SEGMENT_COUNT = 17
    INVALID_START_SEGMENT = 18
    INVALID_SEGMENT_PARAM = 19
    SEGMENT_TOTAL_OVERFLOW = 20
    INVALID_WAIT_CONDITION = 21
    INVALID_INPUT_POINT = 22


STATE_TEXT = {
    State.IDLE: "空闲", State.RUNNING: "运行中", State.COMPLETE: "完成",
    State.ERROR: "错误", State.PAUSED: "已暂停", State.WAITING: "条件等待中",
}
ERROR_TEXT = {item: item.name for item in Error}
ERROR_TEXT[Error.NONE] = "无错误"
WAIT_CONDITION_TEXT = {
    WaitCondition.WAIT_TIME: "Wait时间",
    WaitCondition.WAIT_SIGNAL: "Wait信号（全局X4/X5）",
    WaitCondition.ACT_TIME: "ACT时间",
    WaitCondition.EXT_SIGNAL: "EXT信号（全局X4/X5）",
    WaitCondition.EXT_OR_COMPLETE: "EXT信号或脉冲完成",
}
MODE_TEXT = {RunMode.RELATIVE: "相对模式", RunMode.ABSOLUTE: "绝对模式"}
SEND_MODE_TEXT = {
    SendMode.COMPLETE: "完成模式（段间停止）",
    SendMode.FOLLOW: "后续模式（满足条件时连续衔接）",
}
DIRECTION_LOGIC_TEXT = {
    DirectionLogic.POSITIVE: "正逻辑（正向ON，反向OFF）",
    DirectionLogic.NEGATIVE: "负逻辑（正向OFF，反向ON）",
}
PROFILE_TEXT = {
    ProfileType.LINEAR: "直线加减速",
    ProfileType.S_CURVE: "S曲线加减速",
    ProfileType.SINE: "正弦曲线加减速",
}
OUTPUT_MODE_TEXT = {
    OutputMode.PULSE_DIRECTION: "PUL/DIR（脉冲+方向）",
    OutputMode.AB_QUADRATURE: "AB正交脉冲",
}
INPUT_POINTS = (4, 5)
PULSE_POINTS = {
    0: "Y0 / PF6 / TIM10_CH1", 1: "Y1 / PF8 / TIM13_CH1",
    2: "Y2 / PF7 / TIM11_CH1", 3: "Y3 / PF9 / TIM14_CH1",
}
SECONDARY_POINTS = {
    0: "Y12 / PH9", 1: "Y13 / PH8", 2: "Y14 / PH7", 3: "Y15 / PH6",
}


def point_id_from_label(options: Dict[int, str], label: str) -> int:
    for point, text in options.items():
        if text == label:
            return point
    raise ValueError(f"未知输出点：{label}")


def calculate_execution_total(
    segments: Sequence[Sequence[int]], start_segment: int
) -> Tuple[int, bool]:
    """按跳转段实际路径累计脉冲；检测到循环时返回(cycle前累计值, True)。"""
    if not segments or not 1 <= start_segment <= len(segments):
        raise ValueError("起始执行段超出有效段范围")

    current = start_segment
    visited = set()
    total = 0
    while current not in visited:
        visited.add(current)
        segment = segments[current - 1]
        total += abs(int(segment[0]))
        jump = int(segment[2])
        if jump == 0:
            if current >= len(segments):
                return total, False
            current += 1
        else:
            if not 1 <= jump <= len(segments):
                raise ValueError("跳转段超出有效段范围")
            current = jump
    return total, True


PRESETS = [
    {"name": "LED可视测试：10脉冲 / 1Hz", "pulses": 10, "frequency": 1,
     "acceleration_ms": 0, "deceleration_ms": 0,
     "start_frequency": 1, "end_frequency": 1},
    {"name": "基础测试：1000脉冲 / 1000Hz", "pulses": 1000, "frequency": 1000,
     "acceleration_ms": 1000, "deceleration_ms": 1000,
     "start_frequency": 100, "end_frequency": 100},
    {"name": "长行程测试：100000脉冲 / 5000Hz", "pulses": 100000, "frequency": 5000,
     "acceleration_ms": 1000, "deceleration_ms": 1000,
     "start_frequency": 100, "end_frequency": 100},
    {"name": "最高频率测试：100000脉冲 / 100000Hz", "pulses": 100000,
     "frequency": MAX_FREQUENCY_HZ, "acceleration_ms": 500,
     "deceleration_ms": 500, "start_frequency": 100, "end_frequency": 100},
]


def split_u32(value: int) -> List[int]:
    """参数表规定：低地址存低16位。"""
    if not 0 <= value <= 0xFFFFFFFF:
        raise ValueError("32位参数超出范围")
    return [value & 0xFFFF, (value >> 16) & 0xFFFF]


def signed_to_u32(value: int) -> int:
    if not -0x80000000 <= value <= 0x7FFFFFFF:
        raise ValueError("有符号32位参数超出范围")
    return value & 0xFFFFFFFF


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


def join_i32(low: int, high: int) -> int:
    value = join_u32(low, high)
    return value - 0x100000000 if value >= 0x80000000 else value


def state_text(value: int) -> str:
    try:
        return STATE_TEXT[State(value)]
    except (ValueError, KeyError):
        return f"未知状态 {value}"


def error_text(value: int) -> str:
    try:
        return ERROR_TEXT[Error(value)]
    except (ValueError, KeyError):
        return f"未知错误 {value}"


def parse_status(registers: List[int]) -> Dict[str, int]:
    """解析0x2000开始的状态区；兼容旧固件的11寄存器响应。"""
    if len(registers) < 11:
        raise ValueError("状态寄存器数量不足11个")
    result = {
        "cumulative_position": join_i32(registers[0], registers[1]),
        "current_frequency": join_u32(registers[2], registers[3]),
        "state": registers[4],
        "current_segment": registers[5],
        "error": registers[6],
        "current_pulses": join_u32(registers[7], registers[8]),
        "total_pulses": join_u32(registers[9], registers[10]),
    }
    if len(registers) >= STATUS_REGISTER_COUNT:
        result.update({
            "diagnostic_status": registers[11],
            "diagnostic_profile_type": registers[12],
            "diagnostic_phase": registers[13],
            "diagnostic_expected_pulses": join_u32(registers[14], registers[15]),
            "diagnostic_observed_pulses": join_u32(registers[16], registers[17]),
            "diagnostic_expected_frequency": join_u32(registers[18], registers[19]),
            "diagnostic_actual_frequency": join_u32(registers[20], registers[21]),
            "diagnostic_max_frequency_error": join_u32(registers[22], registers[23]),
            "diagnostic_count_errors": registers[24],
            "diagnostic_frequency_errors": registers[25],
            "diagnostic_profile_errors": registers[26],
        })
    else:
        result.update({
            "diagnostic_status": 0,
            "diagnostic_profile_type": 0,
            "diagnostic_phase": 0,
            "diagnostic_expected_pulses": 0,
            "diagnostic_observed_pulses": 0,
            "diagnostic_expected_frequency": 0,
            "diagnostic_actual_frequency": 0,
            "diagnostic_max_frequency_error": 0,
            "diagnostic_count_errors": 0,
            "diagnostic_frequency_errors": 0,
            "diagnostic_profile_errors": 0,
        })
    return result