[global]
index-url = https://pypi.tuna.tsinghua.edu.cn/simple
[install]
trusted-host = pypi.tuna.tsinghua.edu.cn

pip install pyqt5

pip install pyqt5 -i https://pypi.tuna.tsinghua.edu.cn/simple

pip install pyqt5-tools

pip install pyqt5-tools -i https://pypi.tuna.tsinghua.edu.cn/simple

from PyQt5.QtWidgets import QMainWindow
 
# 導(dǎo)入設(shè)計(jì)的ui界面轉(zhuǎn)換成的py文件
from ui_serial_port import Ui_MainWindow
 
 
class SerialPort(QMainWindow):
    """
     串口行為
    """
 
    def __init__(self):
        # QMainWindow構(gòu)造函數(shù)初始化
        super().__init__()
        self.ui = Ui_MainWindow()
        # 這個(gè)函數(shù)本身需要傳遞一個(gè)MainWindow類，而該類本身就繼承了這個(gè)，所以可以直接傳入self
        self.ui.setupUi(self)

import sys
from serial_port import SerialPort
from PyQt5.QtWidgets import QApplication, QMainWindow
 
if __name__ == '__main__':
    # 先建立一個(gè)app
    app = QApplication(sys.argv)
    # 初始化一個(gè)對(duì)象，調(diào)用init函數(shù)，已加載設(shè)計(jì)的ui文件
    ui = SerialPort()
    # 顯示這個(gè)ui
    ui.show()
    # 運(yùn)行界面，響應(yīng)按鈕等操作
    sys.exit(app.exec_())

def __init_serial_setting__(self):
	"""
	初始化串口設(shè)置相關(guān)控件默認(rèn)參數(shù)
	設(shè)置下拉列表自動(dòng)補(bǔ)全  https://blog.csdn.net/xuleisdjn/article/details/51434118
	:return:
	"""
	# 加載可用串口
	self.ui.cbx_com.setEditable(False)
	self.ui.cbx_com.setMaxVisibleItems(10)  # 設(shè)置最大顯示下列項(xiàng) 超過要使用滾動(dòng)條拖拉
	self.ui.cbx_com.setInsertPolicy(QComboBox.InsertAfterCurrent)  # 設(shè)置插入方式\
	port_list = list(serial.tools.list_ports.comports())  # 獲取當(dāng)前的所有串口，得到一個(gè)列表
	for port in port_list:
		self.ui.cbx_com.addItem(port.device)
 
	# 初始化波特率列表
	self.ui.cbx_baud_rate.setEditable(False)
	self.ui.cbx_baud_rate.setMaxVisibleItems(10)  # 設(shè)置最大顯示下列項(xiàng) 超過要使用滾動(dòng)條拖拉
	self.ui.cbx_baud_rate.setInsertPolicy(QComboBox.InsertAfterCurrent)  # 設(shè)置插入方式
	for baud_rate in [1200, 2400, 4800, 9600, 19200, 384000, 57600, 115200, 460800, 921600, 230400, 1500000]:
		self.ui.cbx_baud_rate.addItem(str(baud_rate), baud_rate)
	# 設(shè)置默認(rèn)值
	self.ui.cbx_baud_rate.setCurrentIndex(7)
 
	# 初始化校驗(yàn)位列表
	self.ui.cbx_parity_bit.setEditable(False)
	self.ui.cbx_parity_bit.setMaxVisibleItems(10)  # 設(shè)置最大顯示下列項(xiàng) 超過要使用滾動(dòng)條拖拉
	self.ui.cbx_parity_bit.setInsertPolicy(QComboBox.InsertAfterCurrent)  # 設(shè)置插入方式
	for (key, value) in {'None': serial.PARITY_NONE, 'Odd': serial.PARITY_ODD,
						 'Even': serial.PARITY_EVEN, 'Mark': serial.PARITY_MARK,
						 'Space': serial.PARITY_SPACE}.items():
		self.ui.cbx_parity_bit.addItem(key, value)
	# 設(shè)置默認(rèn)值
	self.ui.cbx_parity_bit.setCurrentIndex(0)
 
	# 初始化數(shù)據(jù)位列表
	self.ui.cbx_data_bit.setEditable(False)
	self.ui.cbx_data_bit.setMaxVisibleItems(10)  # 設(shè)置最大顯示下列項(xiàng) 超過要使用滾動(dòng)條拖拉
	self.ui.cbx_data_bit.setInsertPolicy(QComboBox.InsertAfterCurrent)  # 設(shè)置插入方式
	for data_bit in [serial.FIVEBITS, serial.SIXBITS, serial.SEVENBITS, serial.EIGHTBITS]:
		self.ui.cbx_data_bit.addItem(str(data_bit), data_bit)
	# 設(shè)置默認(rèn)值
	self.ui.cbx_data_bit.setCurrentIndex(3)
 
	# 初始化停止位列表
	self.ui.cbx_stop_bit.setEditable(False)
	self.ui.cbx_stop_bit.setMaxVisibleItems(10)  # 設(shè)置最大顯示下列項(xiàng) 超過要使用滾動(dòng)條拖拉
	self.ui.cbx_stop_bit.setInsertPolicy(QComboBox.InsertAfterCurrent)  # 設(shè)置插入方式
	for data_bit in [serial.STOPBITS_ONE, serial.STOPBITS_TWO]:
		self.ui.cbx_stop_bit.addItem(str(data_bit), data_bit)
	# 設(shè)置默認(rèn)值
	self.ui.cbx_stop_bit.setCurrentIndex(0)
 
	# 設(shè)置點(diǎn)擊打開串口按鈕對(duì)應(yīng)的槽函數(shù)
	self.ui.btx_start.clicked.connect(self.open_serial_connection)

