用PyQt做一个网络调试助手 (Python socket PyQt5 GUI)

muzing

在校本科生，车辆工程专业-智能网联汽车方向

NetAssist_PyQt 项目已开源分享至 GitHub ，如果这个项目和这篇博客对你有帮助的话，希望你能给我的GitHub仓库一颗小星星✨

0.序

寒假学习了计算机网络方面的知识，把之前稍有了解的socket编程进一步学习，加之从夏天学到冬天一直在学一直没学完的PyQt5终于学到70%入门了，于是萌生了给自己做一个好看又好用的网络调试助手小工具的想法，把socket编程、面向对象编程、PyQt编程、逻辑与界面分离、git多分支等新知识运用在实践中。也便于未来写自己的应用层程序时调试。

1.基本设计与项目结构，逻辑界面分离

实现一个“网络调试助手”程序，要求可以作为TCP服务端、TCP客户端、UDP服务端、UDP客户端接收发送信息，还具有重复发送、16进制发送接收、保存接收信息到txt文件等功能。

尽可能实现逻辑与界面分离： 网络功能的逻辑 、 界面功能的逻辑 、 纯UI 代码分离。

即，网络模块只需略微修改一两行事件机制的代码即可移植到其他任何程序、在界面功能逻辑增加功能不会对其他部分造成影响、通过QtDesigner修改UI布局不会对其他部分产生影响。

UI控件与布局

纯UI界面由Qt Designer设计生成 MainWindowUI.ui 文件后用pyuic5转换为Python代码（ MainWindowUI.py ），只负责控件的显示与布局；

Qt Designer工具可以可视化实时编辑控件与布局，也可以实现比较细致的调整

界面逻辑

界面逻辑由 MainWindowLogic.py 实现，包括用户输入的检查、计数器的实现、重复发送、16进制发送、保存数据到txt等等。

例如，当用户点击“连接网络”按钮，先由这部分代码对用户输入的IP地址端口号等进行获取、检查，再结合协议类型判断连接类型，最后把确认无误的连接信息发送到网络部分进行真正的连接。这样就简化了网络部分的代码。

也有部分高级控件的功能是通过对Qt原生控件的重写实现的，保存在 UI.MyWidgets.py 中，方便其他项目复用。比如带有IP地址输入验证功能的LineEdit、复位计数按钮是一个可以点击的Label

网络功能逻辑

在Network包的三个模块下实现网络连接功能。Tcp.py包括TCP服务端、TCP客户端的连接建立、发送数据、断开连接等；Udp.py除了UDP服务端客户端，还有一个获得本机IP地址的函数 get_host_ip ；WebServer实现了一个简易的Web服务器。

# Network.__init__.py from Network.Udp import get_host_ip from Network.Tcp import TcpLogic from Network.Udp import UdpLogic from Network.WebServer import WebLogic class NetworkLogic(TcpLogic, UdpLogic, WebLogic): pass

网络模块只有信息反馈（事件处理）使用了PyQt5中的信号 pyqtSignal ，也就是说，如果用其他GUI甚至Flask实现了界面， 只需要改动几行代码即可把Network全部功能完美移植过去 。

界面与功能连接

在 main.py 中进行界面与网络功能的连接。通过类的多继承获得具有完整逻辑功能的界面和网络功能，再通过信号与槽的连接实现界面与网络功能的连接。

class MainWindow(WidgetLogic, NetworkLogic): # 使用多继承，获得具有逻辑功能的界面WidgetLogic和NetworkLogic的网络功能 def __init__(self, parent=None): super().__init__(parent) # 进行了许多界面逻辑信号与网络逻辑功能槽函数的连接 self.link_signal.connect(self.link_signal_handler) self.disconnect_signal.connect(self.disconnect_signal_handler) self.send_signal.connect(self.send_signal_handler) self.tcp_signal_write_msg.connect(self.msg_write) self.tcp_signal_write_info.connect(self.info_write) self.udp_signal_write_msg.connect(self.msg_write) self.udp_signal_write_info.connect(self.info_write) self.signal_write_msg.connect(self.msg_write)

2.代码解读

对部分我认为很好玩的代码做个简单说明

获取本机IP地址

最初的设想是在Ubuntu上用ifconfig 加一些管道来截取仅含本机IPv4地址的字符串，在Windows用ipconfig如法炮制。经过一番努力，完美的失败了。换一个思路，打开搜索引擎搜索“Python 获取本机IP地址”，于是我得到了下面这段精巧的代码

# Network.UdpLogic.py import socket def get_host_ip() -> str: """获取本机IP地址""" try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(('8.8.8.8', 80)) ip = s.getsockname()[0] finally: s.close() return ip

