设为首页收藏本站
查看: 724|回复: 0

【米尔瑞萨RZ/G2L开发板-创新应用】基于python的tcp、udp、串口收发使用

[复制链接]

3

主题

0

回帖

31

积分

新手上路

积分
31
感冒不喝水 发表于 2023-9-23 16:30:59 | 显示全部楼层 |阅读模式
由于开发板支持python开发,就想着使用python完成应用产品的开发,免去了交叉编译以及编写makefile的步骤,何乐而不为呢。。。
下面以建立tcp服务器,udp通信先进行说明,由于串口需要使用serial包,故而需要进行安装,安装教程见【米尔瑞萨RZ/G2L开发板-创新应用】+ 安装pip3和第三方Python库 - 米尔RZ/G2L开发板 - 米尔科技论坛 - Powered by Discuz! (myir-tech.com)

tcp服务器的建立
  1. # -*- coding: utf-8 -*-

  2. import socket  # 导入socket模块
  3. import threading

  4. # server 接收端


  5. class tcp(threading.Thread):
  6.     def __init__(self, ip: str = '', port: int = 9000):
  7.         threading.Thread.__init__(self)
  8.         self.ip = ip
  9.         self.port = port

  10.     def run(self):
  11.         # 创建socket
  12.         tcp_server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

  13.         # 本地信息
  14.         address = (self.ip, self.port)

  15.         # 绑定
  16.         tcp_server_socket.bind(address)

  17.         tcp_server_socket.listen(128)

  18.         while True:
  19.             # 等待新的客户端连接
  20.             client_socket, clientAddr = tcp_server_socket.accept()
  21.             while True:
  22.                 # 接收对方发送过来的数据
  23.                 recv_data = client_socket.recv(1024)  # 接收1024个字节
  24.                 if recv_data:
  25.                     print('接收到的数据为:', recv_data.decode('gbk'))
  26.                     client_socket.sendto(recv_data, clientAddr)
  27.                 else:
  28.                     break
  29.             client_socket.close()

  30.         tcp_server_socket.close()
复制代码
udp服务的建立
  1. # -*- coding: utf-8 -*-

  2. import socket  # 导入socket模块
  3. import time  # 导入time模块
  4. import threading

  5. # server 接收端
  6. global cfg

  7. class udp(threading.Thread):
  8.     def __init__(self, ip: str = '', port: int = 9000):
  9.         threading.Thread.__init__(self)
  10.         self.ip = ip
  11.         self.port = port

  12.     def run(self):
  13.         # 创建一个套接字socket对象,用于进行通讯
  14.         # socket.AF_INET 指明使用INET地址集,进行网间通讯
  15.         # socket.SOCK_DGRAM 指明使用数据协议,即使用传输层的udp协议
  16.         server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  17.         # 接收来自任何IP的广播帧
  18.         address = (self.ip, self.port)
  19.         server_socket.bind(address)  # 为服务器绑定一个固定的地址,ip和端口
  20.         # server_socket.settimeout(10)  # 设置一个时间提示,如果10秒钟没接到数据进行提示

  21.         while True:
  22.             # 正常情况下接收数据并且显示,如果10秒钟没有接收数据进行提示(打印 'time out')
  23.             # 当然可以不要这个提示,那样的话把'try:' 以及 'except'后的语句删掉就可以了
  24.             try:
  25.                 now = time.time()  # 获取当前时间
  26.                 # 接收客户端传来的数据 recvfrom接收客户端的数据,默认是阻塞的,直到有客户端传来数据
  27.                 # recvfrom 参数的意义,表示最大能接收多少数据,单位是字节
  28.                 # recvfrom返回值说明
  29.                 # receive_data表示接受到的传来的数据,是bytes类型
  30.                 # client  表示传来数据的客户端的身份信息,客户端的ip和端口,元组
  31.                 receive_data, client = server_socket.recvfrom(1024)
  32.                 print(time.strftime('%Y-%m-%d %H:%M:%S',
  33.                                     time.localtime(now)))  # 以指定格式显示时间
  34.                 print('来自客户端%s,发送的%s\n' % (client, receive_data))  # 打印接收的内容
  35.                 server_socket.sendto(receive_data, client)
  36.             except socket.timeout:  # 如果10秒钟没有接收数据进行提示(打印 'time out')
  37.                 print('time out')
复制代码
串口收发的使用

  1. import time
  2. import serial
  3. import threading

  4. class sci(threading.Thread):
  5.     def __init__(self, port: str = '/dev/ttySC3', baudrate: int = 115200,parity:str = serial.PARITY_NONE):
  6.         threading.Thread.__init__(self)
  7.         self.port = port
  8.         self.baudrate = baudrate
  9.         self.parity = parity
  10.         self.serial_port = serial.Serial(
  11.             port = self.port,
  12.             baudrate = self.baudrate,
  13.             bytesize = serial.EIGHTBITS,
  14.             parity = self.parity,
  15.             stopbits = serial.STOPBITS_ONE
  16.         )
  17.     def run(self):
  18.         if not self.serial_port.is_open:
  19.             self.serial_port.open()
  20.         time.sleep(1)
  21.         # Send a simple header
  22.         try:
  23.             while True:
  24.                 if self.serial_port.in_waiting > 0:
  25.                     data = self.serial_port.read_all()
  26.                     self.serial_port.write(data)
  27.                 time.sleep(1)

  28.         except KeyboardInterrupt as exception:
  29.             print("Exiting Program")
  30.             print(exception.args)

  31.         except Exception as exception_error:
  32.             print("Error occurred. Exiting Program")
  33.             print("Error: " + str(exception_error))
  34.             print(exception_error.args)
  35.             pass
  36.         finally:
  37.             self.serial_port.close()
  38.             pass

  39.             
复制代码


为了进行后期扩展,所以将各模块分别建立为class,需要建立main函数进行初始化
  1. # -*- coding: utf-8 -*-
  2. from udp import *
  3. from tcp import *
  4. from sci import *
  5. from threading import Thread

  6. if __name__ == "__main__":
  7.     # 创建 Thread 实例
  8.     t1 = udp('', 9000)
  9.     t1.start()

  10.     tcp_thread = []
  11.     serial_thread = []
  12.     for x in range(1):
  13.         tcp_thread.append(tcp('', 5000))
  14.         serial_thread.append(
  15.             sci(port='/dev/ttySC3', baudrate=115200), parity= serial.PARITY_NONE)

  16.     # 启动线程运行
  17.     for x in range(1):
  18.         tcp_thread[x].start()
  19.         serial_thread[x].start()

  20.     # 等待所有线程执行完毕
  21.     t1.join()  # join() 等待线程终止,要不然一直挂起
  22.     for x in range(1):
  23.         tcp_thread[x].join()
  24.         serial_thread[x].join()
复制代码
使用命令启动应用程序
  1. python3 main.py
复制代码
运行后下图分别为使用windows测试工具进行测试
tcp通信测试
tcp.jpg
udp通信测试
udp.jpg
串口通信测试
串口.jpg

最后告诉一声测试的小伙伴,使用的串口为 /dev/ttySC3,对应的硬件为调试口下方的串口哦,千万别接错了
回复

使用道具 举报

您需要登录后才可以回帖 登录

本版积分规则

Archiver|手机版|小黑屋|米尔科技论坛   

GMT+8, 2024-7-14 02:53 , Processed in 0.047025 second(s), 22 queries .

Powered by Discuz! X3.5

© 2001-2024 Discuz! Team.

快速回复 返回顶部 返回列表