T1L_Config/SamplingDialog.py

import struct

import numpy as np
from PyQt5.QtCore import QRegExp, QByteArray
from PyQt5.QtGui import QRegExpValidator
from PyQt5.QtNetwork import QTcpSocket
from PyQt5.QtWidgets import (
    QDialog, QLineEdit, QMessageBox,
    QRadioButton, QButtonGroup, QFormLayout
)
from scipy.optimize import leastsq

HEADER_MAGIC = bytes([0xAA, 0x55, 0xAA])
HEADER_SIZE = 6  # PackgeHead除去data字段的大小(3+1+1+1)


class SamplingDialog(QDialog):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.setWindowTitle("采样参数设置")

        self.rate_combo = QComboBox()
        self.rate_combo.addItems([
            "8000", "16000", "24000", "32000", "48000", "96000"
        ])

        self.time_input = QLineEdit()
        self.time_input.setPlaceholderText("输入采样时间（秒）")

        self.ok_button = QPushButton("确定")
        self.ok_button.clicked.connect(self.accept)

        form_layout = QFormLayout()
        form_layout.addRow("采样率（Hz）", self.rate_combo)
        form_layout.addRow("采样时间（秒）", self.time_input)

        layout = QVBoxLayout()
        layout.addLayout(form_layout)
        layout.addWidget(self.ok_button)

        self.setLayout(layout)

    def get_values(self):
        # 返回采样率编号（整数）和采样时间（字符串或整数）
        rate_code_map = {
            "8000": 0x00,
            "16000": 0x01,
            "24000": 0x02,
            "32000": 0x03,
            "48000": 0x04,
            "96000": 0x05
        }
        rate_str = self.rate_combo.currentText()
        rate_code = rate_code_map.get(rate_str, 0x00)
        return rate_code, self.time_input.text()


class PackageHead:
    @staticmethod
    def parse(data: bytes):
        if len(data) < 6:
            return None
        head = data[0:3]
        if head != b'\xAA\x55\xAA':
            return None
        cmd, ver, result_code = data[3:6]
        return {
            'cmd': cmd,
            'version': ver,
            'result_code': result_code,
            'data': data[6:]
        }


class GetCoe:
    def __init__(self, temp_ch1_1, temp_ch1_2, dc_ch1_1, dc_ch1_2, ac_ch1_1, ac_ch1_2, temp_ch2_1, temp_ch2_2, dc_ch2_1,
                 dc_ch2_2,
                 ac_ch2_1, ac_ch2_2):
        self.temp_ch1_1 = temp_ch1_1
        self.temp_ch1_2 = temp_ch1_2
        self.dc_ch1_1 = dc_ch1_1
        self.dc_ch1_2 = dc_ch1_2
        self.ac_ch1_1 = ac_ch1_1
        self.ac_ch1_2 = ac_ch1_2
        self.temp_ch2_1 = temp_ch2_1
        self.temp_ch2_2 = temp_ch2_2
        self.dc_ch2_1 = dc_ch2_1
        self.dc_ch2_2 = dc_ch2_2
        self.ac_ch2_1 = ac_ch2_1
        self.ac_ch2_2 = ac_ch2_2


class Eigenvalue:
    def __init__(self, temp1, temp2, offset1, offset2):
        self.temp1 = temp1
        self.temp2 = temp2
        self.offset1 = offset1
        self.offset2 = offset2


