【米尔瑞萨RZ/G2L开发板-创新应用】串口服务器实现

继续在上一篇帖子的基础上实现完整的串口服务器【米尔瑞萨RZ/G2L开发板-创新应用】基于python的tcp、udp、串口收发使用 - 米尔RZ/G2L开发板 - 米尔科技论坛 - Powered by Discuz! (myir-tech.com)


本次需要实现的功能：
1、实现收到udp广播后，发送串口服务器配置信息；
2、收到tcp客户端发送来的数据后，自动转发到串口；
3、收到串口发送来的数据后，自动转发到tcp客户端；

功能实现：
1、收到UDP信息后，对信息内容进行过滤，如果结果匹配，则直接返回到发送端，代码如下：

1. # -*- coding: utf-8 -*-
   
2. 
   
3. import socket  # 导入socket模块
   
4. import time  # 导入time模块
   
5. import threading
   
6. from config import *
   
7. 
   
8. 
   
9. class udp(threading.Thread):
   
10.     def __init__(self, ip: str = '', port: int = 9000):
    
11.         threading.Thread.__init__(self)
    
12.         self.ip = ip
    
13.         self.port = port
    
14. 
    
15.     def run(self):
    
16.         cfg = config()
    
17.         cfg.read()
    
18.         # 创建一个套接字socket对象，用于进行通讯
    
19.         # socket.AF_INET 指明使用INET地址集，进行网间通讯
    
20.         # socket.SOCK_DGRAM 指明使用数据协议，即使用传输层的udp协议
    
21.         server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    
22.         # 接收来自任何IP的广播帧
    
23.         address = (self.ip, self.port)
    
24.         server_socket.bind(address)  # 为服务器绑定一个固定的地址，ip和端口
    
25.         # server_socket.settimeout(10)  # 设置一个时间提示，如果10秒钟没接到数据进行提示
    
26. 
    
27.         while True:
    
28.             # 正常情况下接收数据并且显示，如果10秒钟没有接收数据进行提示（打印 'time out'）
    
29.             # 当然可以不要这个提示，那样的话把'try:' 以及 'except'后的语句删掉就可以了
    
30.             try:
    
31.                 # now = time.time()  # 获取当前时间
    
32.                 # 接收客户端传来的数据 recvfrom接收客户端的数据，默认是阻塞的，直到有客户端传来数据
    
33.                 # recvfrom 参数的意义，表示最大能接收多少数据，单位是字节
    
34.                 # recvfrom返回值说明
    
35.                 # receive_data表示接受到的传来的数据,是bytes类型
    
36.                 # client  表示传来数据的客户端的身份信息，客户端的ip和端口，元组
    
37.                 receive_data, client = server_socket.recvfrom(1024)
    
38.                 # print(time.strftime('%Y-%m-%d %H:%M:%S',
    
39.                 #                     time.localtime(now)))  # 以指定格式显示时间
    
40.                 # print('来自客户端%s,发送的%s\n' % (client, receive_data))  # 打印接收的内容
    
41.                 recv_str = str(receive_data, encoding="utf-8")
    
42.                 if recv_str == 'where is myir serial port server?':
    
43.                     ip_addr = cfg.ipv4
    
44.                     serial_conf = ''
    
45.                     for x in range(cfg.serial_count):
    
46.                         serial_conf += cfg.tcp_ports[x]+','
    
47.                         serial_conf += cfg.port[x]+','
    
48.                         serial_conf += cfg.baudrate[x]+','
    
49.                         serial_conf += cfg.parity[x]+';'
    
50.                     server_socket.sendto(
    
51.                         bytes(ip_addr+';'+serial_conf, encoding="utf8"), client)
    
52.             except socket.timeout:
    
53.                 print('time out')
    

复制代码

收到客户端发送的 where is myir serial port server? 后，发送串口及tcp服务器配置信息。
2、收到tcp客户端发送的信息后，自动转发到串口队列
注意：此处难点，需要将socket设置为非阻塞模式，否则无法转发串口收到的数据

1. # -*- coding: utf-8 -*-
   
2. 
   
3. import socket  # 导入socket模块
   
4. import threading
   
5. from queue import Queue
   
6. import selectors
   
7. import time
   
8. 
   
9. 
   
10. class tcp(threading.Thread):
    
11.     def __init__(self, tcp2sci: Queue, sci2tcp: Queue, ip: str = '', port: int = 9000):
    
12.         threading.Thread.__init__(self)
    
