1. 农机CAN总线控制指令解析与执行

任务边界与报文约定

本例实现一个可独立运行的 CAN 指令解析与执行模块：上位机通过标准帧下发速度、转向、农具升降三类指令，模块完成帧过滤、字节序解析、量程换算、安全限幅，并按固定周期输出执行器控制信号。为便于验证，先约定报文格式（示例参数，可按实际车型调整）：

执行侧约定：速度输出限幅 0~12 km/h（作业工况），转角限幅 ±35°，升降动作需在收到指令后 50 ms 内响应，且停止指令优先级最高。下面用一张信号链图把“帧 → 解析 → 限幅 → 执行器”的映射关系固定下来，后续代码严格按此实现。

解析与执行模块实现

下面用 Python 实现核心逻辑，不依赖第三方 CAN 库，便于直接运行验证。帧对象用轻量结构体模拟，解析函数与执行函数分离，方便单独测试。

from dataclasses import dataclass, field

@dataclass
class CanFrame:
    can_id: int
    data: bytes

@dataclass
class ActuatorCmd:
    speed_kmh: float = 0.0
    steer_deg: float = 0.0
    lift_action: int = 0      # 0 停止 1 上升 2 下降
    lift_pwm: int = 0
    valid: bool = False

def _s16_le(b0: int, b1: int) -> int:
    v = b0 | (b1 << 8)
    return v - 0x10000 if v & 0x8000 else v

def parse_frame(frame: CanFrame) -> ActuatorCmd:
    cmd = ActuatorCmd()
    if len(frame.data) != 8:
        return cmd
    if frame.can_id == 0x180:
        cmd.speed_kmh = _s16_le(frame.data[0], frame.data[1]) * 0.01
    elif frame.can_id == 0x181:
        cmd.steer_deg = _s16_le(frame.data[0], frame.data[1]) * 0.1
    elif frame.can_id == 0x182:
        action = frame.data[0]
        if action in (0, 1, 2):
            cmd.lift_action = action
            cmd.lift_pwm = min(max(frame.data[1], 0), 100)
    else:
        return cmd
    cmd.valid = True
    return cmd

def clamp(cmd: ActuatorCmd) -> ActuatorCmd:
    cmd.speed_kmh = min(max(cmd.speed_kmh, 0.0), 12.0)
    cmd.steer_deg = min(max(cmd.steer_deg, -35.0), 35.0)
    if cmd.lift_action == 0:
        cmd.lift_pwm = 0
    return cmd

def execute(cmd: ActuatorCmd) -> dict:
    if not cmd.valid:
        return {"speed": 0.0, "steer": 0.0, "lift": 0, "pwm": 0}
    return {
        "speed": round(cmd.speed_kmh, 2),
        "steer": round(cmd.steer_deg, 1),
        "lift": cmd.lift_action,
        "pwm": cmd.lift_pwm,
    }

验证解析正确性与时序

用一组覆盖正负值、边界值和非法帧的用例验证解析结果，并检查停止指令是否强制清零 PWM：

cases = [
    (CanFrame(0x180, bytes([0xE8, 0x03, 0, 0, 0, 0, 0, 0])), 10.0),   # 1000 → 10.00 km/h
    (CanFrame(0x180, bytes([0x20, 0x03, 0, 0, 0, 0, 0, 0])), 8.0),    # 800 → 8.00 km/h
    (CanFrame(0x181, bytes([0x9C, 0xFF, 0, 0, 0, 0, 0, 0])), -10.0),  # -100 → -10.0°
    (CanFrame(0x181, bytes([0x5E, 0x01, 0, 0, 0, 0, 0, 0])), 35.0),   # 350 → 35.0°
    (CanFrame(0x182, bytes([2, 80, 0, 0, 0, 0, 0, 0])), 2),           # 下降 80%
    (CanFrame(0x182, bytes([0, 80, 0, 0, 0, 0, 0, 0])), 0),           # 停止 → PWM 清零
]

for frame, expect in cases:
    cmd = clamp(parse_frame(frame))
    out = execute(cmd)
    key = "speed" if frame.can_id == 0x180 else ("steer" if frame.can_id == 0x181 else "lift")
    assert out[key] == expect, (frame.can_id, out, expect)
    print(f"0x{frame.can_id:03X} → {out}")

# 非法帧：DLC 不足、未知 ID
assert not parse_frame(CanFrame(0x180, bytes([1, 2]))).valid
assert not parse_frame(CanFrame(0x200, bytes(8))).valid
print("全部用例通过")

时序验证关注两点：一是解析到执行输出的处理链应在 10 ms 周期内完成，二是从帧到达至执行器响应不超过 50 ms。可在 parse_frame 入口与 execute 出口各打一个时间戳，统计差值；若超过 50 ms，优先检查总线负载与解析函数中的阻塞操作。