class MacConfigDialog(QDialog):
    def __init__(self, socket: QTcpSocket, parent=None):
        super().__init__(parent)
        self.setWindowTitle("配置 MAC")
        self.socket = socket
        self.setFixedWidth(300)
        self.mac_input1 = QLineEdit()
        self.mac_input2 = QLineEdit()
        hex_reg = QRegExp("[0-9A-Fa-f]{1,2}")
        validator = QRegExpValidator(hex_reg)
        self.mac_input1.setValidator(validator)
        self.mac_input2.setValidator(validator)
        self.save_btn = QPushButton("保存")
        self.cancel_btn = QPushButton("取消")

        mac_line = QHBoxLayout()
        mac_line.addWidget(QLabel("MAC 地址: 50:29:4D:20"))
        mac_line.addWidget(self.mac_input1)
        mac_line.addWidget(QLabel(":"))
        mac_line.addWidget(self.mac_input2)

        btn_line = QHBoxLayout()
        btn_line.addWidget(self.save_btn)
        btn_line.addWidget(self.cancel_btn)

        main_layout = QVBoxLayout()
        main_layout.addLayout(mac_line)
        main_layout.addLayout(btn_line)
        self.setLayout(main_layout)

        self.save_btn.clicked.connect(self.send_mac_config)
        self.cancel_btn.clicked.connect(self.reject)

    def send_mac_config(self):
        try:
            mac1 = int(self.mac_input1.text(), 16)
            mac2 = int(self.mac_input2.text(), 16)
        except ValueError:
            QMessageBox.warning(self, "输入错误", "请输入合法的十六进制数")
            return

        packet = bytes([
            0xAA, 0x55, 0xAA,
            0x0D,  # cmd
            0x01,  # version
            0x00  # result_code (发送时为0)
        ]) + struct.pack('2B', mac1, mac2)

        self.socket.write(packet)
        if not self.socket.waitForBytesWritten(1000):
            QMessageBox.critical(self, "错误", "发送失败")
            return

        if not self.socket.waitForReadyRead(3000):
            QMessageBox.critical(self, "超时", "未收到回应")
            return

        data = bytes(self.socket.readAll())
        pkg = PackageHead.parse(data)

        if not pkg:
            QMessageBox.critical(self, "错误", "返回数据格式无效")
            return

        if pkg['result_code'] == 0x00:
            QMessageBox.information(self, "成功", "配置成功")
            self.accept()
        else:
            QMessageBox.warning(self, "失败", f"错误码: {pkg['result_code']}")


