Python使用pyqt5写一个串口工具

'''
this module contains the main tool class and the main function to run the tool.
main purpose of this module is to implement uart com send and receive functions via pqt5.
copyright (c) 2026 by oushaojun
'''


from pyqt5.qtcore import *
from pyqt5.qtgui import *
from pyqt5.qtwidgets import *
import os
import sys
import serial
import serial.tools.list_ports
import time
from datetime import datetime


def _log(message):
    # timestamp with milliseconds
    timestamp = datetime.now().strftime("%y-%m-%d %h:%m:%s.%f")[:-3]
    print(f"[{timestamp}] {message}")

class settingsdialog(qdialog):
    reopen_serial_port_signal = pyqtsignal()

    def __init__(self, parent=none):
        super().__init__(parent)
        self.setwindowtitle("settings")
        self.resize(400, 200)
        self.init_ui()
        self.setwindowmodality(qt.applicationmodal)

    def init_ui(self):
        layout = qvboxlayout()
        self.setlayout(layout)

        # create a form layout for the settings
        form_layout = qformlayout()
        layout.addlayout(form_layout)

        # add a line edit for the com port
        self.port_combo = qcombobox()
        ports = serial.tools.list_ports.comports()
        self.port_combo.additems([port.device for port in ports])
        form_layout.addrow("com port:", self.port_combo)

        # add a combo box for the baud rate
        self.baudrate_combo = qcombobox()
        self.baudrate_combo.additems(["9600", "19200", "38400", "57600", "115200"])
        form_layout.addrow("baudrate:", self.baudrate_combo)

        # add ok and cancel buttons
        button_box = qdialogbuttonbox(qdialogbuttonbox.ok | qdialogbuttonbox.cancel)
        button_box.accepted.connect(self.accept)
        button_box.rejected.connect(self.reject)
        layout.addwidget(button_box)

    def accept(self):
        # save the settings when the dialog is accepted
        self.parent().com_config["port"] = self.port_combo.currenttext()
        self.parent().com_config["baudrate"] = int(self.baudrate_combo.currenttext())
        self.parent().save_settings()
        self.reopen_serial_port_signal.emit()
        _log(f"settings updated: {self.parent().com_config}")
        super().accept()

    def reject(self):
        # discard changes when the dialog is rejected
        _log("settings changes discarded.")
        super().reject()


class serialreadworker(qobject):
    message_received = pyqtsignal(str)

    def __init__(self, serial_port):
        super().__init__()
        self.serial_port = serial_port
        self.running = true

    @pyqtslot()
    def run(self):
        while self.running:
            try:
                if self.serial_port and self.serial_port.in_waiting > 0:
                    message = self.serial_port.readline().decode(errors='ignore').strip()
                    if message:
                        self.message_received.emit(message)
            except exception:
                # ignore transient serial errors during read
                pass
            time.sleep(0.1)

    @pyqtslot()
    def update_serial_port(self, serial_port):
        self.serial_port = serial_port

    @pyqtslot()
    def stop(self):
        self.running = false