13.         self.ip = ip
    
14.         self.port = port
    
15.         self.tcp2sci = tcp2sci
    
16.         self.sci2tcp = sci2tcp
    
17.         # 创建 Selector 对象
    
18.         self.sel = selectors.DefaultSelector()
    
19.         self.clientAddr = 0
    
20.         self.tcp_server_socket = 0
    
21. 
    
22.     def run(self):
    
23.         self.tcp_server_socket = socket.socket(
    
24.             socket.AF_INET, socket.SOCK_STREAM)
    
25.         address = (self.ip, self.port)
    
26.         self.tcp_server_socket.bind(address)
    
27.         self.tcp_server_socket.listen(128)
    
28.         self.tcp_server_socket.setblocking(False)
    
29.         # 将服务器套接字注册到 Selector 上
    
30.         self.sel.register(self.tcp_server_socket,
    
31.                           selectors.EVENT_READ, self.accept)
    
32. 
    
33.         while True:
    
34.             events = self.sel.select(timeout=0.001)
    
35.             for key, mask in events:
    
36.                 callback = key.data
    
37.                 callback(key.fileobj, mask)
    
38.             while not self.sci2tcp.empty():
    
39.                 item = self.sci2tcp.get()
    
40.                 if self.clientAddr != 0:
    
41.                     self.tcp_server_socket.sendto(item, self.clientAddr)
    
42.             time.sleep(0.001)
    
43. 
    
44.         tcp_server_socket.close()
    
45. 
    
46.     # 定义回调函数
    
47.     def accept(self, sock, mask):
    
48.         self.tcp_server_socket, self.clientAddr = sock.accept()
    
49.         self.tcp_server_socket.setblocking(False)
    
50.         self.sel.register(self.tcp_server_socket,
    
51.                           selectors.EVENT_READ, self.read)
    
52. 
    
53.     def read(self, sock, mask):
    
54.         data = self.tcp_server_socket.recv(1024)
    
55.         if data:
    
56.             self.tcp2sci.put(data)
    
57.         else:
    
58.             self.sel.unregister(self.tcp_server_socket)
    
59.             self.tcp_server_socket.close()
    
60.             self.clientAddr = 0
    

复制代码

3、收到串口发送来的数据后，转发到tcp客户端

1. import serial
   
2. import os
   
3. import socket  # 导入socket模块
   
4. 
   
5. 
   
6. class config():
   
7.     global cfg
   
8.     def __init__(self):
   
9.         self.serial_count = 1
   
10.         self.path = './configurate/port_config'
    
11.         self.device_name = 'serial_port_server'
    
12.         self.ipv4 = self._get_ip_address()
    
13.         self.udp_port = 9000
    
14.         self.tcp_start_port = 5000
    
15.         self.tcp_ports = []
    
16.         self.port = []
    
17.         self.baudrate = []
    
18.         self.parity = []
    
19.         for x in range(self.serial_count):
    
20.             self.tcp_ports.append(x + self.tcp_start_port)
    
21.             self.port.append('/dev/ttySC' + str(x))
    
22.             self.baudrate.append('115200')
    
23.             self.parity.append(serial.PARITY_NONE)
    
24. 
    
25.     def save(self) -> int:
    
26.         list = []
    
27.         for x in range(self.serial_count):
    
28.             list.append(str(self.tcp_ports[x]) + '\n')
    
29.             list.append(self.port[x]+'\n')
    
30.             list.append(self.baudrate[x]+'\n')
    
31.             list.append(self.parity[x]+'\n')
    
32.         if not os.path.exists('./configurate'):
    
33.             os.mkdir('./configurate')
    
34.         if os.path.exists(self.path):
    
35.             os.remove(self.path)
    
36.         with open(self.path, 'w') as file:
    
37.             file.writelines(list)
    
38. 
    
39.     def read(self) -> int:
    
40.         if os.path.exists(self.path):
    
41.             with open(self.path, 'r') as file:
    
42.                 list = file.readlines()
    
43.         index = 0
    
44.         for x in range(self.serial_count):
    
45.             self.tcp_ports[x] = list[index].replace('\n','')
    
46.             index += 1
    
47.             self.port[x] = list[index].replace('\n','')
    
48.             index += 1
    
49.             self.baudrate[x] = list[index].replace('\n','')
    
50.             index += 1
    
51.             self.parity[x] = list[index].replace('\n','')
    
52.             index += 1
    
53.         return 0
    