def __init_recv_setting__(self):
	"""
	接收設(shè)置初始化
	:return:
	"""
	self.ui.rbn_data_format_hex.clicked.connect(self.rbn_data_format_hex_clicked)
	self.ui.rbn_data_format_ascii.clicked.connect(self.rbn_data_format_ascii_clicked)

def rbn_data_format_hex_clicked(self):
	"""
	接收數(shù)據(jù)格式發(fā)生變化
	:return:
	"""
	if self.ui.rbn_data_format_hex.isChecked():
		self.ui.rbn_data_format_ascii.setChecked(False)
		if self.serial_thread:
			self.serial_thread.date_format = 'hex'

def rbn_data_format_ascii_clicked(self):
	"""
	接收數(shù)據(jù)格式發(fā)生變化
	:return:
	"""
	if self.ui.rbn_data_format_ascii.isChecked():
		self.ui.rbn_data_format_hex.setChecked(False)
		if self.serial_thread:
			self.serial_thread.date_format = 'ascii'

def __init_recv_data_viewer__(self):
	"""
	初始化串口數(shù)據(jù)接收區(qū)域
	:return:
	"""
	self.ui.txt_recv_data_viewer.setReadOnly(True)
	self.ui.txt_recv_data_viewer.textChanged.connect(
		lambda: self.ui.txt_recv_data_viewer.moveCursor(QTextCursor.End))

def __init_send_data_viewer__(self):
	"""
	初始化串口數(shù)據(jù)發(fā)送區(qū)域
	:return:
	"""
	self.ui.btn_send.clicked.connect(self.send_serial_data)