class mainwindow(qmainwindow):
    def __init__(self):
        super().__init__()
        self.init_settings()    #call first to create variable self.com_config
        self.setwindowtitle("uart com tool")
        self.resize(1200, 600)
        self.init_ui()
        self.init_menu()
        self.open_serial_port()

    def open_serial_port(self):
        try:
            if hasattr(self, 'serial_port') and self.serial_port.is_open:
                self.serial_port.close()
                _log(f"closed serial port: {self.com_config['port']}")
            
            self.serial_port = serial.serial(self.com_config["port"], self.com_config["baudrate"], timeout=1)
            _log(f"opened serial port: {self.com_config['port']} at {self.com_config['baudrate']} baud.")
        except serial.serialexception as e:
            _log(f"error opening serial port: {e}")
            self.output_text_edit.append(f"error opening serial port: {e}")
        else:
            # start a thread to read from the serial port
            if not hasattr(self, 'read_thread') or not self.read_thread.isrunning():
                self.read_thread = qthread()
                self.read_worker = serialreadworker(self.serial_port)
                self.read_worker.movetothread(self.read_thread)
                self.read_thread.started.connect(self.read_worker.run)
                self.read_thread.finished.connect(self.read_worker.stop)
                self.read_worker.message_received.connect(self.handle_received_message)
                self.read_thread.start()
                _log("started serial read thread.")
            else:
                if hasattr(self, 'read_worker') and self.read_worker is not none:
                    self.read_worker.update_serial_port(self.serial_port)
                _log("serial read thread is already running.")
            self.output_text_edit.append(f"\r\nopened serial port: {self.com_config['port']} at {self.com_config['baudrate']} baudrate.")

    def handle_received_message(self, message):
        _log(f"received message: {message}")
        timestamp = datetime.now().strftime("%y-%m-%d %h:%m:%s.%f")[:-3]
        self.output_text_edit.append(f"[{timestamp}] received: {message}")

    def init_settings(self):
        # load settings from qsettings
        self.settings = qsettings("config.ini", qsettings.iniformat)
        self.load_settings()

    def init_ui(self):
        # create a central widget
        central_widget = qwidget()
        self.setcentralwidget(central_widget)

        # create a layout for the central widget
        layout = qvboxlayout()
        central_widget.setlayout(layout)

        # add a label to the layout
        label = qlabel("output message:")
        label.setalignment(qt.alignleft)
        layout.addwidget(label)

        #add a text edit to the layout to show the output message
        self.output_text_edit = qtextedit()
        self.output_text_edit.setreadonly(true)
        layout.addwidget(self.output_text_edit)

        #add a label to the layout
        label2 = qlabel("input message:")
        label2.setalignment(qt.alignleft)
        layout.addwidget(label2)

        #add a text edit to the layout to input the message to send
        self.input_text_edit = qlineedit()
        layout.addwidget(self.input_text_edit)

        #add a button to the layout to send the message
        hbox = qhboxlayout()
        hbox.addstretch()
        self.send_button = qpushbutton("send")
        self.send_button.setfixedheight(30)
        self.send_button.setfixedwidth(100)
        self.send_button.clicked.connect(self.send_message)
        hbox.addwidget(self.send_button)
        layout.addlayout(hbox)

        #status bar to show the current com port and baudrate
        self.status_label = qlabel()
        self.status_label.setalignment(qt.alignright | qt.alignvcenter)
        self.status_bar = qstatusbar()
        self.status_bar.addpermanentwidget(self.status_label, 1)
        self.status_label.settext(
            f"com port: {self.com_config['port']}, baudrate: {self.com_config['baudrate']}"
        )
        self.setstatusbar(self.status_bar)

    def send_message(self):
        # get the message from the input text edit
        message = self.input_text_edit.text()
        timestamp = datetime.now().strftime("%y-%m-%d %h:%m:%s.%f")[:-3]
        if not message:
            _log("no message to send.")
            return

        # send the message via the configured com port and baudrate
        try:
            if not hasattr(self, 'serial_port') or not self.serial_port.is_open:
                _log("serial port is not open. attempting to open it.")
                self.open_serial_port()
                if not self.serial_port.is_open:
                    _log("failed to open serial port. cannot send message.")
                    self.output_text_edit.append(f"[{timestamp}] error: serial port is not open.")
                    return

            self.serial_port.write((message + "\n").encode())
            _log(f"sent message: {message}")
            self.output_text_edit.append(f"[{timestamp}] sent: {message}")
        except serial.serialexception as e:
            _log(f"error sending message: {e}")
            self.output_text_edit.append(f"[{timestamp}] error: {e}")
        else:
            # clear the input text edit after sending
            #self.input_text_edit.clear()
            pass

    def reopen_serial_port(self):
        _log("reopening serial port with updated settings.")
        self.open_serial_port()
        self.update_status_bar()

    def update_status_bar(self):
        self.status_label.settext(
            f"com port: {self.com_config['port']}, baudrate: {self.com_config['baudrate']}"
        )
        _log(f"status bar updated: com port: {self.com_config['port']}, baudrate: {self.com_config['baudrate']}")

    def init_menu(self):
        # create a menu bar
        menu_bar = self.menubar()

        # menu bar contains two sub-menus: "file" and "config"
        file_menu = menu_bar.addmenu("file")
        config_menu = menu_bar.addmenu("settings")

        # exit action to close the application
        exit_action = qaction("exit", self)
        exit_action.triggered.connect(self.close)

        # info action to show the tool information
        info_action = qaction("info", self)
        info_action.triggered.connect(self.show_info)

        # save output message to a file action
        save_action = qaction("save output", self)
        save_action.triggered.connect(self.save_output_to_file)

        #clear output message action
        clear_action = qaction("clear output", self)
        clear_action.triggered.connect(self.output_text_edit.clear)

        # add the exit action to the file menu
        file_menu.addaction(exit_action)
        file_menu.addaction(info_action)
        file_menu.addaction(save_action)
        file_menu.addaction(clear_action)
        file_menu.addseparator()

        # add the settings action to the config menu
        config_menu.addaction("settings")
        config_menu.triggered.connect(self.open_settings_dialog)

    def save_output_to_file(self):
        # open a file dialog to select the file to save the output message
        timestamp = datetime.now().strftime("%y-%m-%d %h:%m:%s.%f")[:-3]
        filename = f"{timestamp.replace(' ', '_').replace(':', '-')}_{self.com_config['port']}.txt"
        options = qfiledialog.options()
        options |= qfiledialog.dontusenativedialog
        file_name, _ = qfiledialog.getsavefilename(self, "save output message", filename, "text files (*.txt);;all files (*)", options=options)
        if file_name:
            try:
                with open(file_name, 'w', encoding='utf-8') as f:
                    f.write(self.output_text_edit.toplaintext())
                _log(f"output message saved to {file_name}")
            except exception as e:
                _log(f"error saving output message: {e}")
                self.output_text_edit.append(f"[{timestamp}] error saving output message: {e}")

    def load_settings(self):

        ports = serial.tools.list_ports.comports()
        _log(f"available com ports: {[port.device for port in ports]}")
        if os.name == 'nt':
            # windows: use the first available com port or default to com1
            default_port = ports[0].device if ports else "com1"
        elif os.name == 'posix':
            # linux: use the first available /dev/ttyusb* or /dev/ttys* port or default to /dev/ttyusb0
            default_port = next((port.device for port in ports if port.device.startswith("/dev/ttyusb") or port.device.startswith("/dev/ttys")), "/dev/ttyusb0")
        else:
            # other os: default to com1
            default_port = "com1"

        self.com_config = {
            "port": self.settings.value("port", default_port),
            "baudrate": int(self.settings.value("baudrate", 115200)),
        }
        _log(f"loaded settings: {self.com_config}")

    def save_settings(self):
        self.settings.setvalue("port", self.com_config["port"])
        self.settings.setvalue("baudrate", self.com_config["baudrate"])
        self.settings.sync()
        _log(f"saved settings: {self.com_config}")

    def open_settings_dialog(self):
        dialog = settingsdialog(self)
        dialog.port_combo.setcurrenttext(self.com_config["port"])
        dialog.baudrate_combo.setcurrenttext(str(self.com_config["baudrate"]))
        dialog.reopen_serial_port_signal.connect(self.reopen_serial_port)
        dialog.exec_()
        self.update_status_bar()

    def closeevent(self, event):
        # save settings when the application is closed
        self.save_settings()
        if hasattr(self, 'read_thread') and self.read_thread is not none and self.read_thread.isrunning():
            try:
                # ask the worker to stop and quit the thread
                if hasattr(self, 'read_worker') and self.read_worker is not none:
                    self.read_worker.stop()
                self.read_thread.quit()
                self.read_thread.wait()
            except exception:
                pass
        _log("application closed. settings saved.")
        event.accept()

    def show_info(self):
        # show a message box with the tool information
        msg = qmessagebox()
        msg.settext("uart com tool\nversion 1.0\ncopyright (c) 2026 by oushaojun")
        msg.setinformativetext("this tool is used for sending and receiving data via uart com ports.")
        msg.setdetailedtext("this tool is implemented using pyqt5 and provides a simple gui for uart communication. it allows users to configure com port settings, send data, and receive data in real-time.")
        msg.setstandardbuttons(qmessagebox.ok | qmessagebox.cancel)
        msg.setwindowtitle("info")
        msg.seticon(qmessagebox.information)
        msg.resize(600, 400)
        retval = msg.exec_()



if __name__ == "__main__":
    app = qapplication(sys.argv)
    app.setwindowicon(qapplication.style().standardicon(qstyle.sp_computericon))
    win = mainwindow()
    win.show()
    sys.exit(app.exec_())

pyqt5==5.15.11
pyqt5_sip==12.18.0
pyserial==3.5