54.    
    
55.     def _get_ip_address(self) -> str:
    
56.         s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    
57.         s.connect(("8.8.8.8", 80))
    
58.         return str(s.getsockname()[0])
    

复制代码

4、另外需要读取配置文件

1. import serial
   
2. import os
   
3. import socket  # 导入socket模块
   
4. 
   
5. 
   
6. class config():
   
7.     global cfg
   
8.     def __init__(self):
   
9.         self.serial_count = 1
   
10.         self.path = './configurate/port_config'
    
11.         self.device_name = 'serial_port_server'
    
12.         self.ipv4 = self._get_ip_address()
    
13.         self.udp_port = 9000
    
14.         self.tcp_start_port = 5000
    
15.         self.tcp_ports = []
    
16.         self.port = []
    
17.         self.baudrate = []
    
18.         self.parity = []
    
19.         for x in range(self.serial_count):
    
20.             self.tcp_ports.append(x + self.tcp_start_port)
    
21.             self.port.append('/dev/ttySC' + str(x))
    
22.             self.baudrate.append('115200')
    
23.             self.parity.append(serial.PARITY_NONE)
    
24. 
    
25.     def save(self) -> int:
    
26.         list = []
    
27.         for x in range(self.serial_count):
    
28.             list.append(str(self.tcp_ports[x]) + '\n')
    
29.             list.append(self.port[x]+'\n')
    
30.             list.append(self.baudrate[x]+'\n')
    
31.             list.append(self.parity[x]+'\n')
    
32.         if not os.path.exists('./configurate'):
    
33.             os.mkdir('./configurate')
    
34.         if os.path.exists(self.path):
    
35.             os.remove(self.path)
    
36.         with open(self.path, 'w') as file:
    
37.             file.writelines(list)
    
38. 
    
39.     def read(self) -> int:
    
40.         if os.path.exists(self.path):
    
41.             with open(self.path, 'r') as file:
    
42.                 list = file.readlines()
    
43.         index = 0
    
44.         for x in range(self.serial_count):
    
45.             self.tcp_ports[x] = list[index].replace('\n','')
    
46.             index += 1
    
47.             self.port[x] = list[index].replace('\n','')
    
48.             index += 1
    
49.             self.baudrate[x] = list[index].replace('\n','')
    
50.             index += 1
    
51.             self.parity[x] = list[index].replace('\n','')
    
52.             index += 1
    
53.         return 0
    
54.    
    
55.     def _get_ip_address(self) -> str:
    
56.         s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    
57.         s.connect(("8.8.8.8", 80))
    
58.         return str(s.getsockname()[0])
    

复制代码

5、在main中统一对所有线程进行初始化，并构建队列用于串口有tcp服务器之间传递信息

1. # -*- coding: utf-8 -*-
   
2. from udp import *
   
3. from tcp import *
   
4. from sci import *
   
5. from config import *
   
6. from threading import Thread
   
7. from queue import Queue
   
8. 
   
9. cfg = 0
   
10. 
    
11. if __name__ == "__main__":
    
12.     cfg = config()
    
13.     # cfg.save()
    
14.     cfg.read()
    
15.     # 创建 Thread 实例
    
16.     t1 = udp('', cfg.udp_port)
    
17.     t1.start()
    
18. 
    
19.     tcp_thread = []
    
20.     serial_thread = []
    
21.     for x in range(cfg.serial_count):
    
22.         tcp2sci = Queue(maxsize=1000)
    
23.         sci2tcp = Queue(maxsize=1000)
    
24.         tcp_thread.append(
    
25.             tcp(tcp2sci, sci2tcp, ip=cfg.ipv4, port=int(cfg.tcp_ports[x])))
    
26.         # serial_thread.append(
    
27.         #     sci(sci2tcp, tcp2sci, port=cfg.port[x], baudrate=cfg.baudrate[x], parity=cfg.parity[x]))
    
28. 
    
29.     # 启动线程运行
    
30.     for x in range(cfg.serial_count):
    
31.         tcp_thread[x].start()
    
32.         # serial_thread[x].start()
    
33. 
    
34.     # 等待所有线程执行完毕
    
35.     t1.join()  # join() 等待线程终止，要不然一直挂起
    
36.     for x in range(cfg.serial_count):
    
37.         tcp_thread[x].join()
    
38.         # serial_thread[x].join()
    

复制代码

演示视频如下

https://www.bilibili.com/video/BV1KN411n7Jx/