def __init_package_setting__(self):
	self.ui.let_nav_header.setReadOnly(True)
	self.ui.let_nav_work_state.setReadOnly(True)
	self.ui.let_nav_param_state.setReadOnly(True)
	self.ui.let_nav_run_time.setReadOnly(True)
	self.ui.let_nav_latitude.setReadOnly(True)
	self.ui.let_nav_longitude.setReadOnly(True)
	self.ui.let_nav_heave.setReadOnly(True)
	self.ui.let_nav_east_speed.setReadOnly(True)
	self.ui.let_nav_north_speed.setReadOnly(True)
	self.ui.let_nav_vertical_speed.setReadOnly(True)
	self.ui.let_nav_attitude_angle1.setReadOnly(True)
	self.ui.let_nav_attitude_angle2.setReadOnly(True)
	self.ui.let_nav_attitude_angle3.setReadOnly(True)
	self.ui.let_nav_attitude_velocity1.setReadOnly(True)
	self.ui.let_nav_attitude_velocity2.setReadOnly(True)
	self.ui.let_nav_attitude_velocity3.setReadOnly(True)
	self.ui.let_nav_fault_code.setReadOnly(True)
	self.ui.let_nav_imu_time.setReadOnly(True)
	self.ui.let_nav_reserved.setReadOnly(True)
	self.ui.let_nav_response_flag.setReadOnly(True)
	self.ui.let_nav_check_sum.setReadOnly(True)
	self.ui.let_raw_header.setReadOnly(True)
	self.ui.let_raw__imu_time.setReadOnly(True)
	self.ui.let_raw_gyrox.setReadOnly(True)
	self.ui.let_raw_gyroy.setReadOnly(True)
	self.ui.let_raw_gyroz.setReadOnly(True)
	self.ui.let_raw_accex.setReadOnly(True)
	self.ui.let_raw_accey.setReadOnly(True)
	self.ui.let_raw_accez.setReadOnly(True)
	self.ui.let_raw_reserved.setReadOnly(True)
	self.ui.let_raw_turntable_angle1.setReadOnly(True)
	self.ui.let_raw_turntable_angle2.setReadOnly(True)
	self.ui.let_raw_longitude.setReadOnly(True)
	self.ui.let_raw_latitude.setReadOnly(True)
	self.ui.let_raw_para4.setReadOnly(True)
	self.ui.let_raw_para2_5.setReadOnly(True)
	self.ui.let_raw_para3_6.setReadOnly(True)
	self.ui.let_raw_para7.setReadOnly(True)
	self.ui.let_raw_para8.setReadOnly(True)
	self.ui.let_raw_comdata_valid.setReadOnly(True)
	self.ui.let_raw_check_sum.setReadOnly(True)

def open_serial_connection(self):
	"""
	打開串口
	:return:
	"""
	if not self.serial_thread or not self.serial_thread.isRunning():
		# 參數(shù)校驗(yàn)
		if not self.__validate_setting__():
			return
 
		# 建立一個(gè)串口
		self.serial_thread = SerialThread(self.ui.cbx_com.currentText(),
												 self.ui.cbx_baud_rate.currentData(),
												 self.ui.cbx_data_bit.currentData(),
												 self.ui.cbx_parity_bit.currentData(),
												 self.ui.cbx_stop_bit.currentData(),
													   'hex' if self.ui.rbn_data_format_hex.isChecked() else 'ascii')
 
		self.serial_thread.data_received.connect(self.handle_data_received)
		self.serial_thread.serial_error.connect(self.handler_serial_error)
		self.serial_thread.start()
		self.ui.btx_start.setText("關(guān)閉串口")
	else:
		self.serial_thread.stop()
		# 打開串口操作
		self.ui.btx_start.setText("打開串口")

def __validate_setting__(self):
	"""
	校驗(yàn)串口設(shè)置參數(shù)
	:return:
	"""
	# 參數(shù)校驗(yàn)
	if self.ui.cbx_com.currentIndex() == -1:
		QMessageBox.warning(self, "Warning", "請(qǐng)選擇串口！")
		return False
	if self.ui.cbx_baud_rate.currentIndex() == -1:
		QMessageBox.warning(self, "Warning", "請(qǐng)選擇波特率！")
		return False
	if self.ui.cbx_parity_bit.currentIndex() == -1:
		QMessageBox.warning(self, "Warning", "請(qǐng)選擇校驗(yàn)位！")
		return False
	if self.ui.cbx_data_bit.currentIndex() == -1:
		QMessageBox.warning(self, "Warning", "請(qǐng)選擇數(shù)據(jù)位！")
		return False
	if self.ui.cbx_data_bit.currentIndex() == -1:
		QMessageBox.warning(self, "Warning", "請(qǐng)選擇停止位！")
		return False
 
	return True