class IPConfigDialog(QDialog):
    def __init__(self, socket: QTcpSocket, parent=None):
        super().__init__(parent)
        self.setWindowTitle("配置 IP")
        self.socket = socket
        self.setFixedWidth(300)
        self.setFixedHeight(150)
        self.dhcp_radio = QRadioButton("DHCP")
        self.static_radio = QRadioButton("Static")
        self.dhcp_radio.setChecked(True)
        radio_layout = QHBoxLayout()
        radio_layout.addWidget(self.dhcp_radio)
        radio_layout.addWidget(self.static_radio)

        self.radio_group = QButtonGroup()
        self.radio_group.addButton(self.dhcp_radio)
        self.radio_group.addButton(self.static_radio)

        # IPv4地址输入框及标签
        self.ip_input = QLineEdit()
        self.mask_input = QLineEdit()
        self.gw_input = QLineEdit()

        # IP地址校验器，限制0-255的IPv4格式
        ip_regex = QRegExp(r"^(25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)"
                           r"\.(25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)"
                           r"\.(25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)"
                           r"\.(25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)$")
        ip_validator = QRegExpValidator(ip_regex)
        self.ip_input.setValidator(ip_validator)
        self.mask_input.setValidator(ip_validator)
        self.gw_input.setValidator(ip_validator)

        form_layout = QFormLayout()
        form_layout.setLabelAlignment(Qt.AlignRight)

        form_layout.addRow(QLabel("IP 地址:"), self.ip_input)
        form_layout.addRow(QLabel("子网掩码:"), self.mask_input)
        form_layout.addRow(QLabel("网关:"), self.gw_input)
        self.ip_input.setText("192.168.0.151")
        self.mask_input.setText("255.255.255.0")
        self.gw_input.setText("192.168.0.1")

        self.save_btn = QPushButton("保存")
        self.cancel_btn = QPushButton("取消")
        btn_line = QHBoxLayout()
        btn_line.addWidget(self.save_btn)
        btn_line.addWidget(self.cancel_btn)

        main_layout = QVBoxLayout()
        main_layout.addLayout(radio_layout)
        main_layout.addLayout(form_layout)
        main_layout.addLayout(btn_line)
        self.setLayout(main_layout)

        self.save_btn.clicked.connect(self.send_ip_config)
        self.cancel_btn.clicked.connect(self.reject)

        # DHCP模式下禁用IP等输入
        self.dhcp_radio.toggled.connect(self.update_input_enabled)
        self.update_input_enabled()

    def update_input_enabled(self):
        enabled = self.static_radio.isChecked()
        self.ip_input.setEnabled(enabled)
        self.mask_input.setEnabled(enabled)
        self.gw_input.setEnabled(enabled)

    def send_ip_config(self):
        mode = 0 if self.dhcp_radio.isChecked() else 1

        def parse_ip(ip_str):
            parts = ip_str.strip().split('.')
            if len(parts) != 4:
                raise ValueError()
            return [int(p) for p in parts]

        if mode == 0:
            packet = bytes([
                0xAA, 0x55, 0xAA,
                0x0E,  # cmd
                0x01,  # version
                0x00  # result_code (发送时为0)
            ]) + struct.pack('B', mode)
        else:
            try:
                ip = parse_ip(self.ip_input.text())
                mask = parse_ip(self.mask_input.text())
                gw = parse_ip(self.gw_input.text())
            except ValueError:
                QMessageBox.warning(self, "输入错误", "请输入合法的 IPv4 地址格式")
                return
            packet = bytes([
                0xAA, 0x55, 0xAA,
                0x0E,  # cmd
                0x01,  # version
                0x00  # result_code (发送时为0)
            ]) + struct.pack('B', mode) + bytes(ip) + bytes(mask) + bytes(gw)

        self.socket.write(packet)
        if not self.socket.waitForBytesWritten(1000):
            QMessageBox.critical(self, "错误", "发送失败")
            return

        if not self.socket.waitForReadyRead(3000):
            QMessageBox.critical(self, "超时", "未收到回应")
            return

        data = bytes(self.socket.readAll())
        pkg = PackageHead.parse(data)

        if not pkg:
            QMessageBox.critical(self, "错误", "返回数据格式无效")
            return

        if pkg['result_code'] == 0x00:
            QMessageBox.information(self, "成功", "配置成功")
            self.accept()
        else:
            QMessageBox.warning(self, "失败", f"错误码: {pkg['result_code']}")


from PyQt5.QtWidgets import (
    QDialog, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QComboBox,
    QGridLayout
)
from PyQt5.QtCore import Qt


