LogTT_tools/main.py

# -*- coding: gb2312 -*-

import time
from os import path, makedirs
import sys
from datetime import datetime
from PyQt5.QtWidgets import QApplication, QMainWindow
from PyQt5.QtCore import pyqtSignal
from PyQt5.QtCore import QUrl
from PyQt5.QtCore import QTimer
from PyQt5.QtWebEngineWidgets import QWebEngineView
from PyQt5.QtGui import QTextCursor
from mainwindow import *
import logging
from platform import node
from paho.mqtt import client as mqtt_client
from json import loads as json_loads, dumps as json_dumps, JSONDecodeError
from threading import Thread
from requests import get as requests_get


# pyinstaller.exe -D -n LogTT_tools_V0.0.0.2 main.py
# pyinstaller.exe -w -F -n LogTT_tools_V0.0.0.3 main.py

client_id = "LogTT_tool_" + node()
g_client = mqtt_client.Client(client_id)
mqtt_broker = 'test-mqtt.cpyypt.cn'
mqtt_port = 9000
mqtt_user = 'admin'
mqtt_pwd = 'houjianwei'
mqtt_keepalive = 5
mqtt_timer_time_out = 3*1000
# mqtt_subtopic = [("cpyypt/up/9002/#", 0), ("cpyypt/down/9002/broadcast", 0)]
mqtt_subtopic = [("cpyypt/up/9002/#", 0)]
mqtt_pubtopic = "cpyypt/down/9002/0000000001"

version = "0.0.0.3"
version_w_ascii = 1
version_w_hex = 1

error_msg = ""
info_msg = ""

start_ttl = "TTL_U="
start_rs232 = "RS232_U="
start_rs485 = "RS485_U="
start_fixed = "FIXED_INFO="
start_heart = "HEART_MSG="
start_pt = "PT_CMD="
data_buf = {}

LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s"
logging.basicConfig(filename='formater.log', level=logging.DEBUG, format=LOG_FORMAT)
logging.debug("\n\n................")

def get_log_file_name_ascii(sn):
    # global
    log_file_name = f"./log/LogTT_{sn}_ascii_" + datetime.now().strftime('%Y-%m-%d') + ".log"
    return log_file_name


def get_log_file_name_hex(sn):
    # global
    log_file_name = f"./log/LogTT_{sn}_hex_" + datetime.now().strftime('%Y-%m-%d') + ".log"
    return log_file_name

def is_valid_json(json_str):
    try:
        json_loads(json_str)
        return True
    except JSONDecodeError:
        return False