def handler_serial_error(self, error):
	"""
	串口接收線程異常
	:param error:
	:return:
	"""
	QMessageBox.critical(self, '錯(cuò)誤', error)

def handle_data_received(self, data):
	"""
	接收數(shù)據(jù)
	:param data:接收到的數(shù)據(jù)
	:return:
	"""
	# 獲取時(shí)間
	current_time = QDateTime.currentDateTime().toString("yyyy-MM-dd hh:mm:ss")
 
	# 超過5000字符清空
	if len(self.ui.txt_recv_data_viewer.toPlainText()) > 5000:
		self.ui.txt_recv_data_viewer.clear()
	# 更新顯示區(qū)域中的數(shù)據(jù)
	if self.ui.cbx_show_time.isChecked():
		self.ui.txt_recv_data_viewer.insertPlainText(f"[{current_time}] {data}")
	else:
		self.ui.txt_recv_data_viewer.insertPlainText(f"[{data}")
 
	# 必須是hex格式
	if self.ui.rbn_data_format_ascii.isChecked():
		return
 
	# 如果報(bào)文最后兩位是換行符，則移除
	data = data.replace('0d 0a', '')
	data = bytes.fromhex(data.replace(' ', ''))
	# 解析data
	while data is not None:
		package_parse = PackageParse(data)
		data_format = package_parse.get_data_format()
		if data_format == PackageParse.Type.NAl_SOL:
			# 解析數(shù)據(jù)
			dict_msg = package_parse.get_nav_sol()
			# 寫入當(dāng)前時(shí)間，并保存到文件
			dict_msg["current_time"] = current_time
			save_csv('nav_data.csv', dict_msg)
			# 界面顯示
			self.ui.let_nav_header.setText(dict_msg["header"])
			self.ui.let_nav_work_state.setText(dict_msg["work_state"])
			self.ui.let_nav_param_state.setText(dict_msg["param_state"])
			self.ui.let_nav_run_time.setText(str(dict_msg["run_time"]))
			self.ui.let_nav_latitude.setText(str(dict_msg["latitude"]))
			self.ui.let_nav_longitude.setText(str(dict_msg["longitude"]))
			self.ui.let_nav_heave.setText(str(dict_msg["heave"]))
			self.ui.let_nav_east_speed.setText(str(dict_msg["east_speed"]))
			self.ui.let_nav_north_speed.setText(str(dict_msg["north_speed"]))
			self.ui.let_nav_vertical_speed.setText(str(dict_msg["vertical_speed"]))
			self.ui.let_nav_attitude_angle1.setText(str(dict_msg["attitude_angle1"]))
			self.ui.let_nav_attitude_angle2.setText(str(dict_msg["attitude_angle2"]))
			self.ui.let_nav_attitude_angle3.setText(str(dict_msg["attitude_angle3"]))
			self.ui.let_nav_attitude_velocity1.setText(str(dict_msg["attitude_velocity1"]))
			self.ui.let_nav_attitude_velocity2.setText(str(dict_msg["attitude_velocity2"]))
			self.ui.let_nav_attitude_velocity3.setText(str(dict_msg["attitude_velocity3"]))
			self.ui.let_nav_fault_code.setText(dict_msg["fault_code"])
			self.ui.let_nav_imu_time.setText(dict_msg["imu_time"])
			self.ui.let_nav_reserved.setText(dict_msg["reserved"])
			self.ui.let_nav_response_flag.setText(dict_msg["response_flag"])
			self.ui.let_nav_check_sum.setText(dict_msg["check_sum"])
		if data_format == PackageParse.Type.RAW:
			# 解析數(shù)據(jù)
			data = package_parse.get_raw()
			# 寫入當(dāng)前時(shí)間，并保存到文件
			data["current_time"] = current_time
			save_csv('raw_data.csv', data)
			# 界面顯示
			self.ui.let_raw_header.setText(dict_msg["header"])
			self.ui.let_raw__imu_time.setText(dict_msg["imu_time"])
			self.ui.let_raw_gyrox.setText(dict_msg["gyrox"])
			self.ui.let_raw_gyroy.setText(dict_msg["gyroy"])
			self.ui.let_raw_gyroz.setText(dict_msg["gyroz"])
			self.ui.let_raw_accex.setText(dict_msg["accex"])
			self.ui.let_raw_accey.setText(dict_msg["accey"])
			self.ui.let_raw_accez.setText(dict_msg["accez"])
			self.ui.let_raw_reserved.setText(dict_msg["reserved"])
			self.ui.let_raw_turntable_angle1.setText(dict_msg["turntable_angle1"])
			self.ui.let_raw_turntable_angle2.setText(dict_msg["turntable_angle2"])
			self.ui.let_raw_longitude.setText(dict_msg["longitude"])
			self.ui.let_raw_latitude.setText(dict_msg["latitude"])
			self.ui.let_raw_para4.setText(dict_msg["para4"])
			self.ui.let_raw_para2_5.setText(dict_msg["para2_5"])
			self.ui.let_raw_para3_6.setText(dict_msg["para3_6"])
			self.ui.let_raw_para7.setText(dict_msg["para7"])
			self.ui.let_raw_para8.setText(dict_msg["para8"])
			self.ui.let_raw_comdata_valid.setText(dict_msg["comdata_valid"])
			self.ui.let_raw_check_sum.setText(dict_msg["check_sum"])
		# 一次收到若干個(gè)數(shù)據(jù)包
		if data_format != PackageParse.Type.UNkNOWN and len(data) > data_format.value:
			data = data[data_format.value:]
		else:
			data = None

