# -*- coding: utf-8 -*- # Copyright (c) 2025 SDRLDQ # All rights reserved # import functions import const as Const from machine import UART from machine import Pin import struct import time import machine class Serial: def __init__(self, uart_id, baudrate=9600, data_bits=8, stop_bits=1, parity=None, pins=None, ctrl_pin=None): self._uart = UART(uart_id, baudrate=baudrate, bits=data_bits, parity=parity,stop=stop_bits) if ctrl_pin is not None: self._ctrlPin = Pin(ctrl_pin, mode=Pin.OUT) #定义发送接收控制管脚 else: self._ctrlPin = None self.char_time_ms = (1000 * (data_bits + stop_bits + 2)) // baudrate def _calculate_crc16(self, data): #CRC16校验 crc = 0xFFFF for char in data: crc = (crc >> 8) ^ Const.CRC16_TABLE[((crc) ^ char) & 0xFF] return struct.pack('= Const.ERROR_BIAS: if len(response) < Const.ERROR_RESP_LEN: return False elif (Const.READ_COILS <= response[1] <= Const.READ_INPUT_REGISTER): expected_len = Const.RESPONSE_HDR_LENGTH + 1 + response[2] + Const.CRC_LENGTH if len(response) < expected_len: return False elif len(response) < Const.FIXED_RESP_LEN: return False return True def _uart_read(self): response = bytearray()#定义 存储从串口读取的数据 for x in range(1, 40): if self._uart.any():#检查串口缓冲区是否有待读数据 response.extend(self._uart.read())#读串口所有数据 # variable length function codes may require multiple reads if self._exit_read(response):#判断串口数据是否存在停止条件 break time.sleep(0.05)#50ms return response def _send_receive(self, modbus_pdu, slave_addr, count): serial_resp = bytearray()#存储从设备返回的响应数据 serial_pdu = bytearray()#构建要发送的 modbus RTU协议数据单元PDU serial_pdu.append(slave_addr)#构建modbus RTU 帧头 从机地址：1字节 serial_pdu.extend(modbus_pdu)#构建modbus RTU 帧头附件协议数据： 功能码+数据段 crc = self._calculate_crc16(serial_pdu) #CRC16校验码 serial_pdu.extend(crc)#将CRC16校验码附加到帧尾，形成完整帧 self._uart.read() # flush the Rx FIFO 接收并丢弃串口缓冲区的数据，避免干扰本次通讯 if self._ctrlPin: self._ctrlPin(1) self._uart.write(serial_pdu) if self._ctrlPin: while not self._uart.wait_tx_done(2): machine.idle() time.sleep_ms(1 + self.char_time_ms) self._ctrlPin(0) serial_resp = self._validate_resp_hdr(self._uart_read(), slave_addr, modbus_pdu[0], count) #读取数据并验证 地址功能码modbus_pdu[0] 数据长度及CRC校验 #剥离协议头尾，返回业务数据 return serial_resp ################################################## #_validate_resp_hdr() #读取数据并验证 地址功能码modbus_pdu[0] 数据长度及CRC校验 #剥离协议头尾，返回业务数据 ################################################### def _validate_resp_hdr(self, response, slave_addr, function_code, count): if len(response) == 0: raise OSError('no data received from slave') resp_crc = response[-Const.CRC_LENGTH:] expected_crc = self._calculate_crc16(response[0:len(response) - Const.CRC_LENGTH]) if (resp_crc[0] != expected_crc[0]) or (resp_crc[1] != expected_crc[1]): raise OSError('invalid response CRC') if (response[0] != slave_addr): raise ValueError('wrong slave address') if (response[1] == (function_code + Const.ERROR_BIAS)): raise ValueError('slave returned exception code: {:d}'.format(response[2])) hdr_length = (Const.RESPONSE_HDR_LENGTH + 1) if count else Const.RESPONSE_HDR_LENGTH return response[hdr_length : len(response) - Const.CRC_LENGTH] # READ_COILS = 0x01 读线圈寄存器 def read_coils(self, slave_addr, starting_addr, coil_qty): modbus_pdu = functions.read_coils(starting_addr, coil_qty) resp_data = self._send_receive(modbus_pdu, slave_addr, True) status_pdu = self._bytes_to_bool(resp_data) return status_pdu # READ_DISCRETE_INPUTS = 0x02 读离散输入寄存器 def read_discrete_inputs(self, slave_addr, starting_addr, input_qty): modbus_pdu = functions.read_discrete_inputs(starting_addr, input_qty) resp_data = self._send_receive(modbus_pdu, slave_addr, True) status_pdu = self._bytes_to_bool(resp_data) return status_pdu # READ_HOLDING_REGISTERS = 0x03 读保持寄存器 def read_holding_registers(self, slave_addr, starting_addr, register_qty, signed=True): modbus_pdu = functions.read_holding_registers(starting_addr, register_qty) resp_data = self._send_receive(modbus_pdu, slave_addr, True) register_value = self._to_short(resp_data, signed) return register_value # READ_INPUT_REGISTER = 0x04 读输入寄存器 def read_input_registers(self, slave_addr, starting_address, register_quantity, signed=True): modbus_pdu = functions.read_input_registers(starting_address, register_quantity) resp_data = self._send_receive(modbus_pdu, slave_addr, True) register_value = self._to_short(resp_data, signed) return register_value # WRITE_SINGLE_COIL = 0x05 写单个线圈寄存器 def write_single_coil(self, slave_addr, output_address, output_value): modbus_pdu = functions.write_single_coil(output_address, output_value) resp_data = self._send_receive(modbus_pdu, slave_addr, False) operation_status = functions.validate_resp_data(resp_data, Const.WRITE_SINGLE_COIL, output_address, value=output_value, signed=False) return operation_status # WRITE_SINGLE_REGISTER = 0x06 写单个保持寄存器 def write_single_register(self, slave_addr, register_address, register_value, signed=True): modbus_pdu = functions.write_single_register(register_address, register_value, signed) resp_data = self._send_receive(modbus_pdu, slave_addr, False) operation_status = functions.validate_resp_data(resp_data, Const.WRITE_SINGLE_REGISTER, register_address, value=register_value, signed=signed) return operation_status # WRITE_MULTIPLE_COILS = 0x0F 写多个线圈寄存器 def write_multiple_coils(self, slave_addr, starting_address, output_values): modbus_pdu = functions.write_multiple_coils(starting_address, output_values) resp_data = self._send_receive(modbus_pdu, slave_addr, False) operation_status = functions.validate_resp_data(resp_data, Const.WRITE_MULTIPLE_COILS, starting_address, quantity=len(output_values)) return operation_status # WRITE_MULTIPLE_REGISTERS = 0x10 写多个保持寄存器 def write_multiple_registers(self, slave_addr, starting_address, register_values, signed=True): modbus_pdu = functions.write_multiple_registers(starting_address, register_values, signed) resp_data = self._send_receive(modbus_pdu, slave_addr, False) operation_status = functions.validate_resp_data(resp_data, Const.WRITE_MULTIPLE_REGISTERS, starting_address, quantity=len(register_values)) return operation_status if __name__ == '__main__': port = UART() port.write_multiple_registers(2,0,0xff00) # End File