如果直接用 socket.gethostbyname(socket.gethostname()) 获取地址，很有可能是错误的（Vmware虚拟机的地址、127.0.0.1等）

通过UDP尝试连接'8.8.8.8:80'，不管是否连接成功，获得的本机IP一定是正确的。在Ubuntu和Windows上都可用，还省去了判断操作系统的大段代码。

TCP连接中的shutdown

Python官方文档中socket.close()方法下面还有一个小Note

Note
close() releases the resource associated with a connection but does not necessarily close the connection immediately. If you want to close the connection in a timely fashion, call shutdown() before close() .
注解
close() 释放与连接相关联的资源，但不一定立即关闭连接。如果需要及时关闭连接，请在调用 close() 之前调用 shutdown() 。

在close()前显式调用shutdown()方法，以实现立即关闭连接，这可以解决我之前遇到的问题：明明TCP客户端已经关闭，但服务端仍尝试与其发送消息

下面是文档中shutdown方法的部分：

socket. shutdown ( how )
Shut down one or both halves of the connection. If how is SHUT_RD , further receives are disallowed. If how is SHUT_WR , further sends are disallowed. If how is SHUT_RDWR , further sends and receives are disallowed.
socket. shutdown ( how )
关闭一半或全部的连接。如果 how 为 SHUT_RD ，则后续不再允许接收。如果 how 为 SHUT_WR ，则后续不再允许发送。如果 how 为 SHUT_RDWR ，则后续的发送和接收都不允许。

所以在我的代码中，在socket.close()之前加上一行socket.shutdown(socket.SHUT_RDWR) 即可

# Network.Tcp.py class TcpLogic: def tcp_close(self) -> None: """功能函数，关闭网络连接的方法""" if self.link_flag == self.ServerTCP: for client, address in self.client_socket_list: client.shutdown(socket.SHUT_RDWR) # 显式调用shutdown方法 client.close() self.client_socket_list = list() self.tcp_socket.close() msg = '已断开网络\n' self.tcp_signal_write_msg.emit(msg) elif self.link_flag == self.ClientTCP: self.tcp_socket.shutdown(socket.SHUT_RDWR) # 显式调用shutdown方法 self.tcp_socket.close() msg = '已断开网络\n' self.tcp_signal_write_msg.emit(msg)

强制关闭线程的代码

# Network.StopThreading.py import ctypes import inspect # 强制关闭线程的方法 def _async_raise(tid, exc_type): tid = ctypes.c_long(tid) if not inspect.isclass(exc_type): exc_type = type(exc_type) res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exc_type)) if res == 0: raise ValueError("invalid thread id") elif res != 1: ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None) raise SystemError("PyThreadState_SetAsyncExc failed") def stop_thread(thread): _async_raise(thread.ident, SystemExit)

UI控件布局Form类的继承

通过Qt Designer生成的MainWindowUI.py中只有一个 Ui_Form 类，下有 setupUi 、 retranslateUi 两个方法，前者记录了所有控件布局信息，后者记录界面上的所有文字内容（方便实现中文英语等多语言翻译切换）。

# UI.MainWindowUI.py # Created by: PyQt5 UI code generator 5.15.2 class Ui_Form(object): def setupUi(self, Form): # Ui_Form类本身没有Widget控件，需要在调用setupUi方法时传入需要被Ui_Form类布局的窗口Form Form.setObjectName("Form") Form.resize(700, 570) Form.setMinimumSize(QtCore.QSize(600, 500)) # ...... def retranslateUi(self, Form): # 所有界面上的文字都是在这个方法中设置的，这是为了方便软件实现国际化多语言 _translate = QtCore.QCoreApplication.translate Form.setWindowTitle(_translate("Form", "网络调试助手")) self.ProtocolTypeLabel.setText(_translate("Form", "协议类型")) self.ProtocolTypeComboBox.setItemText(0, _translate("Form", "TCP")) # ......

Ui_Form类并无QWidget窗口，需要在MainWindowLogic.py中创建一个继承自QWidget的QmyWidget类，把这个类实例传入Ui_Form.setupUi方法中。

# MainWindowLogic.py from UI import MainWindowUI class WidgetLogic(QWidget): def __init__(self, parent=None): super().__init__(parent) # 调用父类构造函数，创建QWidget窗体 self.__ui = MainWindowUI.Ui_Form() # 把Ui_Form设置为QmyWidget的私有属性 self.__ui.setupUi(self) # 调用setupUi()函数创建UI窗体 self.__ui.retranslateUi(self) # 设置文字内容