def save_csv(file_name, dict: {}):
    """
    保存到csv文件
    :param file_name:文件名
    :param dict: 數(shù)據(jù)，字典格式
    :return:
    """
    file_exists = os.path.exists(file_name)
    # Save data to CSV file
    with open(file_name, 'a', newline='') as csvfile:
        writer = csv.writer(csvfile)
        # 如果文件不存在，則寫入列名
        if not file_exists:
            writer.writerow(dict.keys())
        # 寫入數(shù)據(jù)
        writer.writerow(dict.values())

def validate_nal_sol(self):
	"""
	校驗(yàn)和
		導(dǎo)航結(jié)算：第49個(gè)字節(jié)為3-48字節(jié)累加和
	"""
	# 取出第3到第48字節(jié)
	selected_bytes = self.__byte_array[2:48]
 
	# 計(jì)算累加和并只保留低兩位數(shù)字
	checksum = sum(selected_bytes) & 0xFF
 
	return checksum == self.__byte_array[48]
 
def validate_raw(self):
	"""
	校驗(yàn)和
		原始信息：第58個(gè)字節(jié)為3-57字節(jié)累加和
	"""
	# 取出第3到第48字節(jié)
	selected_bytes = self.__byte_array[2:57]
 
	# 計(jì)算累加和并只保留低兩位數(shù)字
	checksum = sum(selected_bytes) & 0xFF
 
	return checksum == self.__byte_array[57]
 
