#!/usr/bin/env python # -*- coding: utf-8 -*- """ RS485 总线管理器实现 此模块实现了 IBusManager 接口，用于管理 RS485 总线通信。 """ from ..logs.logger import log # 导入 MicroPython 的 UART 和 Pin 库 from machine import UART, Pin import time # 用于添加延时，确保RS485方向切换 import _thread # 用于线程同步 import struct # 用于浮点数转换 class RS485Manager: """ RS485 总线管理器。 负责 RS485 设备的指令发送、响应接收和数据解析。 """ def __init__(self, bus_config, default_timeouts): """ 构造函数，注入配置。 根据传入的配置初始化 RS485 总线对应的 UART 管理器。 Args: bus_config: 包含所有总线配置的字典。 键是总线ID，值是该总线的详细配置。 default_timeouts: 包含各种默认超时设置的字典。 """ self.bus_config = bus_config self.default_timeouts = default_timeouts # 存储以总线号为key的UART管理器实例、RTS引脚和锁 self.bus_ports = {} log("RS485Manager 已使用配置初始化。") log(f"总线配置: {self.bus_config}") log(f"默认超时设置: {self.default_timeouts}") # 遍历 bus_config，初始化 RS485 端口 for bus_id, config in bus_config.items(): if config.get('protocol') == 'RS485': try: uart_id = config['uart_id'] baudrate = config['baudrate'] pins = config['pins'] tx_pin_num = pins['tx'] rx_pin_num = pins['rx'] rts_pin_num = pins['rts'] # RS485 的 DE/RE 方向控制引脚 # 初始化 Pin 对象 rts_pin = Pin(rts_pin_num, Pin.OUT) # RTS 引脚设置为输出模式 rts_pin.value(0) # 默认设置为接收模式 # 初始化 UART 对象 # 注意：MicroPython 的 UART 构造函数可能不支持直接传入 Pin 对象，而是 Pin 编号 # 并且 rts 参数通常用于硬件流控制，RS485 的 DE/RE 需要手动控制 uart = UART(uart_id, baudrate=baudrate, tx=tx_pin_num, rx=rx_pin_num, timeout=self.default_timeouts.get('rs485_response', 500)) self.bus_ports[bus_id] = { 'uart': uart, 'rts_pin': rts_pin, 'lock': _thread.allocate_lock() } log(f"总线 {bus_id} (RS485) 的 UART 管理器初始化成功。UART ID: {uart_id}, 波特率: {baudrate}, TX: {tx_pin_num}, RX: {rx_pin_num}, RTS(DE/RE): {rts_pin_num}") except KeyError as e: log(f"错误: 总线 {bus_id} 的 RS485 配置缺少关键参数: {e}") except Exception as e: log(f"错误: 初始化总线 {bus_id} 的 RS485 管理器失败: {e}") else: log(f"总线 {bus_id} 的协议不是 RS485，跳过初始化。") @staticmethod def _calculate_crc16_modbus(data): """ 计算 Modbus RTU 的 CRC16 校验码。 """ crc = 0xFFFF for byte in data: crc ^= byte for _ in range(8): if crc & 0x0001: crc >>= 1 crc ^= 0xA001 else: crc >>= 1 return crc @staticmethod def _parse_modbus_rtu_float_default(response_bytes): """ 默认解析 Modbus RTU 响应中的 32 位 IEEE 754 单精度浮点数。 假定为大端 (ABCD) 字节序。 """ # 最小预期长度: 从站ID(1) + 功能码(1) + 字节计数(1) + 4字节数据 + CRC(2) = 9字节 min_response_len = 9 expected_data_byte_count = 4 # 32位浮点数占用4字节 if not response_bytes or len(response_bytes) < min_response_len: log(f"警告: 响应字节过短或为空，无法解析为浮点数。响应: {response_bytes.hex() if response_bytes else 'None'}") return None # 提取响应组件 # 注意: Modbus RTU CRC是LSB在前，所以这里需要调整 # response_bytes[:-2] 是用于CRC计算的数据部分 # response_bytes[-2:] 是CRC本身 data_for_crc = response_bytes[:-2] received_crc = (response_bytes[-1] << 8) | response_bytes[-2] # CRC的低字节在前，高字节在后 # 1. CRC 校验 calculated_crc = RS485Manager._calculate_crc16_modbus(data_for_crc) if calculated_crc != received_crc: log(f"错误: CRC校验失败。接收CRC: {received_crc:04X}, 计算CRC: {calculated_crc:04X}. 响应: {response_bytes.hex()}") return None function_code = response_bytes[1] byte_count = response_bytes[2] data_bytes = response_bytes[3:3 + expected_data_byte_count] # 2. 功能码检查 (假设读取保持寄存器0x03或输入寄存器0x04) if function_code not in [0x03, 0x04]: log(f"警告: 响应功能码 {function_code:02X} 不符合预期 (期望0x03或0x04)。响应: {response_bytes.hex()}") return None # 3. 字节计数检查 if byte_count != expected_data_byte_count: log(f"警告: 响应字节计数 {byte_count} 不符合预期 (期望{expected_data_byte_count})。响应: {response_bytes.hex()}") return None # 4. 提取的数据字节长度检查 (与字节计数检查有重叠，但更安全) if len(data_bytes) != expected_data_byte_count: log(f"错误: 提取的数据字节长度不正确。期望{expected_data_byte_count}, 实际{len(data_bytes)}. 响应: {response_bytes.hex()}") return None # 5. 转换为浮点数 (大端, ABCD) try: parsed_float = struct.unpack('>f', data_bytes)[0] log(f"成功解析浮点数: {parsed_float}") return parsed_float except Exception as e: log(f"错误: 浮点数转换失败: {e}. 数据字节: {data_bytes.hex()}. 响应: {response_bytes.hex()}") return None def execute_raw_command(self, bus_id, command): """ 【契约】执行一个“发后不理”的原始指令。 Args: bus_id (int): 目标总线的编号。 command (bytes): 要发送的原始命令字节。 """ if bus_id not in self.bus_ports: log(f"错误: 未找到总线 {bus_id} 的 RS485 配置。") return port_info = self.bus_ports[bus_id] uart = port_info['uart'] rts_pin = port_info['rts_pin'] lock = port_info['lock'] with lock: try: rts_pin.value(1) # 设置为发送模式 (DE/RE = HIGH) time.sleep_us(100) # 短暂延时，确保方向切换完成 uart.write(command) # 等待所有数据发送完毕 uart.flush() time.sleep_us(100) # 短暂延时，确保数据完全发出 rts_pin.value(0) # 切换回接收模式 (DE/RE = LOW) log(f"总线 {bus_id} 原始命令发送成功: {command.hex()}") except Exception as e: log(f"错误: 在总线 {bus_id} 上执行原始命令失败: {e}") def execute_collect_task(self, task): """ 【契约】执行一个完整的采集任务，并直接返回最终的数值。 一个符合本接口的实现必须自己处理所有细节： - 从task字典中解析出 bus_id, command, parser_type。 - 发送指令。 - 接收响应。 - 根据parser_type选择正确的内部解析器进行解析。 - 返回最终的float数值，或在任何失败情况下返回None。 Args: task: 从Protobuf解码出的单个CollectTask消息字典。 期望结构: {"command": {"bus_number": int, "command_bytes": bytes}} Returns: 成功解析则返回数值，否则返回None。 """ # I. 任务参数解析与初步验证 try: command_info = task.get("command") if not command_info: log("错误: CollectTask 缺少 'command' 字段。") return None bus_id = command_info.get("bus_number") command_bytes = command_info.get("command_bytes") if bus_id is None or command_bytes is None: log("错误: Raw485Command 缺少 'bus_number' 或 'command_bytes' 字段。") return None except Exception as e: log(f"错误: 解析CollectTask失败: {e}. 任务: {task}") return None if bus_id not in self.bus_ports: log(f"错误: 未找到总线 {bus_id} 的 RS485 配置。") return None port_info = self.bus_ports[bus_id] uart = port_info['uart'] rts_pin = port_info['rts_pin'] lock = port_info['lock'] response_bytes = None with lock: try: # II. 线程安全与指令发送 rts_pin.value(1) # 设置为发送模式 (DE/RE = HIGH) time.sleep_us(100) # 短暂延时，确保方向切换完成 uart.write(command_bytes) uart.flush() time.sleep_us(100) # 短暂延时，确保数据完全发出 rts_pin.value(0) # 切换回接收模式 (DE/RE = LOW) log(f"总线 {bus_id} 原始命令发送成功: {command_bytes.hex()}") # III. 接收响应 response_bytes = uart.read() if response_bytes: log(f"总线 {bus_id} 收到响应: {response_bytes.hex()}") else: log(f"警告: 总线 {bus_id} 未收到响应或响应超时。") return None except Exception as e: log(f"错误: 在总线 {bus_id} 上执行采集命令失败: {e}") return None # IV. 响应解析与数据提取 (默认 Modbus RTU 浮点数) # TODO: 根据CollectTask的Protobuf定义，此处需要根据parser_type来选择具体的解析逻辑和类型。 # 目前默认使用Modbus RTU大端浮点数解析。 parsed_value = RS485Manager._parse_modbus_rtu_float_default(response_bytes) # V. 返回结果 return parsed_value