class CalibrationDialog(QDialog):
    def __init__(self, socket: QTcpSocket, parent=None):
        super().__init__(parent)
        self.setWindowTitle("校准设置")
        self.socket = socket
        self.type_index = 0
        self.setFixedWidth(300)
        # 通道与类型选择
        self.partial_data = QByteArray()
        self.recv_state = ''
        self.channel_cb = QComboBox()
        self.channel_cb.addItems(["通道1", "通道2"])

        self.type_cb = QComboBox()
        self.type_cb.addItems(["温度", "电压DC", "电压AC"])
        self.type_cb.currentTextChanged.connect(self.update_voltage_points)
        self.channel_cb.currentTextChanged.connect(self.update_voltage_points)

        top_layout = QHBoxLayout()
        top_layout.addWidget(QLabel("通道:"))
        top_layout.addWidget(self.channel_cb)
        top_layout.addWidget(QLabel("类型:"))
        top_layout.addWidget(self.type_cb)

        # 电压点信息
        self.voltage_labels = []
        self.value_labels = []
        self.get_buttons = []
        self.voltage_points = []

        grid = QGridLayout()
        grid.addWidget(QLabel("电压点"), 0, 0)
        grid.addWidget(QLabel("校准值"), 0, 1)
        grid.addWidget(QLabel("操作按钮"), 0, 2)

        for i in range(3):
            v_label = QLabel()
            self.voltage_labels.append(v_label)

            val_label = QLabel("未获取")
            self.value_labels.append(val_label)

            btn = QPushButton("获取")
            btn.clicked.connect(lambda _, index=i: self.fetch_voltage(index))
            self.get_buttons.append(btn)

            grid.addWidget(v_label, i + 1, 0)
            grid.addWidget(val_label, i + 1, 1)
            grid.addWidget(btn, i + 1, 2)

        self.update_voltage_points()

        # 初始化与校准按钮
        self.init_btn = QPushButton("初始化")
        self.calibrate_btn = QPushButton("校准")
        self.get_coe_btn = QPushButton("获取系数")
        self.init_btn.clicked.connect(self.initialize_calibration)
        self.calibrate_btn.clicked.connect(self.calibrate)
        self.get_coe_btn.clicked.connect(self.get_coe)
        self.socket.readyRead.connect(self.on_ready_read)

        btn_layout = QHBoxLayout()
        btn_layout.addStretch()
        btn_layout.addWidget(self.init_btn)
        btn_layout.addWidget(self.calibrate_btn)
        btn_layout.addWidget(self.get_coe_btn)

        main_layout = QVBoxLayout()
        main_layout.addLayout(top_layout)
        main_layout.addLayout(grid)
        main_layout.addLayout(btn_layout)

        self.setLayout(main_layout)

    def update_voltage_points(self):
        type_text = self.type_cb.currentText()
        if type_text == "温度":
            self.voltage_points = [0.5, 1.5, 2.5]
        elif type_text == "电压DC":
            self.voltage_points = [1, 8, 15]
        else:
            self.voltage_points = [1, 5, 9]

        for i, v in enumerate(self.voltage_points):
            self.voltage_labels[i].setText(f"{v:.1f}V")
            self.value_labels[i].setText("未获取")

    def fetch_voltage(self, index):
        # 模拟读取值
        # import random
        # val = round(self.voltage_points[index] + random.uniform(-0.05, 0.05), 3)
        # self.value_labels[index].setText(str(val))
        self.type_index = index
        if self.type_cb.currentIndex() != 2:
            self.get_temp_dc()
        else:
            self.get_wave()

    def initialize_calibration(self):
        for label in self.value_labels:
            label.setText("未获取")
        # 获取通道号和类型索引
        ch = self.channel_cb.currentIndex() + 1  # 通道1 对应 1
        group = self.type_cb.currentIndex()  # 0=temp, 1=dc, 2=ac
        print(f"ch:{ch},group:{group}")
        a = 1
        b = 0
        header = bytes([0xAA, 0x55, 0xAA, 0x11, 0x01, 0x00])  # 可根据实际协议修改
        payload = struct.pack('<BBff', ch, group, a, b)
        packet = header + payload
        # 发送给 MCU
        self.socket.write(packet)
        self.socket.waitForReadyRead()

    def get_coe(self):
        try:
            self.socket.write(bytes(
                [0xAA, 0x55, 0xAA, 0x10, 0x01, 0x00]))  # 发送数据
            self.socket.waitForReadyRead()


        except Exception as e:
            self.status_bar.showMessage(f"状态: 错误 - {str(e)}")

    def calibrate(self):
        try:
            self.recv_state = ''
            x = self.voltage_points
            y = [float(label.text()) for label in self.value_labels]
            if any(label.text() == "未获取" for label in self.value_labels):
                raise ValueError("未获取所有电压点数据")
        except ValueError:
            from PyQt5.QtWidgets import QMessageBox
            QMessageBox.warning(self, "错误", "请先获取所有电压点的校准值")
            return

        # 非线性最小二乘拟合 y = a*x + b
        def func(p, x):
            a, b = p
            return a * x + b

        def error(p, x, y):
            return func(p, x) - y

        p0 = [1, 0]
        plsq = leastsq(error, p0, args=(np.array(y), np.array(x)))
        a, b = plsq[0]
        # import numpy as np
        #
        # a, b = np.polyfit(x, y, 1)

        print(f"拟合结果: a = {a:.4f}, b = {b:.4f}")

        # 获取通道号和类型索引
        ch = self.channel_cb.currentIndex() + 1  # 通道1 对应 1
        group = self.type_cb.currentIndex()  # 0=temp, 1=dc, 2=ac
        print(f"ch:{ch},group:{group}")

        header = bytes([0xAA, 0x55, 0xAA, 0x11, 0x01, 0x00])  # 可根据实际协议修改
        payload = struct.pack('<BBff', ch, group, a, b)
        packet = header + payload

        # 发送给 MCU
        self.socket.write(packet)
        self.socket.waitForReadyRead()

    def get_extremes(self, data):
        """获取最大和最小的10个值（排序法）"""
        sorted_data = np.sort(data)
        return {
            'top10_max': sorted_data[-10:][::-1].tolist(),  # 降序排列
            'top10_min': sorted_data[:10].tolist()
        }

    def mean_without_max_optimized(self, data):
        """优化内存使用的版本"""
        arr = np.array(data)
        if len(arr) < 2:
            return np.nan

        max_val = np.max(arr)
        # 计算总和和计数时直接排除最大值
        total = np.sum(arr) - max_val
        count = len(arr) - 1
        return total / count

    def mean_without_min_optimized(self, data):
        """优化内存使用的版本"""
        arr = np.array(data)
        if len(arr) < 2:
            return np.nan

        min_val = np.min(arr)
        # 计算总和和计数时直接排除最大值
        total = np.sum(arr) - min_val
        count = len(arr) - 1
        return total / count

    def get_temp_dc(self):
        try:
            self.recv_state = ''
            self.socket.write(bytes(
                [0xAA, 0x55, 0xAA, 0x0B, 0x01, 0x00]))  # 发送数据
            self.socket.waitForReadyRead()
        except Exception as e:
            QMessageBox.warning(self, "错误", f"{str(e)}")

    def get_wave(self):
        try:
            self.recv_state = 'WAIT_HEADER'
            rate_bytes = struct.pack('<I', 4)
            time_bytes = struct.pack('<I', 1)
            ch = self.channel_cb.currentIndex() + 1
            packet = bytes([
                0xAA, 0x55, 0xAA,
                0x01, 0x01, 0x00,
                ch
            ]) + rate_bytes + time_bytes

            self.socket.write(packet)
            print(packet)
            self.socket.waitForReadyRead()
        except Exception as e:
            QMessageBox.warning(self, "错误", f"{str(e)}")

    def on_ready_read(self):
        while self.socket.bytesAvailable():
            if self.recv_state == 'WAIT_HEADER':
                if self.socket.bytesAvailable() >= HEADER_SIZE:
                    self.header = self.socket.read(HEADER_SIZE)
                    if self.header[:3] != HEADER_MAGIC:
                        print("无效 header magic，跳过")
                        continue
                    self.cmd = self.header[3]
                    if self.cmd == 0x01:
                        self.recv_state = 'WAIT_WAVE_HEADER'
                    else:
                        print(f"未知或暂不处理的 cmd: {self.cmd}")
                        self.recv_state = 'WAIT_HEADER'
                        return
                else:
                    return
            elif self.recv_state == 'WAIT_WAVE_HEADER':
                if self.socket.bytesAvailable() >= 5:
                    wave_header = self.socket.read(5)
                    self.sampling_rate = struct.unpack('<i', wave_header[1:5])[0]
                    self.expected_length = self.sampling_rate * 4
                    self.partial_data.clear()
                    self.recv_state = 'WAIT_WAVE_DATA'
                else:
                    return
            elif self.recv_state == 'WAIT_WAVE_DATA':
                bytes_needed = self.expected_length - len(self.partial_data)
                chunk = self.socket.read(min(self.socket.bytesAvailable(), bytes_needed))
                self.partial_data += chunk
                if len(self.partial_data) >= self.expected_length:
                    self.process_wave_packet(bytes(self.partial_data))
                    self.recv_state = 'WAIT_HEADER'
                    self.partial_data.clear()
            else:
                if self.socket.bytesAvailable() >= HEADER_SIZE:
                    recv_data = self.socket.read(1340)
                    if recv_data[:3] != HEADER_MAGIC:
                        print("无效 header magic，跳过")
                        continue
                    cmd = recv_data[3]
                    if cmd == 0x0C:
                        body_format = '< H H H H'
                        body_size = struct.calcsize(body_format)
                        body_data = recv_data[HEADER_SIZE:HEADER_SIZE + body_size]
                        unpacked_data = struct.unpack(body_format, body_data)
                        value = Eigenvalue(*unpacked_data)
                        print(
                            f"温度1：{value.temp1 / 1000},温度2：{value.temp2 / 1000},偏置电压1：{value.offset1 / 100},偏置电压2：{value.offset2 / 100}")
                        if self.channel_cb.currentIndex() == 0 and self.type_cb.currentIndex() == 0:
                            self.value_labels[self.type_index].setText(str(value.temp1 / 1000))
                        elif self.channel_cb.currentIndex() == 0 and self.type_cb.currentIndex() == 1:
                            self.value_labels[self.type_index].setText(str(value.offset1 / 100))
                        elif self.channel_cb.currentIndex() == 1 and self.type_cb.currentIndex() == 0:
                            self.value_labels[self.type_index].setText(str(value.temp2 / 1000))
                        elif self.channel_cb.currentIndex() == 1 and self.type_cb.currentIndex() == 1:
                            self.value_labels[self.type_index].setText(str(value.offset2 / 100))
                    elif cmd == 0x11:
                        pkg = PackageHead.parse(recv_data)
                        if not pkg:
                            QMessageBox.critical(self, "错误", "返回数据格式无效")
                            return
                        if pkg['result_code'] == 0x00:
                            QMessageBox.information(self, "成功", "配置成功")
                            return
                    elif cmd == 0x10:
                        body_format = '< f f f f f f f f f f f f'
                        body_size = struct.calcsize(body_format)
                        body_data = recv_data[HEADER_SIZE:HEADER_SIZE + body_size]
                        unpacked_data = struct.unpack(body_format, body_data)
                        value = GetCoe(*unpacked_data)
                        print(
                            f"{value.temp_ch1_1, value.temp_ch1_2, value.dc_ch1_1, value.dc_ch1_2, value.ac_ch1_1, value.ac_ch1_2}")
                        print(
                            f"{value.temp_ch2_1, value.temp_ch2_2, value.dc_ch2_1, value.dc_ch2_2, value.ac_ch2_1, value.ac_ch2_2}")

    def process_wave_packet(self, wave_data):
        data = wave_data  # 接收所有数据
        data = np.frombuffer(data, dtype=np.int32)  # 根据实际数据格式转换
        LSB_32BIT = (2.8 / (2 ** 31)) * ((750 + 287) / 287) * 1000 / 10.2
        self.scaled_data = data * LSB_32BIT

        result = self.get_extremes(self.scaled_data)
        print("最大的10个值:", result['top10_max'])
        print("最小的10个值:", result['top10_min'])
        mean_max = self.mean_without_max_optimized(result['top10_max'])
        print(f"top10_max 去除最大的数据后的平均值1：{mean_max}")
        mean_min = self.mean_without_min_optimized(result['top10_min'])
        print(f"top10_min 去除最大的数据后的平均值2：{mean_min}")
        pp = mean_max - mean_min
        if self.type_cb.currentIndex() == 2:
            self.value_labels[self.type_index].setText(str(round(pp, 6)))
        print(f"pp :{mean_max - mean_min}")

    def closeEvent(self, event):
        try:
            print("closeEvent")
            self.socket.readyRead.disconnect(self.on_ready_read)
        except TypeError:
            pass
        super().closeEvent(event)