def get_data_format(self):
	"""
	返回?cái)?shù)據(jù)類型：在慣性裝置的AXS31接口，里面有兩路數(shù)據(jù)，一路稱為導(dǎo)航解算，一路稱為原始數(shù)
	:return: 0: 導(dǎo)航解算   1:原始信息  2：非法數(shù)據(jù)
	"""
	# 原始信息報(bào)文
	if (len(self.__byte_array) >= 58 and self.__byte_array[0] == 0x5a and self.__byte_array[1] == 0x5a
			and self.validate_raw()):
		return PackageParse.Type.RAW
	# 導(dǎo)航解算的報(bào)文內(nèi)容
	if (len(self.__byte_array) >= 49 and self.__byte_array[0] == 0x5a and self.__byte_array[1] == 0x5a
			and self.validate_nal_sol()):
		return PackageParse.Type.NAl_SOL
	return PackageParse.Type.UNkNOWN

def translate(hex_array: [], unit=None):
    """
    將字節(jié)數(shù)組轉(zhuǎn)換為16進(jìn)制字符串  [0x0a,0x04]-> 0x0a 0x04
    :param hex_array: 字節(jié)數(shù)組
    :param unit 量綱
    :return:
    """
    if unit is not None:
        # 將字節(jié)數(shù)組按照小端格式轉(zhuǎn)換為數(shù)字
        num = int.from_bytes(hex_array, byteorder='little')
        return num / unit
    return ' '.join(f'0x{val:02X}' for val in hex_array)
    
def get_nav_sol(self):
	"""
	獲取導(dǎo)航解算數(shù)據(jù)
	:return:
	"""
	if self.get_data_format() != PackageParse.Type.NAl_SOL:
		print("數(shù)據(jù)格式錯(cuò)誤，非導(dǎo)航解算報(bào)文！")
		return None
 
	message_dict = {}
 
	# 解析報(bào)文頭
	message_dict['header'] = translate(self.__byte_array[0:2])
 
	# 解析工作狀態(tài)
	message_dict['work_state'] = translate(self.__byte_array[2:3])
 
	# 解析參數(shù)狀態(tài)
	message_dict['param_state'] = translate(self.__byte_array[3:4])
 
	# 解析運(yùn)行時(shí)間
	message_dict['run_time'] = translate(self.__byte_array[4:8], 20)
 
	# 解析緯度
	message_dict['latitude'] = translate(self.__byte_array[8:11], 93206.75556)
 
	# 解析經(jīng)度
	message_dict['longitude'] = translate(self.__byte_array[11:14], 46603.37778)
 
	# 解析升沉
	message_dict['heave'] = translate(self.__byte_array[14:16], 100)
 
	# 解析東速
	message_dict['east_speed'] = translate(self.__byte_array[16:18], 100)
 
	# 解析北速
	message_dict['north_speed'] = translate(self.__byte_array[18:20], 100)
 
	# 解析垂速
	message_dict['vertical_speed'] = translate(self.__byte_array[20:22], 100)
 
	# 解析姿態(tài)角1
	message_dict['attitude_angle1'] = translate(self.__byte_array[22:25], 0.25 * 93206.75556)
 
	# 解析姿態(tài)角2
	message_dict['attitude_angle2'] = translate(self.__byte_array[25:28], 93206.75556)
 
	# 解析姿態(tài)角3
	message_dict['attitude_angle3'] = translate(self.__byte_array[28:31], 93206.75556)
 
	# 姿態(tài)角速率1
	message_dict['attitude_velocity1'] = translate(self.__byte_array[31:34], 93206.75556)
 
	# 姿態(tài)角速率2
	message_dict['attitude_velocity2'] = translate(self.__byte_array[34:37], 93206.75556)
 
	# 姿態(tài)角速率3
	message_dict['attitude_velocity3'] = translate(self.__byte_array[37:40], 93206.75556)
 
	# 故障碼
	message_dict['fault_code'] = translate(self.__byte_array[40:41])
 
	# IMU時(shí)間
	message_dict['imu_time'] = translate(self.__byte_array[41:43])
 
	# 備用
	message_dict['reserved'] = translate(self.__byte_array[43:47])
 
	# 應(yīng)答標(biāo)志
	message_dict['response_flag'] = translate(self.__byte_array[47:48])
 
	# 校驗(yàn)和
	message_dict['check_sum'] = translate(self.__byte_array[48:49])
 
	return message_dict