class MyWindow(QMainWindow, Ui_MainWindow):
    map_refresh_signal = pyqtSignal(int, float, float)
    rx_data_signal = pyqtSignal(str, str, str)
    mqtt_connect_state_signal = pyqtSignal(str)
    log_display_update_signal = pyqtSignal(str)

    def __init__(self, parent=None):
        print("__init__")
        logging.info("__init__")

        super(MyWindow, self).__init__(parent)
        self.setupUi(self)
        self.initUI()
        self.set_svr_env = "test"
        self.sn_current = 0
        self.mqtt_connect_state = "未连接"

        browser = QWebEngineView()
        self.browser = browser
        self.horizontalLayout.layout().addWidget(self.browser)
        self.horizontalLayoutWidget.setGeometry(QtCore.QRect(0, 0, 1024, 600))

        # self.pushButton_set_subscribe.clicked.connect(lambda: {self.mqtt_subscribe(g_client), Thread(target=g_client.loop_forever, daemon=True).start()})
        self.pushButton_set_subscribe.clicked.connect(lambda: {self.mqtt_subscribe(g_client)})
        self.mqtt_connect(g_client)
        self.mqtt_subscribe(g_client)
        self.mqtt_disconnect(g_client)
        self.mqtt_thread = Thread(target=g_client.loop_forever, daemon=True)
        self.mqtt_thread.start()

        if not path.exists("log"):
            makedirs("log")
        self.textEdit_log_0rx_display.setText("")
        self.textEdit_log_0rx_display.moveCursor(QTextCursor.End)
        self.textEdit_log_0rx_display.document().setMaximumBlockCount(10000)
        self.display_clear()

    def initUI(self):
        print("initUI")
        logging.info("initUI")
        self.setWindowTitle("LogTT_tools:test-" + version + ",未连接")
        # self.tabWidget.setCurrentIndex(self.tabWidget.indexOf(self.tab_lsb))

        # self.comboBox_log_sn.currentIndexChanged.connect(self.comboBox_log_sn_change)
        self.comboBox_log_sn.activated.connect(self.display_update)
        self.comboBox_log_sn.activated.connect(self.sn_update_something)

        self.pushButton_log_rx_clr.clicked.connect(self.log_rx_clr)

        self.map_refresh_signal.connect(self.map_refresh)
        self.rx_data_signal.connect(self.parse_data)
        self.mqtt_connect_state_signal.connect(self.mainWindow_title_update)
        self.log_display_update_signal.connect(self.log_display_update)

        self.pushButton_set_create.clicked.connect(self.set_create)
        self.pushButton_set_send.clicked.connect(self.set_send)
        self.pushButton_set_create_send.clicked.connect(self.set_create_send)

        # self.pushButton_set_send.clicked.connect(lambda: {self.publish(g_client)})
        self.pushButton_set_connect.clicked.connect(lambda: {self.connect(g_client)})

        self.comboBox_set_svr.activated.connect(self.set_svr)
        self.checkBox_set_io_up_enable.clicked.connect(self.set_io_up_able)
        self.checkBox_set_io_up_disable.clicked.connect(self.set_io_up_disable)
        self.checkBox_set_io_down_enable.clicked.connect(self.set_io_down_able)
        self.checkBox_set_io_down_disable.clicked.connect(self.set_io_down_disable)
        self.checkBox_set_io_parameter.clicked.connect(self.set_io_parameter)
        self.checkBox_set_io_cfg.clicked.connect(self.set_io_cfg)

        self.checkBox_set_default_tx_sn.clicked.connect(self.set_default_tx_sn)
        self.lineEdit_tx_sn.textChanged.connect(self.set_tx_sn_change)

        self.pushButton_lsb_refresh.clicked.connect(self.lsb_get_fixed_info)
        self.pushButton_log_send.clicked.connect(self.log_send)
        self.pushButton_log_refresh.clicked.connect(self.log_refresh)
        # self.lineEdit_log_rx_buf_max.textChanged.connect(self.log_rx_buf_max)
        self.pushButton_log_2rs485_0on.clicked.connect(self.log_rs485_up_enable)
        self.pushButton_log_2rs485_1off.clicked.connect(self.log_rs485_up_disable)

    def timer_test_f(self, str):
        print("mqtt timer_test_f," + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
        self.timer_test_t.start(3*1000)  # 10秒后重连

    def mqtt_timer_f(self):
        print("mqtt mqtt_timer_f," + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
        # self.mqtt_timer.start(3*1000)  # 10秒后重连

    def log_rx_clr(self):
        self.textEdit_log_0rx_display.setText("")

    def mainWindow_title_update(self, state):
        self.setWindowTitle("LogTT_tools:" + "test-" + version + "," + state)
        if state == "已连接":
            self.mqtt_subscribe(g_client)

    def mqtt_connect(self, client):
        print("mqtt mqtt_connect," + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
        try:
            def on_connect(client, userdata, flags, rc):
                if rc == 0:
                    print("Connected to MQTT Broker!" + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
                    self.mqtt_connect_state_signal.emit("已连接")
                else:
                    print("Failed to connect, return code %d\n", rc)
                    self.mqtt_connect_state_signal.emit("连接失败")
                    # self.mqtt_timer.start(mqtt_timer_time_out)  # 10秒后重连

            client.username_pw_set(mqtt_user, mqtt_pwd)
            client.on_connect = on_connect
            client.connect(mqtt_broker, mqtt_port, mqtt_keepalive)
            print("[√]连接成功," + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])

        except Exception as e:
            print("[!]Connect error")
            print("[×]连接失败," + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
            pass

    def mqtt_disconnect(self, client):
        print("mqtt mqtt_disconnect," + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])

        def on_disconnect(client, userdata, rc):
            print("self.sender()", self.sender())
            if rc != 0:
                print("Unexpected disconnection." + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
            else:
                print("Disconnected from MQTT." + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
            # 更新 UI 线程
            self.mqtt_connect_state_signal.emit("已断开")

        client.on_disconnect = on_disconnect

    def mqtt_subscribe(self, client):
        print("mqtt mqtt_subscribe," + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
        global info_msg
        global error_msg

        def on_message(client, userdata, msg):
            LogTT_SN = str.split(msg.topic, "/")[-1]
            head = b""
            body = b""
            cmd_type = ""
            flag_ascii = False
            flag_hex = True
            if msg.payload[:len(start_ttl)] == start_ttl.encode("utf-8"):
                cmd_type = start_ttl
                flag_ascii = True
                flag_hex = True
                head = msg.payload[:len(start_ttl)]
                body = msg.payload[len(start_ttl):]

            elif msg.payload[:len(start_rs232)] == start_rs232.encode("utf-8"):
                cmd_type = start_rs232
                flag_ascii = False
                flag_hex = True
                head = msg.payload[:len(start_rs232)]
                body = msg.payload[len(start_rs232):]

            elif msg.payload[:len(start_rs485)] == start_rs485.encode("utf-8"):
                cmd_type = start_rs485
                flag_ascii = False
                flag_hex = True
                head = msg.payload[:len(start_rs485)]
                body = msg.payload[len(start_rs485):]

            elif msg.payload[:len(start_fixed)] == start_fixed.encode("utf-8"):
                cmd_type = start_fixed
                flag_ascii = True
                flag_hex = False
                head = msg.payload[:len(start_fixed)]
                body = msg.payload[len(start_fixed):]

            elif msg.payload[:len(start_heart)] == start_heart.encode("utf-8"):
                cmd_type = start_heart
                flag_ascii = True
                flag_hex = False
                head = msg.payload[:len(start_heart)]
                body = msg.payload[len(start_heart):]

            elif msg.payload[:len(start_pt)] == start_pt.encode("utf-8"):
                cmd_type = start_pt
                flag_ascii = True
                flag_hex = False
                head = msg.payload[:len(start_pt)]
                body = msg.payload[len(start_pt):]
            else:
                cmd_type = ""
                flag_ascii = False
                flag_hex = True
                body = msg.payload


            if flag_ascii:
                try:
                    body_str = body.decode('gb2312')
                except UnicodeDecodeError:
                    try:
                        body_str = body.decode('utf-8')
                    except UnicodeDecodeError:
                        body_str = body.hex().upper()
                head_str = head.decode('utf-8')
                msg_ascii = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3] + "-->:"
                msg_ascii = msg_ascii + f"topic:{msg.topic} msg:{head_str}\r\n{body_str}"
                msg_ascii = msg_ascii.rstrip() + "\r\n"
                # 输出信息
                try:
                    with open(get_log_file_name_ascii(LogTT_SN), 'a') as f:
                        global version_w_ascii
                        if version_w_ascii > 0:
                            f.write("version:" + version + "\n")
                            version_w_ascii = 0
                        f.write(msg_ascii)
                        f.close()
                except Exception as e:
                    print(str(e))

                print(msg_ascii, end="")
                if self.comboBox_log_sn.currentIndex() == -1:
                    # self.textEdit_log_0rx_display.append(msg_ascii)
                    self.log_display_update_signal.emit(msg_ascii)
                elif self.comboBox_log_sn.currentText() == LogTT_SN:
                    # self.textEdit_log_0rx_display.append(msg_ascii)
                    self.log_display_update_signal.emit(msg_ascii)

                self.rx_data_signal.emit(LogTT_SN, head_str, body_str)
                # self.parse_data(LogTT_SN, head_str, body_str)


            if flag_hex:
                msg_hex = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3] + "-->:"
                msg_hex = msg_hex + f"topic:{msg.topic} msg:{head.decode('utf-8')}{body.hex().upper()}"
                # 输出信息
                try:
                    with open(get_log_file_name_hex(LogTT_SN), 'a') as f:
                        global version_w_hex
                        if version_w_hex > 0:
                            f.write("version:" + version + "\n")
                            version_w_hex = 0
                        f.write(msg_hex + "\r\n")
                        f.close()
                except Exception as e:
                    print(str(e))
                if not flag_ascii:
                    print(msg_hex)
                    # self.textEdit_log_0rx_display.append(msg_ascii)
                    self.log_display_update_signal.emit(msg_hex)

        client.subscribe(mqtt_subtopic)
        client.on_message = on_message
        print("[√]订阅成功," + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])

    def mqtt_publish_test(self, client):
        print("mqtt mqtt_publish_test," + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
        try:
            msg = "qqqqqqqqqqqqqqq"

            result = client.publish(mqtt_pubtopic, msg)
            # result: [0, 1]
            status = result[0]

            if status == 0:
                print(f"[√]发送成功, Send `{msg}` to topic `{mqtt_pubtopic}`")
            else:
                print(f"[×]发送失败, Failed to send message to topic {mqtt_pubtopic}")

        except:
            print("[!]请先连接服务器！")

    def parse_data(self, sn, head, body):

        if data_buf.get(sn) is None:
            data_buf[sn] = {}
        if data_buf[sn].get("base_info") is None:
            data_buf[sn]["base_info"] = {}
        if data_buf[sn].get("pos_info") is None:
            data_buf[sn]["pos_info"] = {}
        data_buf[sn]["heart_time_stamp"] = datetime.now().strftime('%H:%M:%S.%f')[:-3]

        # start_fixed = "FIXED_INFO="
        # start_heart = "HEART_MSG="
        # FIXED_INFO={"base_info":{"name":"张三","uplinkB":1010,"rsrp":-99,"tel":"13888888888","reset_times":49,"project":"LogTT","csq":23,"downlinkB":573,"rsrq":-13,"dev_type":"9002","simid":0,"muid":"20230818221817A863212A0300364949","snr":-1,"version":"9002.4.002","iccid":"89860622330053052122","imei":"869861069998891","rssi":-66,"dev_pd":"20240329","dev_sn":1},"pos_info":{"lng":0,"lat":0}}

        # if head == start_fixed or head == start_heart:
        if is_valid_json(body):
            body_dict = json_loads(body)
            if body_dict["base_info"]:
                base_info = body_dict["base_info"]
                if base_info.get("project"):    data_buf[sn]["base_info"]["project"]    = base_info.get("project")
                if base_info.get("version"):    data_buf[sn]["base_info"]["version"]    = base_info.get("version")
                if base_info.get("name"):       data_buf[sn]["base_info"]["name"]       = base_info.get("name")
                if base_info.get("tel"):        data_buf[sn]["base_info"]["tel"]        = base_info.get("tel")
                if base_info.get("dev_type"):   data_buf[sn]["base_info"]["dev_type"]   = base_info.get("dev_type")
                if base_info.get("dev_sn"):     data_buf[sn]["base_info"]["dev_sn"]     = str(sn)
                if base_info.get("dev_pd"):     data_buf[sn]["base_info"]["dev_pd"]     = base_info.get("dev_pd")
                if base_info.get("reset_times"): data_buf[sn]["base_info"]["reset_times"] = base_info.get("reset_times")
                if base_info.get("imei"):       data_buf[sn]["base_info"]["imei"]       = base_info.get("imei")
                if base_info.get("muid"):       data_buf[sn]["base_info"]["muid"]       = base_info.get("muid")
                if base_info.get("iccid"):      data_buf[sn]["base_info"]["iccid"]      = base_info.get("iccid")
                if base_info.get("simid"):      data_buf[sn]["base_info"]["simid"]      = base_info.get("simid")
                if base_info.get("csq"):        data_buf[sn]["base_info"]["csq"]        = base_info.get("csq")
                if base_info.get("rssi"):       data_buf[sn]["base_info"]["rssi"]       = base_info.get("rssi")
                if base_info.get("snr"):        data_buf[sn]["base_info"]["snr"]        = base_info.get("snr")
                if base_info.get("rsrq"):       data_buf[sn]["base_info"]["rsrq"]       = base_info.get("rsrq")
                if base_info.get("rsrp"):       data_buf[sn]["base_info"]["rsrp"]       = base_info.get("rsrp")
                if base_info.get("uplinkB"):    data_buf[sn]["base_info"]["uplinkB"]    = base_info.get("uplinkB")
                if base_info.get("downlinkB"):  data_buf[sn]["base_info"]["downlinkB"]  = base_info.get("downlinkB")

            if body_dict["pos_info"]:
                pos_info = body_dict["pos_info"]
                if pos_info.get("lng"): data_buf[sn]["pos_info"]["lng"] = pos_info.get("lng")
                if pos_info.get("lat"): data_buf[sn]["pos_info"]["lat"] = pos_info.get("lat")
        # 更新数据
        self.data_update(sn)

    def data_update(self, sn):

        self.comboBox_log_sn_add(sn)
        self.display_update(sn)
        self.dev_list_refresh(sn)

        if self.comboBox_log_sn.currentIndex() != -1:
            sn_cur = self.comboBox_log_sn.currentText()
            if sn == sn_cur:
                if data_buf.get(sn_cur):
                    if data_buf[sn_cur].get("pos_info"):
                        lng = data_buf[sn_cur]["pos_info"].get("lng", 116.1666213)
                        lat = data_buf[sn_cur]["pos_info"].get("lat", 39.9007607)
                        # self.map_refresh_signal.emit(sn_cur, float(lng), float(lat))

    def comboBox_log_sn_change(self):
        # self.sn_current = sn
        self.display_update()

    def comboBox_log_sn_add(self, sn):
        if sn == "0000000116":
            return
        if self.comboBox_log_sn.findText(sn) == -1:
            self.comboBox_log_sn.addItem(sn)
            num_items = self.comboBox_log_sn.count()
            current_text = self.comboBox_log_sn.currentText()
            # 插入，且排序
            text_list = []
            for index in range(num_items):
                text_list.append(self.comboBox_log_sn.itemText(index))
            text_list = sorted(text_list)
            for index in range(num_items):
                self.comboBox_log_sn.setItemText(index, text_list[index])
            self.comboBox_log_sn.setCurrentText(current_text)

    def display_base_info_clear(self):
        self.label_log_01dev_type.setText("设备类型:")
        self.label_log_02sn.setText("设备SN:")
        self.label_log_03reset_times.setText("复位次数:")
        self.label_log_04app_ver.setText("固件版本:")
        self.label_log_05name.setText("姓名:")
        self.label_log_06telephone.setText("电话:")
        self.label_log_09imei.setText("IMEI:")
        self.label_log_10iccid.setText("ICCID:")
        self.label_log_11rssi.setText("信号强度:")
        self.label_log_12snr.setText("信噪比:")

    def display_pos_info_clear(self):
        self.label_log_07longitude.setText("经度:")
        self.label_log_08latitude.setText("维度:")

    def display_clear(self):
        self.display_base_info_clear()
        self.display_pos_info_clear()
        self.groupBox_log_1base.setTitle("日志设备基础信息")

    def display_update(self, sn=0):
        sn_cur = self.comboBox_log_sn.currentText()
        if sn_cur in data_buf:
            if data_buf[sn_cur]["base_info"]:
                base_info = data_buf[sn_cur]["base_info"]
                self.label_log_01dev_type.setText("设备类型:" + base_info.get("dev_type", ""))
                self.label_log_02sn.setText("设备SN:" + str(sn_cur))
                self.label_log_03reset_times.setText("复位次数:" + str(base_info.get("reset_times", 0)))
                self.label_log_04app_ver.setText("固件版本:" + base_info.get("version", ""))
                self.label_log_05name.setText("姓名:" + base_info.get("name", ""))
                self.label_log_06telephone.setText("电话:" + base_info.get("tel", ""))
                self.label_log_09imei.setText("IMEI:" + base_info.get("imei", ""))
                self.label_log_10iccid.setText("ICCID:" + base_info.get("iccid", ""))
                self.label_log_11rssi.setText("信号强度:" + str(base_info.get("rssi", 0)))
                self.label_log_12snr.setText("信噪比:" + str(base_info.get("snr", 0)))
            else:
                self.display_base_info_clear()
            if data_buf[sn_cur]["pos_info"]:
                pos_info = data_buf[sn_cur]["pos_info"]
                lng = pos_info.get("lng", 0) if pos_info.get("lng", 0) != 0 else 116.1666213
                self.label_log_07longitude.setText("经度:" + str(lng))

                lat = pos_info.get("lat", 0) if pos_info.get("lat", 0) != 0 else 39.9007607
                self.label_log_08latitude.setText("维度:" + str(lat))
            else:
                self.display_pos_info_clear()
            self.groupBox_log_1base.setTitle("日志设备基础信息:" + str(data_buf[sn_cur]["heart_time_stamp"]))

        else:
            self.display_clear()

        self.label_lsb_01dev_type.setText(self.label_log_01dev_type.text())
        self.label_lsb_02sn.setText(self.label_log_02sn.text())
        self.label_lsb_03reset_times.setText(self.label_log_03reset_times.text())
        self.label_lsb_04app_ver.setText(self.label_log_04app_ver.text())
        self.label_lsb_05name.setText(self.label_log_05name.text())
        self.label_lsb_06telephone.setText(self.label_log_06telephone.text())
        self.label_lsb_09imei.setText(self.label_log_09imei.text())
        self.label_lsb_10iccid.setText(self.label_log_10iccid.text())
        self.label_lsb_11rssi.setText(self.label_log_11rssi.text())
        self.label_lsb_12snr.setText(self.label_log_12snr.text())
        self.label_lsb_07longitude.setText(self.label_log_07longitude.text())
        self.label_lsb_08latitude.setText(self.label_log_08latitude.text())

        self.groupBox_lsb_1base.setTitle(self.groupBox_log_1base.title())

    def sn_update_something(self):
        if self.checkBox_set_default_tx_sn.isChecked():
            sn_text = self.comboBox_log_sn.currentText()
            sn_num = int(sn_text)
            sn_text = f"%u" % sn_num
            self.lineEdit_tx_sn.setText(sn_text)
            # self.lineEdit_5pub_topic.setText("cpyypt/down/9002/" + self.lineEdit_tx_sn.text().zfill(10))

    def dev_list_refresh(self, sn):
        dev_str = self.plainTextEdit_log_dev_list.toPlainText()
        dev_list = str.split(dev_str)
        # dev_list = [dev for dev in data_buf.keys()]
        if sn not in dev_list:
            # print(dev_list)
            dev_list.append(sn)
            dev_list = sorted(dev_list)
            dev_text = ""
            for dev in dev_list:
                dev_text = dev_text + str(dev) + "\n"
            self.plainTextEdit_log_dev_list.setPlainText(dev_text)

    def mqtt_send_data(self, client, pub_topic, userdata):

        try:
            result = client.publish(pub_topic, userdata)
            # result: [0, 1]
            status = result[0]

            if status == 0:
                msg = f"[√]发送成功, Send `{userdata}` to topic `{pub_topic}`"

            else:
                msg = f"[×]发送失败, Failed to send message to topic {pub_topic}"
        except:
            msg = "[!]请先连接服务器！"

        # self.textEdit_log_0rx_display.append(msg + "\n")
        self.log_display_update_signal.emit(msg + "\n")

    def log_display_update(self, text):
        # 当前光标是否在最后
        cursor_end = False
        position = self.textEdit_log_0rx_display.textCursor().position()
        length = len(self.textEdit_log_0rx_display.toPlainText())

        if position == length:
            cursor_end = True

        self.textEdit_log_0rx_display.append(text)


        # 是否将光标移动到最后
        if cursor_end:
            self.textEdit_log_0rx_display.moveCursor(QTextCursor.End)

    def log_rx_buf_max(self):
        max_text = self.lineEdit_log_rx_buf_max.text()
        max_num = 10000
        try:
            max_num = int(max_text)
        except ValueError:
            max_num = 10000

        if max_num <= 100:
            max_num = 100
            self.lineEdit_log_rx_buf_max.setText(max_num)
        if max_num >= 100000:
            max_num = 100000
            self.lineEdit_log_rx_buf_max.setText(f'{max_num}')
        self.textEdit_log_0rx_display.document().setMaximumBlockCount(max_num)

    def log_refresh(self):
        if self.comboBox_log_sn.currentIndex() != -1:
            sn_cur = self.comboBox_log_sn.currentText()
            send_topic = "cpyypt/down/9002/" + sn_cur
            send_data = 'PT_CMD={"get_fixed_info": "true"}'
            self.mqtt_send_data(g_client, send_topic, send_data)

    def log_send(self):
        if self.comboBox_log_sn.currentIndex() != -1:
            sn_cur = self.comboBox_log_sn.currentText()

            cmd_header = "TTD_TTL"
            cmd_tail = ""

            io_select = self.comboBox_tx_io_select.currentIndex()
            if io_select == 0:
                cmd_header = 'TTD_TTL'
            elif io_select == 1:
                cmd_header = 'TTD_RS232'
            elif io_select == 2:
                cmd_header = 'TTD_RS485'

            if self.radioButton_log_1tx_ascii.isChecked():
                cmd_header = cmd_header + "_A="
                if self.checkBox_r_n.isChecked():
                    cmd_tail = "\r\n"
            else:
                cmd_header = cmd_header + "_H="

            cmd_body = self.textEdit_log_1tx.toPlainText()

            send_data = cmd_header + cmd_body + cmd_tail
            send_topic = "cpyypt/down/9002/" + sn_cur

            self.mqtt_send_data(g_client, send_topic, send_data)

    def log_rs485_up_enable(self):
        if self.comboBox_log_sn.currentIndex() != -1:
            sn_cur = self.comboBox_log_sn.currentText()
            send_data = 'PT_CMD={"rs485_cfg": {"up_enable": 1}}'
            send_topic = "cpyypt/down/9002/" + sn_cur
            self.mqtt_send_data(g_client, send_topic, send_data)

    def log_rs485_up_disable(self):
        if self.comboBox_log_sn.currentIndex() != -1:
            sn_cur = self.comboBox_log_sn.currentText()
            send_data = 'PT_CMD={"rs485_cfg": {"up_enable": 0}}'
            send_topic = "cpyypt/down/9002/" + sn_cur
            self.mqtt_send_data(g_client, send_topic, send_data)

    def map_refresh(self, sn=0, longitude=116.1798005, latitude=39.9030239):
        print("map_refresh")
        self.map_get_regeocode(sn, longitude, latitude)
        self.map_load_static_map(sn, longitude, latitude)

    def map_get_regeocode(self, sn=0, longitude=116.1798005, latitude=39.9030239, key='ebe21d6ee4409542d070b8be762ab0a5'):
        # print(longitude, latitude, key)
        # 构建请求URL
        url = f"https://restapi.amap.com/v3/geocode/regeo?key={key}&location={longitude},{latitude}"
        # 发送请求
        response = requests_get(url)
        # 解析JSON响应
        data = response.json()
        text = "信号基站位置信息："
        if data['status'] == '1':
            # 打印地理编码结果
            geo_result = data['regeocode']
            # print(f"地址: {geo_result['formatted_address']}")
            # print(f"省: {geo_result['addressComponent']['province']}")
            # print(f"市: {geo_result['addressComponent']['city']}")
            # print(f"区: {geo_result['addressComponent']['district']}")

            text = text + f"\n省: {geo_result['addressComponent']['province']}"
            text = text + f"\n市: {geo_result['addressComponent']['city']}"
            text = text + f"\n区: {geo_result['addressComponent']['district']}"
            text = text + f"\n地址: {geo_result['formatted_address']}"


        else:
            text = "逆向地理编码失败"
        print(text)
        self.textEdit_lsb_address.setText(text)


    def map_load_static_map(self, sn=0, longitude=116.1798005, latitude=39.9030239, key='ebe21d6ee4409542d070b8be762ab0a5'):
        # 高德地图URL，location为初始定位，可以是城市名或经纬度
        location_str = f'{longitude},{latitude}'
        #print(self, sn, longitude, latitude, key)
        #location_str = "116.1798005,39.9030239"
        url = f"https://restapi.amap.com/v3/geocode/regeo?key=ebe21d6ee4409542d070b8be762ab0a5&location=116.1798005,39.9030239"
        # url = f"https://restapi.amap.com/v3/geocode/regeo?output=xml&location=116.1798005,39.9030239&key=ebe21d6ee4409542d070b8be762ab0a5&radius=1000&extensions=all"
        url = f"https://restapi.amap.com/v3/staticmap?location={location_str}&zoom=13&size=1024*600&markers=mid,,A:{location_str}&key={key}"

        print(url)
        self.browser.load(QUrl(url))
        # self.setCentralWidget(self.browser)
        # self.browser.show()

    def lsb_get_fixed_info(self):
        sn_cur = ""
        send_cmd_flag = False
        if self.comboBox_log_sn.currentIndex() != -1:
            sn_cur = self.comboBox_log_sn.currentText()
        else:
            return

        if data_buf.get(sn_cur):
            if data_buf[sn_cur].get("pos_info"):
                lng = data_buf[sn_cur]["pos_info"].get("lng", 116.1666213)
                lat = data_buf[sn_cur]["pos_info"].get("lat", 39.9007607)
                self.map_refresh(sn_cur, lng, lat)
            else:
                send_cmd_flag = True
        else:
            send_cmd_flag = True


        if send_cmd_flag:
            send_topic = "cpyypt/down/9002/" + sn_cur
            send_data = 'PT_CMD={"get_fixed_info": "true"}'
            self.mqtt_send_data(g_client, send_topic, send_data)

    def set_svr(self):
        index_cur = self.comboBox_set_svr.currentIndex()
        if index_cur == 0:
            self.lineEdit_set_0ip.setText("test-******")
            self.lineEdit_set_1port.setText("******")
            self.lineEdit_set_2usr.setText("******")
            self.lineEdit_set_3password.setText("******")
            self.set_svr_env = "test"
        elif index_cur == 1:
            self.lineEdit_set_0ip.setText("prod-******")
            self.lineEdit_set_1port.setText("******")
            self.lineEdit_set_2usr.setText("******")
            self.lineEdit_set_3password.setText("******")
            self.set_svr_env = "prod"
        else:
            self.set_svr_env = "usr_def"

    def set_create(self):
        # PT_CMD={"imei_check":"869861069999337","reboot":"true","dev_cfg":{"dev_type":"9002","dev_sn":5,"dev_pd":"20240329"}}
        # PT_CMD={"reboot":"true","dev_cfg":{"dev_type":"9002","dev_sn":"0000000001","dev_pd":"20240322"},"reg_cfg":{"name":"李四","tel":"13999999999"}}
        set_dict = {}
        # 下发指令，是否绑定IMEI
        if self.checkBox_set_bind_imei.isChecked():
            set_dict["imei_check"] = self.lineEdit_0imei.text()
        else:
            set_dict.pop("imei_check", None)

        # 获取基础信息
        if self.checkBox_set_get_fixed_info.isChecked():
            set_dict["get_fixed_info"] = "true"
        else:
            set_dict.pop("get_fixed_info", None)

        # 恢复默认参数
        if self.checkBox_set_default_cfg.isChecked():
            set_dict["default_cfg"] = "true"
        else:
            set_dict.pop("default_cfg", None)

        # 重启
        if self.checkBox_set_reboot.isChecked():
            set_dict["reboot"] = "true"
        else:
            set_dict.pop("reboot", None)

        # 设备信息
        if self.checkBox_set_dev_cfg.isChecked():
            set_dict["dev_cfg"] = {}
            set_dict["dev_cfg"]["dev_sn"] = self.lineEdit_set_dev_sn.text()
            set_dict["dev_cfg"]["dev_pd"] = self.lineEdit_set_dev_pd.text()
        else:
            set_dict.pop("dev_cfg", None)

        # 注册信息
        if self.checkBox_set_reg_cfg.isChecked():
            set_dict["reg_cfg"] = {}
            set_dict["reg_cfg"]["name"] = self.lineEdit_set_reg_name.text()
            set_dict["reg_cfg"]["tel"] = self.lineEdit_set_reg_tel.text()
        else:
            set_dict.pop("reg_cfg", None)

        # 服务器信息
        if self.checkBox_set_svr_cfg.isChecked():
            set_dict["svr_cfg"] = {}
            set_dict["svr_cfg"]["svr_ip"] = self.lineEdit_set_svr_cfg.text()
            set_dict["svr_cfg"]["svr_port"] = self.lineEdit_set_svr_port.text()
            set_dict["svr_cfg"]["svr_usr_name"] = self.lineEdit_set_svr_usr_name.text()
            set_dict["svr_cfg"]["svr_usr_pwd"] = self.lineEdit_set_svr_usr_pwd.text()
        else:
            set_dict.pop("svr_cfg", None)

        # 接口参数
        io_name = self.comboBox_set_io_select.currentText() + "_cfg"
        if self.checkBox_set_io_cfg.isChecked():
            io_name = io_name.lower()
            set_dict[io_name] = {}
            if self.checkBox_set_io_up_enable.isChecked():
                set_dict[io_name]["up_enable"] = 1
            if self.checkBox_set_io_up_disable.isChecked():
                set_dict[io_name]["up_enable"] = 0
            if (not self.checkBox_set_io_up_enable.isChecked()) and (not self.checkBox_set_io_up_disable.isChecked()):
                set_dict[io_name].pop("up_enable", None)

            if self.checkBox_set_io_down_enable.isChecked():
                set_dict[io_name]["down_enable"] = 1
            if self.checkBox_set_io_down_disable.isChecked():
                set_dict[io_name]["down_enable"] = 0
            if (not self.checkBox_set_io_down_enable.isChecked()) and (not self.checkBox_set_io_down_disable.isChecked()):
                set_dict[io_name].pop("down_enable", None)

            if self.checkBox_set_io_parameter.isChecked():
                if self.lineEdit_set_baud_rate.text() != "":
                    set_dict[io_name]["baud_rate"]      = int(self.lineEdit_set_baud_rate.text())
                if self.lineEdit_set_data_bits.text() != "":
                    set_dict[io_name]["data_bits"]      = int(self.lineEdit_set_data_bits.text())
                if self.lineEdit_set_stop_bits.text() != "":
                    set_dict[io_name]["stop_bits"]      = int(self.lineEdit_set_stop_bits.text())
                if self.lineEdit_set_partiy.text() != "":
                    set_dict[io_name]["partiy"]         = int(self.lineEdit_set_partiy.text())
                '''
                if self.lineEdit_set_bit_order.text() != "":
                    set_dict[io_name]["bit_order"]      = int(self.lineEdit_set_bit_order.text())
                if self.lineEdit_set_buff_size.text() != "":
                    set_dict[io_name]["buff_size"]      = int(self.lineEdit_set_buff_size.text())
                if self.lineEdit_set_rs485_gpio.text() != "":
                    set_dict[io_name]["rs485_gpio"]     = int(self.lineEdit_set_rs485_gpio.text())
                if self.lineEdit_set_rs485_level.text() != "":
                    set_dict[io_name]["rs485_level"]    = int(self.lineEdit_set_rs485_level.text())
                if self.lineEdit_set_rs485_delay.text() != "":
                    set_dict[io_name]["rs485_delay"]    = int(self.lineEdit_set_rs485_delay.text())
                '''
        else:
            set_dict.pop(io_name, None)

        set_str = json_dumps(set_dict, ensure_ascii=False)
        print(set_str)
        if len(set_str) > 4:
            self.textEdit_set_tx.setText("PT_CMD=" + set_str)

    def set_send(self):
        print("mqtt set_send," + datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3])
        send_topic = self.lineEdit_5pub_topic.text()
        send_data = self.textEdit_set_tx.toPlainText()
        self.mqtt_send_data(g_client, send_topic, send_data)

    def set_create_send(self):
        self.set_create()
        self.set_send()

    def set_io_up_able(self):
        if self.checkBox_set_io_up_enable.isChecked():
            self.checkBox_set_io_up_disable.setChecked(False)
            self.checkBox_set_io_cfg.setChecked(True)
        if (not self.checkBox_set_io_parameter.isChecked()) and \
                (not self.checkBox_set_io_up_enable.isChecked()) and \
                (not self.checkBox_set_io_up_disable.isChecked()) and \
                (not self.checkBox_set_io_down_enable.isChecked()) and \
                (not self.checkBox_set_io_down_disable.isChecked()):
            self.checkBox_set_io_cfg.setChecked(False)

    def set_io_up_disable(self):
        if self.checkBox_set_io_up_disable.isChecked():
            self.checkBox_set_io_up_enable.setChecked(False)
            self.checkBox_set_io_cfg.setChecked(True)
        if (not self.checkBox_set_io_parameter.isChecked()) and \
                (not self.checkBox_set_io_up_enable.isChecked()) and \
                (not self.checkBox_set_io_up_disable.isChecked()) and \
                (not self.checkBox_set_io_down_enable.isChecked()) and \
                (not self.checkBox_set_io_down_disable.isChecked()):
            self.checkBox_set_io_cfg.setChecked(False)

    def set_io_down_able(self):
        if self.checkBox_set_io_down_enable.isChecked():
            self.checkBox_set_io_down_disable.setChecked(False)
            self.checkBox_set_io_cfg.setChecked(True)
        if (not self.checkBox_set_io_parameter.isChecked()) and \
                (not self.checkBox_set_io_up_enable.isChecked()) and \
                (not self.checkBox_set_io_up_disable.isChecked()) and \
                (not self.checkBox_set_io_down_enable.isChecked()) and \
                (not self.checkBox_set_io_down_disable.isChecked()):
            self.checkBox_set_io_cfg.setChecked(False)

    def set_io_down_disable(self):
        if self.checkBox_set_io_down_disable.isChecked():
            self.checkBox_set_io_down_enable.setChecked(False)
            self.checkBox_set_io_cfg.setChecked(True)
        if (not self.checkBox_set_io_parameter.isChecked()) and \
                (not self.checkBox_set_io_up_enable.isChecked()) and \
                (not self.checkBox_set_io_up_disable.isChecked()) and \
                (not self.checkBox_set_io_down_enable.isChecked()) and \
                (not self.checkBox_set_io_down_disable.isChecked()):
            self.checkBox_set_io_cfg.setChecked(False)

    def set_io_parameter(self):
        if self.checkBox_set_io_parameter.isChecked():
            self.checkBox_set_io_cfg.setChecked(True)
        if (not self.checkBox_set_io_parameter.isChecked()) and \
                (not self.checkBox_set_io_up_enable.isChecked()) and \
                (not self.checkBox_set_io_up_disable.isChecked()) and \
                (not self.checkBox_set_io_down_enable.isChecked()) and \
                (not self.checkBox_set_io_down_disable.isChecked()):
            self.checkBox_set_io_cfg.setChecked(False)

    def set_io_cfg(self):
        if not self.checkBox_set_io_parameter.isChecked():
            self.checkBox_set_io_parameter.setChecked(False)
            self.checkBox_set_io_up_enable.setChecked(False)
            self.checkBox_set_io_up_disable.setChecked(False)
            self.checkBox_set_io_down_enable.setChecked(False)
            self.checkBox_set_io_down_disable.setChecked(False)

    def set_default_tx_sn(self):
        if self.checkBox_set_default_tx_sn.isChecked():
            self.lineEdit_tx_sn.setEnabled(False)
            self.lineEdit_5pub_topic.setEnabled(False)
        else:
            self.lineEdit_tx_sn.setEnabled(True)
            self.lineEdit_5pub_topic.setEnabled(True)

    def set_tx_sn_change(self):
        sn = self.lineEdit_tx_sn.text()
        sn_str = sn.zfill(10)
        self.lineEdit_5pub_topic.setText("cpyypt/down/9002/" + sn_str)

def main():
    app = QApplication(sys.argv)  # 创建一个QApplication，也就是你要开发的软件app
    myWin = MyWindow()  # 创建一个QMainWindow，用来装载你需要的各种组件、控件，myWin()类的实例化对象
    myWin.show()  # 执行QMainWindow的show()方法，显示这个QMainWindow

    # 捕获并处理异常
    try:
        sys.exit(app.exec_())
        # sys.exit(app.exec_())  # 使用exit()或者点击关闭按钮退出QApplication
    except SystemExit as e:
        # 异常处理代码
        print(f"程序异常退出: {e}")
        # 这里可以添加额外的处理代码，例如保存状态、清理资源等

if __name__ == '__main__':
    main()