python udp 线程接受 demo

udp使用socketserver 接受简单方便
使用是python 的threading 快速构建udp 接受线程
使用 pyqt5的QThread，用于发现信号到ui线程，跟新ui
使用queue接受udp数据，并通过queue在 udp接受线程和数据解析线程间数据传递。

from socketserver import*
import threading
import time
import sys
from queue import *
from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
from PyQt5 import *
'''udp REC线程'''
class MyUDPServer(UDPServer):
    def __init__(self, server_address, RequestHandlerClass, queue):
        self.udp_queue = queue  # 增加的参数
        UDPServer.__init__(self, server_address, RequestHandlerClass, bind_and_activate=True)


class Handler(BaseRequestHandler):
    def handle(self):
        self.data = self.request[0].strip()
        self.server.udp_queue.put(self.data)
        # print(self.server.udp_queue.get())
        # print(self.data)

'''DATA 解析 线程'''
class mythread(QThread):
    signal = pyqtSignal(bytes)
    def __init__(self,queue,rec_data_Instance):
        super( mythread,self).__init__()
        self.running = True
        self.queue = queue
        self.rec_data_Instance=rec_data_Instance
        if self.rec_data_Instance:
            self.signal.connect(self.rec_data_Instance.rec_data)


    def __del__(self):
        self.running = False
        self.wait()

    def run(self):
        while self.running:
            # if self.udp_queue.
            if self.queue.empty()!= True :
                # print(self.queue.get())
                self.signal.emit(self.queue.get())

    def stop(self):
        self.running = False

'''ui 显示线程'''
class MY_Widget(QWidget):
    def __init__(self):
        super().__init__()
        self.initUI()
        self.button_num = 0  
    def initUI(self):
        self.ip = "10.7.28.51"
        self.port = 9999
        ### 窗体大小
        self.height = 100
        self.width = 200
        self.resize(self.width,self.height)
        ### 设置窗体 标题
        self.setWindowTitle("UDP ")
        v1layout = QVBoxLayout()
        self.open_button = QPushButton("打开")
        self.open_button.clicked.connect(self.open_button_func)
        self.edit= QTextEdit(self)
        self.edit.setText("test")
        v1layout.addWidget(self.open_button)
        v1layout.addWidget(self.edit)
        self.setLayout(v1layout)
        self.show()

    def open_udp_server(self,ip,port):  
            ADDR = ip, port
            #接受队列
            self.udp_queue=Queue(maxsize=5)
            #udp 服务
            self.server = MyUDPServer(ADDR, Handler, self.udp_queue)  
            #创建线程，线程用于UDP接收
            self.server_thread = threading.Thread(target=self.server.serve_forever)  
            #设置后台线程
            self.server_thread.setDaemon(True) 
            self.server_thread.start()   
            #创建数据解析
            self.up_mythread = mythread(self.udp_queue,self)
            self.up_mythread.start()
    def close_udp_server(self):  
            self.up_mythread.stop()
            self.server.shutdown()
            self.server.server_close()
    def print_hex(self,bytes):
        l = [hex(int(i)) for i in bytes]
        str = " ".join(l)
        return str

    def rec_data(self,data):
        print(self.print_hex(data))
        self.edit.setText(self.print_hex(data))

    def open_button_func(self):
        self.button_num = self.button_num+1
        if self.button_num %2 == 1:
            self.open_button.setText("关闭")
            self.open_udp_server(self.ip,self.port)

        else:
            self.open_button.setText("打开")
            self.close_udp_server()



if __name__ == "__main__":  
    app = QApplication(sys.argv)
    #创建ui
    ex = MY_Widget()
    app.exit(app.exec_())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117