def get_raw(self):
	"""
	獲取原始信息報(bào)文
	:return:
	"""
	if self.get_data_format() != PackageParse.Type.RAW:
		print("數(shù)據(jù)格式錯(cuò)誤，非原始信息報(bào)文！")
		return None
 
	message_dict = {}
 
	# 解析報(bào)文頭
	message_dict['header'] = translate(self.__byte_array[0:2])
 
	# IMUtime
	message_dict['imu_time'] = translate(self.__byte_array[2:6])
 
	# GYROX
	message_dict['gyrox'] = translate(self.__byte_array[6:10])
 
	# GYROY
	message_dict['gyroy'] = translate(self.__byte_array[10:14])
 
	# GYROZ
	message_dict['gyroz'] = translate(self.__byte_array[14:18])
 
	# ACCEX
	message_dict['accex'] = translate(self.__byte_array[18:22])
 
	# ACCEY
	message_dict['accey'] = translate(self.__byte_array[22:26])
 
	# ACCEZ
	message_dict['accez'] = translate(self.__byte_array[26:30])
 
	# 備用
	message_dict['reserved'] = translate(self.__byte_array[30:34])
 
	# 轉(zhuǎn)臺(tái)角1
	message_dict['turntable_angle1'] = translate(self.__byte_array[34:37])
 
	# 轉(zhuǎn)臺(tái)角2
	message_dict['turntable_angle2'] = translate(self.__byte_array[37:40])
 
	# GPS經(jīng)度
	message_dict['longitude'] = translate(self.__byte_array[40:43])
 
	# GPS緯度
	message_dict['latitude'] = translate(self.__byte_array[43:46])
 
	# Para[4](電磁/牽引時(shí)為航向)
	message_dict['para4'] = translate(self.__byte_array[46:48])
 
	# Para[2]/para[5]
	message_dict['para2_5'] = translate(self.__byte_array[48:50])
 
	# Para[3]/para[6]
	message_dict['para3_6'] = translate(self.__byte_array[50:52])
 
	# Para[7]
	message_dict['para7'] = translate(self.__byte_array[52:54])
 
	# Para[8]
	message_dict['para8'] = translate(self.__byte_array[54:56])
 
	# comdatavalid
	message_dict['comdata_valid'] = translate(self.__byte_array[56:57])
 
	# 校驗(yàn)和
	message_dict['check_sum'] = translate(self.__byte_array[57:58])
 
	return message_dict

def send_serial_data(self, data: str):
	"""
	發(fā)送數(shù)據(jù)
	:return:
	"""
	if not self.serial_thread or not self.serial_thread.isRunning():
		QMessageBox.warning(self, "Warning", "請(qǐng)先打開串口！")
		return
 
	data = self.ui.let_send_data_viewer.text()
	if data != "":
		self.serial_thread.send_data(data)
		self.ui.let_send_data_viewer.clear()

pip install pyserial -i https://pypi.tuna.tsinghua.edu.cn/simple

from PyQt5.QtCore import QThread,pyqtSignal
import serial
 
class SerialThread(QThread):
    """
    創(chuàng)建一個(gè)繼承自QThread的SerialThread類，實(shí)現(xiàn)串口數(shù)據(jù)的讀取/發(fā)送
    """
 
    # 用于發(fā)送串口數(shù)據(jù)接收信號(hào)
    data_received = pyqtSignal(str)
    # 串口打開/接收異常
    serial_error = pyqtSignal(str)
    
    .......

