工控机和树莓派采用USB转TTL串口通信,后者使用GPIO口,收发16进制数组的Python代码
接线
采用USB转TTL模块,Linux系统的工控机接USB插口,TTL模块输出RX、TX、VCC、GND,树莓派的GPIO14口也就是RXD0接TTL串口的TX。
在树莓派上使用Python的RPi.GPIO库,配置GPIO14(RXD0)作为接收引脚。
在树莓派上编写接收数据的代码,通过GPIO.input(14)检测GPIO14(RXD0)的状态,如果为高电平表示有数据接收,然后使用ser.read(ser.in_waiting)读取串口接收缓冲区中的数据。
通信协议
串口通信发送一串数据,包含2个头、总长度为6个的16进制字符数组,表示为uint8 data[6]。每条数据前2个字节是0xAA,0xAA的开头;第3个字节是指令的类型;第4,5位是指令,它被编码为2位的uint8;采用CRC8进制对前面5位校验,得到第6位。典型符合规范的数据格式如下:
0xAA , 0xAA , 0x09 , 0x07 , 0x01 , 0x32
其中有0x07 , 0x01 是被编码好的指令。
发送
有无发送指令的python版的程序可以参考的。需要将每个子功能写成单独的函数,包括如打开串口、发送指令、指令的编码、CRC校验等
ChatGPT
以下是一个示例的Python程序,用于发送串口数据、进行指令编码和CRC校验的过程。程序假设使用pyserial库进行串口通信。
# -*- coding: UTF-8 -*-
#!/usr/bin/python3 python3
import serial
import struct
import time# 配置串口参数
port = '/dev/ttyUSB0' # 串口设备路径
baudrate = 9600 # 波特率# 打开串口
ser = serial.Serial(port, baudrate)# 定义指令类型的常量
INSTRUCTION_TYPE = 0x09# CRC8校验函数
def crc8(data):crc = 0x00polynomial = 0x8C # CRC-8多项式for byte in data:crc ^= bytefor _ in range(8):if crc & 0x80:crc = (crc << 1) ^ polynomialelse:crc <<= 1return crc & 0xFF# 将时间秒数编码为两位16进制数字
def encode_seconds(seconds):seconds_hex = hex(seconds)[2:].zfill(2) # 将秒数转换为两位的十六进制字符串return int(seconds_hex, 16) # 将十六进制字符串转换为整数# 编码指令
def encode_instruction(data):encoded_data = []for value in data:encoded_data.append(value & 0xFF)return encoded_data# 发送指令
def send_instruction(instruction):# 构建指令data = bytearray()data.extend([0xAA, 0xAA]) # 前导字节data.append(INSTRUCTION_TYPE) # 指令类型data.extend(encode_instruction(instruction)) # 编码后的指令数据# 计算校验码crc = crc8(data[:5]) # 计算校验码时不包括最后一个字节(校验码本身)data.append(crc) # 添加校验码的低字节return datatry:# 主循环while True:# 测试发送指令# instruction = [0x07, 0x08] # 指令数据# 获取当前时间的秒数seconds = int(time.time())# print(seconds)# 发送时间秒数编码instruction = encode_seconds(seconds).to_bytes(4, 'big')# 发送一个字节的数据,高位在前data = send_instruction(instruction[2:4])formatted = ' '.join(f'{byte:02X}' for byte in data)print(formatted)# 发送指令ser.write(data)# # 接收指令# response = receive_command()# if response:# print('Received:', response)# 休眠一段时间,控制发送频率time.sleep(0.2) # 休眠0.2秒except KeyboardInterrupt:# Ctrl+C被按下,执行异常处理ser.close()print("程序终止,关闭串口")except Exception as e:# 其他异常处理print("发生异常:", str(e))ser.close()
在上述示例中,我们定义了一个encode_instruction()
函数来对指令数据进行编码,将每个元素限定在0-255的范围内,并返回编码后的数据。然后我们构建指令,添加前导字节、指令类型、编码后的指令数据,并计算校验码。最后,我们使用ser.write()
函数将指令数据发送到串口。
请注意,在实际应用中,你可能还需要添加错误处理、超时设置、数据处理等功能,以确保稳定和可靠的串口通信。以上示例仅提供了一个基本的框架,具体实现可能需要根据你的具体需求进行调整和扩展。
接收
串口通信接收一串数据,包含2个头、总长度为6个的16进制字符数组,表示为uint8 data[6]。每条数据前2个字节是0xAA,0xAA的开头;第3个字节是指令的类型;第4,5位是被编码好的指令,需要被解析为int16数字;采用CRC16进制对前面5位校验,得到第6位,MODBUS 算法。
有无python版的程序可以参考的。需要将每个子功能写成单独的函数,包括如打开串口、接收指令、指令的解析、CRC校验等
ChatGPT
以下是一个示例的Python程序,用于接收串口数据、进行CRC校验和解析指令的过程。程序假设使用pyserial库进行串口通信。
# -*- coding: UTF-8 -*-
#!/usr/bin/python3 python3
import serial
import struct
import RPi.GPIO as GPIO# 配置GPIO
GPIO.setmode(GPIO.BCM)
GPIO.setup(14, GPIO.IN) # GPIO14作为接收引脚# 配置串口参数
port = '/dev/serial0' # 树莓派上UART接口所对应的串口设备路径
baudrate = 9600 # 波特率
# 打开串口
ser = serial.Serial(port, baudrate)# 配置GPIO
GPIO.setmode(GPIO.BCM)
GPIO.setup(14, GPIO.IN) # GPIO14作为接收引脚# 定义指令类型的常量
INSTRUCTION_TYPE = 0x09# CRC8校验
def crc8(data):crc = 0x00polynomial = 0x8C # CRC-8多项式for byte in data:crc ^= bytefor _ in range(8):if crc & 0x80:crc = (crc << 1) ^ polynomialelse:crc <<= 1return crc & 0xFF# 解析指令
def parse_instruction(data):if len(data) != 6:print("Invalid instruction length")return# 校验头部if data[0] != 0xAA or data[1] != 0xAA:print("Invalid header")return# 校验指令类型if data[2] != INSTRUCTION_TYPE:print("Invalid instruction type")return# 校验CRCcrc = crc8(data[:5]) # 计算校验码时不包括最后一个字节(校验码本身)if crc != data[5]:print("CRC check failed")return# 获取指令数据instruction = data[3:5]# 执行相应的指令逻辑process_instruction(instruction)# 处理指令逻辑
def process_instruction(instruction):# 按照具体需求进行指令处理cmd16 = struct.unpack('<h', bytearray(instruction))[0] # 解析为int16数字print("Received Instruction:", cmd16)try:# 接收数据while True:if GPIO.input(14): # 检查GPIO14状态,判断是否有数据接收# data = ser.read(ser.in_waiting) # 一直读取数据data = ser.read(6) # 读取固定长度的数据# 接收数据if len(data) == 6 and data[:2] == b'\xAA\xAA':parse_instruction(data)except KeyboardInterrupt:# Ctrl+C被按下,执行异常处理ser.close()print("程序终止,关闭串口")except Exception as e:# 其他异常处理print("发生异常:", str(e))ser.close()
在上述示例中,我们定义了一个parse_instruction()
函数来解析接收到的指令数据。首先,我们检查指令长度是否为6个字节,并进行CRC校验。如果校验失败或指令类型不匹配,将输出相应的错误信息。如果校验通过,我们解析指令数据并调用process_instruction()
函数来执行相应的指令逻辑。
请注意,以上示例仅提供了一个基本的框架,你可能需要根据具体的需求进行调整和扩展。在实际应用中,你可能还需要添加错误处理、超时设置、数据处理等功能,以确保稳定和可靠的串口通信。
串口通信如何控制数据发送和解析的频率?
ChatGPT
在串口通信中,你可以使用不同的方法来控制数据发送的频率。以下是几种常见的方法:
-
延时发送:在发送每个数据包之间添加固定的延时时间。你可以使用
time.sleep()
函数在发送数据包之间添加延时,以控制发送频率。例如,time.sleep(0.1)
将在发送数据包后延迟100毫秒。这个方式不推荐。 -
定时发送:使用定时器来定期发送数据。你可以使用Python的
threading.Timer
类创建一个定时器,并在定时器回调函数中发送数据。
在串口通信中,使用threading.Timer
控制数据接收和解析的频率可以通过以下步骤实现:
- 创建一个定时器对象,指定定时器的时间间隔。
- 在定时器回调函数中执行数据接收和解析的逻辑。
- 启动定时器,使其开始定时执行回调函数。
下面是一个示例,演示如何使用threading.Timer控制串口数据接收和解析的频率:
import serial
import threading# 配置串口参数
port = '/dev/ttyUSB0' # 串口设备路径
baudrate = 9600 # 波特率# 打开串口
ser = serial.Serial(port, baudrate)# 定义数据接收和解析的函数
def receive_and_parse_data():# 接收数据data = ser.readline().strip()# 解析数据# TODO: 在这里添加你的数据解析逻辑print("Received data:", data)# 重新启动定时器,控制数据接收和解析的频率timer = threading.Timer(1.0, receive_and_parse_data)timer.start()# 启动定时器,控制数据接收和解析的频率
timer = threading.Timer(1.0, receive_and_parse_data)
timer.start()# 阻塞主线程,保持程序运行
while True:pass# 关闭串口
ser.close()
在上述示例中,receive_and_parse_data()
函数被定时器调用,定时执行串口数据的接收和解析。定时器的时间间隔为1.0秒,你可以根据需要进行调整。
无论使用延时发送还是定时发送,都可以根据实际需求调整发送数据的频率。请注意,实际的发送频率可能受到串口传输速率、数据包大小和目标设备的处理能力等因素的限制。因此,需要根据具体的应用场景和需求进行调整和测试。