显示创建MainWindowUI类的私有属性 self.__ui ，包含了可视化设计的窗体上的所有组件。只有通过 self.__ui 才能访问窗体上的组件，外部无法访问， 更符合面向对象封装隔离的设计思想 。

IP地址输入框的验证器

为IP地址输入框专门写了 IPv4AddrLineEdit 类，使得用户在输入IP地址时，只有键盘输入正确的格式才能真正输入，比如按下键盘上字母键是没有效果的。同时方便起见，把中文输入法输入的句号 。 也自动转化成英文输入法的句点 .

（可以参考前面的博文 PyQt5 输入验证器-正则方式 )

# UI.MyWidgets.py class IPv4AddrLineEdit(QLineEdit): 带有验证输入IPv4地址功能的LineEdit class IPValidator(QRegExpValidator): def validate(self, inputs: str, pos: int) -> [QValidator.State, str, int]: # 重写validate方法以实现可以自动把中文句号转化为英文句点的功能 inputs = inputs.replace('。', '.') return super().validate(inputs, pos) # 一串神秘的正则表达式，据说可以验证IPv4类型的地址 reg_ex = QRegExp("((2[0-4]\\d|25[0-5]|[01]?\\d\\d?)\\.){3}(2[0-4]\\d|25[0-5]|[01]?\\d\\d?)") def __init__(self, parent=None): super().__init__(parent) ip_input_validator = self.IPValidator(self.reg_ex, parent) # 实例化一个验证器对象 self.setValidator(ip_input_validator) # 为LineEdit设置验证器

类似的，也为端口号的输入设置了验证器

# UI.MyWidgets.py class PortLineEdit(QLineEdit): 带有验证器的端口号输入LineEdit class PortValidator(QIntValidator): # 重写整数型验证器来实现更精确的控制 def fixup(self, inputs: str) -> str: if len(inputs) == 0: return '' # 防止输入框为空时报错 elif int(inputs) > 65535: return '7777' # 如果用户输入的内容无效，则焦点离开后内容自动变成7777 return inputs def __init__(self, parent=None): super().__init__(parent) validator = self.PortValidator(0, 65535, parent) # 确保端口号为int整数、范围合理 self.setValidator(validator)

然后在Qt Designer中把对应的控件进行提升即可

连接状态的标识

通过 self.link_flag 属性保存当前连接状态。界面逻辑的类 WidgetLogic 和网络功能的类 NetworkLogic 中都有这个属性：前者根据用户操作变化其值，后者根据其值实现对应网络功能。

~ self.link_flag 的值主要由 WidgetLogic 下的方法来设置：

# MainWindowLogic.py class WidgetLogic(QWidget): def __init__(self, parent=None): # ...... self.link_flag = self.NoLink # 初始化连接状态为未连接 self.protocol_type = 'TCP' # ...... def click_link_handler(self): """连接按钮连接时的槽函数""" # 一些获取用户输入的代码，在此省略 # ...... if self.protocol_type == "TCP" and server_flag: self.link_flag = self.ServerTCP # 把连接状态置为TCP服务端 elif self.protocol_type == "TCP" and not server_flag: self.link_flag = self.ClientTCP # 把连接状态置为TCP客户端 elif self.protocol_type == "UDP" and server_flag: self.link_flag = self.ServerUDP # 把连接状态置为UDP服务端 elif self.protocol_type == "UDP" and not server_flag: self.link_flag = self.ClientUDP # 把连接状态置为UDP客户端 elif self.protocol_type == "Web Server" and server_flag and self.dir: self.link_flag = self.WebServer # 连接状态置为WebServer def click_disconnect(self): 实现断开连接的功能函数 # ...... self.link_flag = self.NoLink # 断开连接后把连接状态重置为未连接 # 把int类型的标识位保存在类属性中，用self.NoLink替换-1，增强代码可读性 NoLink = -1 ServerTCP = 0 ClientTCP = 1 ServerUDP = 2 ClientUDP = 3 WebServer = 4

有了连接状态标识，就可以把目前的连接状态作为许多操作的判断依据，比如：