def __init__(self, port, baud_rate, data_bits, parity_bits, stop_bits, data_format):
	"""
	初始化
	:param port: 串口號(hào)
	:param baud_rate: 波特率
	:param data_bits: 數(shù)據(jù)位
	:param stop_bits: 停止位
	:param parity_bits: 奇偶校驗(yàn)位
	"""
	super().__init__()
	self.port = port
	self.baud_rate = baud_rate
	self.data_bits = data_bits
	self.stop_bits = stop_bits
	self.parity_bits = parity_bits
 
	# 串口已經(jīng)運(yùn)行標(biāo)志位
	self.running = False
	# 串口
	self.serial = None
	# 數(shù)據(jù)格式
	self.__date_format = data_format
	# 發(fā)送開啟追加換行
	self.__auto_line = True
    
@property
def date_format(self):
	"""
	#把一個(gè)getter方法變成屬性
	:return:
	"""
	return self.__date_format
 
@date_format.setter
def date_format(self, value):
	"""
	# 負(fù)責(zé)把一個(gè)setter方法變成屬性賦值
	:param value:
	:return:
	"""
	if not isinstance(value, str):
		raise ValueError('date_format must be an str')
	if value not in ['hex', 'ascii']:
		raise ValueError('date_format must in [hex,ascii]')
	self.__date_format = value

def run(self):
	"""
	打開串口
	:return:
	"""
	# 串口已經(jīng)運(yùn)行
	if self.running:
		return
 
	try:
		# 建立一個(gè)串口
		with serial.Serial(port=self.port,
						   baudrate=self.baud_rate,
						   parity=self.parity_bits,
						   bytesize=self.data_bits,
						   stopbits=self.stop_bits,
						   timeout=2) as self.serial:
			self.running = True
			while self.running:
				data = self.__read_data__()
				if data:
					self.data_received.emit(data)
	except Exception as e:
		print("打開串口時(shí)失?。?, e)
		self.serial_error.emit("打開串口時(shí)失敗！")

def __read_data__(self):
	"""
	按行接收數(shù)據(jù)
	:return:
	"""
	if self.serial is None or not self.serial.isOpen():
		return
 
	try:
		# 讀取串口數(shù)據(jù) 例如：b'DDR V1.12 52218f4949 cym 23/07/0'
		byte_array = self.serial.readline()
		if len(byte_array) == 0:
			return None
		# ascii顯示
		if self.__date_format == 'ascii':
			# 串口接收到的字符串為b'ABC',要轉(zhuǎn)化成unicode字符串才能輸出到窗口中去
			data_str = byte_array.decode('utf-8')
		else:
			# 串口接收到的字符串為b'ZZ\x02\x03Z'，要轉(zhuǎn)換成16進(jìn)制字符串顯示
			data_str = ' '.join(format(x, '02x') for x in byte_array)
			if self.__auto_line:
				data_str += '\r\n'
		return data_str
	except Exception as e:
		print("接收數(shù)據(jù)異常：", e)
		self.serial_error.emit("接收數(shù)據(jù)異常！")

def send_data(self, data: str):
	"""
	發(fā)送數(shù)據(jù)
	:return:
	"""
	if not self.running:
		self.serial_error.emit("請(qǐng)先打開串口！")
		return
 
	# hex發(fā)送 比如：5a 5a 02 03 5a -> b'ZZ\x02\x03Z'
	if self.__date_format == 'hex':
		data_str = data.strip()
		send_list = []
		while data_str != '':
			try:
				num = int(data_str[0:2], 16)
			except ValueError:
				self.serial_error.emit('請(qǐng)輸入十六進(jìn)制數(shù)據(jù)，以空格分開!')
				return
			data_str = data_str[2:].strip()
			send_list.append(num)
		if self.__auto_line:
			send_list.append(0x0d)
			send_list.append(0x0a)
		byte_array = bytes(send_list)
	else:
		if self.__auto_line:
			data += '\r\n'
		# ascii發(fā)送 比如：'ABC' -> b'ABC'
		byte_array = data.encode('utf-8')
 
	try:
		self.serial.write(byte_array)
	except Exception as e:
		print("發(fā)送失敗", e)
		self.serial_error.emit('發(fā)送失敗!')