#!/usr/bin/env python # # Copyright (c) 2019, Pycom Limited. # # This software is licensed under the GNU GPL version 3 or any # later version, with permitted additional terms. For more information # see the Pycom Licence v1.0 document supplied with this file, or # available at https://www.pycom.io/opensource/licensing # 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): 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) return response def _send_receive(self, modbus_pdu, slave_addr, count): serial_resp = bytearray() serial_pdu = bytearray() serial_pdu.append(slave_addr) serial_pdu.extend(modbus_pdu) crc = self._calculate_crc16(serial_pdu) serial_pdu.extend(crc) # flush the Rx FIFO self._uart.read() 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) return serial_resp 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)