# Network.Tcp.py class TcpLogic: def __init__(self): # ...... self.link_flag = self.NoLink # 用于标记是否开启了连接 def tcp_send(self, send_msg): """功能函数，用于TCP服务端和TCP客户端发送消息""" # ...... if self.link_flag == self.ServerTCP: # 向所有连接的客户端发送消息 if self.client_socket_list: for client, address in self.client_socket_list: client.send(send_info_encoded) msg = 'TCP服务端已发送' self.tcp_signal_write_msg.emit(msg) self.tcp_signal_write_info.emit(send_info, self.InfoSend) if self.link_flag == self.ClientTCP: self.tcp_socket.send(send_info_encoded) msg = 'TCP客户端已发送' self.tcp_signal_write_msg.emit(msg) self.tcp_signal_write_info.emit(send_info, self.InfoSend) def tcp_close(self): """功能函数，关闭网络连接的方法""" if self.link_flag == self.ServerTCP: # 断开TCP服务端连接的代码 for client, address in self.client_socket_list: # 先关闭所有已连接的客户端 client.shutdown(2) client.close() self.client_socket_list = list() # 把已连接的客户端列表重新置为空列表 # 再关闭服务端 self.tcp_socket.close() msg = '已断开网络\n' self.tcp_signal_write_msg.emit(msg) # ...停止线程的代码... elif self.link_flag == self.ClientTCP: # 断开TCP客户端连接的代码 self.tcp_socket.shutdown(2) self.tcp_socket.close() msg = '已断开网络\n' self.tcp_signal_write_msg.emit(msg) # ...停止线程的代码... NoLink = -1 ServerTCP = 0 ClientTCP = 1

通过 self.link_flag 实现了分用，不管Server还是Client，发送消息断开连接时调用的函数都是同一个。同理， main.py 中，通过标识实现断开连接分用。

# main.py class MainWindow(WidgetLogic, NetworkLogic): def disconnect_signal_handler(self): """断开连接的槽函数""" if self.link_flag == self.ServerTCP or self.link_flag == self.ClientTCP: self.tcp_close() elif self.link_flag == self.ServerUDP or self.link_flag == self.ClientUDP: self.udp_close() elif self.link_flag == self.WebServer: self.web_close()

该值除了在网络部分有应用，也用在一些界面逻辑的控制上，比如：

# MainWindowLogic.py class WidgetLogic(QWidget): def open_file_handler(self): """打开文件按钮的槽函数""" if self.link_flag in [self.ServerTCP, self.ClientTCP, self.ClientUDP]: # 如果连接状态为TCP服务端/客户端、UDP客户端，则“打开文件”按钮功能为打开文本文件 # ...... 打开文本文件加载到发送输入框的代码 ...... elif self.link_flag == self.NoLink and self.protocol_type == 'Web Server': # 如果连接状态为WebServer，则按钮功能为选择工作目录 self.dir = QFileDialog.getExistingDirectory(self, "选择index.html所在路径", './') self.__ui.SendPlainTextEdit.clear() self.__ui.SendPlainTextEdit.appendPlainText(str(self.dir)) self.__ui.SendPlainTextEdit.setEnabled(False) # 如果连接状态为未连接或UDP服务端等，则按钮无作用

用户输入检查

我的软件思路是，如果只输入本机端口号，则作为Server启动，绑定这个端口；如果只输入目标IP和目标端口，则作为Client启动，向该IP端口发送数据。所以必须对用户的异常输入（如只输入目标端口）进行处理：

未输入任何信息

# MainWindowLogic.py def click_link_handler(self): """连接按钮连接时的槽函数""" if my_port == -1 and target_port == -1 and target_ip == '': mb = QMessageBox(QMessageBox.Critical, '错误', '请输入信息', QMessageBox.Ok, self) mb.open() self.editable(True) # 恢复可编辑状态 self.__ui.ConnectButton.setChecked(False) # 恢复连接按钮状态 # 提前终止槽函数 return None

仅输入目标IP

elif target_port == -1 and target_ip != '': input_d = PortInputDialog(self) # 在UI.MyWidgets中定义，具有端口号检查功能 input_d.setWindowTitle("服务启动失败") input_d.setLabelText("请输入目标端口号作为Client启动，或取消") input_d.intValueSelected.connect(lambda val: self.__ui.TargetPortLineEdit.setText(str(val))) input_d.open() self.__ui.ConnectButton.setChecked(False) # 提前终止槽函数 return None

仅输入目标端口

elif target_port != -1 and target_ip == '': mb = QMessageBox(QMessageBox.Critical, 'Client启动错误', '请输入目标IP地址', QMessageBox.Ok, self) mb.open() self.__ui.ConnectButton.setChecked(False) # 提前终止槽函数 return None

同时输入了本机端口、目标IP、目标端口

WebServer未选择工作目录

如果连接之前没有使用“选择路径”按钮选择工作目录，会在按下“连接网络”按钮时弹出文件夹选择对话框

def click_link_handler(self): """连接按钮连接时的槽函数""" if self.protocol_type == "Web Server" and not self.dir: # 处理用户未选择工作路径情况下连接网络 self.dir = QFileDialog.getExistingDirectory(self, "选择index.html所在路径", './') if self.dir: self.__ui.SendPlainTextEdit.clear()