python/LaserMehtanePT_RB/pythonProject-v2/calibration_manager.py

import logging
import time
from PyQt5.QtCore import QTimer, pyqtSignal, QObject
from PyQt5.QtGui import QColor


class CalibrationManager(QObject):
    query_finished = pyqtSignal(str, bytes)

    # 定义常量
    WRITE_SINGLE_REGISTER = 0x06
    READ_HOLDING_REGISTERS = 0x03
    QUERY_INTERVAL = 1000  # 查询间隔时间，单位为毫秒

    def __init__(self, serial_manager, data_processor, ui):
        super().__init__()
        self.serial_manager = serial_manager
        self.data_processor = data_processor
        self.ui = ui

        self.num = 0
        self.modbus_data = [0] * 20
        self.query_modbus_bytes = None
        self.query_timer = QTimer(self.ui)
        self.query_timer.timeout.connect(self.check_query_value)  # 定时查询
        self.query_finished.connect(self.process_query_value)     # 处理数据

    def write_plain_text_edit(self, text, log_level='info'):
        """统一处理 plainTextEditStastusCalib 的文本写入和日志记录"""
        self.ui.plainTextEditStastusCalib.appendPlainText(text)
        if log_level == 'info':
            logging.info(text)
        elif log_level == 'error':
            logging.error(text)
        elif log_level == 'debug':
            logging.debug(text)

    def calibrate_air_tempera(self, temp_str):
        """开始温度标定"""
        self.write_plain_text_edit('################\r\n环境温度，开始标定')
        if not temp_str.isdigit():
            self.write_plain_text_edit('环境温度输入错误', log_level='error')
            return
        temp_int = int(float(temp_str) * 100)
        temp_hex = temp_int.to_bytes(2, 'big')
        self.write_register(0x01, 0x2C, temp_hex)
        time.sleep(1)
        self.check_air_tempera()

    def check_air_tempera(self):
        self.reset_port_colors()
        self.write_plain_text_edit('################\r\n空气温度标定状态：')
        self.read_register(0x01, 0x2C, 0x00, 0x01)

    def calibrate_laser_tempra(self):
        self.write_plain_text_edit('################')
        self.write_plain_text_edit('laser温度标定')
        self.write_register(0x01, 0x2D, b'\x00\x01')
        time.sleep(1)
        self.check_laser_tempra()

    def stop_calibrate_laser_tempra(self):
        self.write_plain_text_edit('################')
        self.write_register(0x01, 0x2D, b'\x00\x00')
        self.write_plain_text_edit('laser温度标定结束')

    def check_laser_tempra(self):
        self.reset_port_colors()
        self.write_plain_text_edit('查询中...')
        self.read_register(0x01, 0x2D, 0x00, 0x01)

    def check_work_tempra(self):
        self.reset_port_colors()
        self.write_plain_text_edit('查询当前激光工作点')
        self.read_register(0x00, 0x01, 0x00, 0x01)

    def calibrate_laser_tempera_start(self, temp_str):
        self.write_plain_text_edit('################')
        self.write_plain_text_edit('laser温度标定起始点修改')
        if not temp_str.isdigit():
            self.write_plain_text_edit('环境温度输入错误')
            return
        temp_int = int(float(temp_str) * 100)
        temp_hex = temp_int.to_bytes(2, 'big')
        self.write_register(0x00, 0x16, temp_hex)
        time.sleep(1)
        self.check_laser_tempera_start()

    def check_laser_tempera_start(self):
        self.reset_port_colors()
        self.write_plain_text_edit('################')
        self.write_plain_text_edit('laser温度标定起始点查询')
        self.read_register(0x00, 0x16, 0x00, 0x01)

    def calibrate_conc_ppm(self, ppm_str):
        self.write_plain_text_edit('################')
        self.write_plain_text_edit('浓度标定')
        if not ppm_str.isdigit():
            self.write_plain_text_edit('浓度输入错误')
            return
        ppm_int = int(float(ppm_str))
        ppm_hex = ppm_int.to_bytes(2, 'big')
        self.write_register(0x01, 0x2E, ppm_hex)
        time.sleep(1)
        self.check_conc_ppm()

    def check_conc_ppm(self):
        self.reset_port_colors()
        self.write_plain_text_edit('################')
        self.write_plain_text_edit('浓度标定,查询')
        self.read_register(0x01, 0x2E, 0x00, 0x01)

    def calibrate_reset_conc_ppm(self):
        self.write_plain_text_edit('################')
        self.write_plain_text_edit('浓度标定，重置')
        self.write_register(0x01, 0x2F, b'\x00\x00')
        time.sleep(1)
        self.check_reset_conc_ppm()

    def check_reset_conc_ppm(self):
        self.reset_port_colors()
        self.write_plain_text_edit('################')
        self.write_plain_text_edit('浓度标定，重置')
        self.read_register(0x00, 0x64, 0x00, 0x01)

    def set_out_hex_length(self,length_str):
        """设置输出的hex长度"""
        self.write_plain_text_edit('设置输出的hex长度')
        if not length_str.isdigit():
            self.write_plain_text_edit('参数输入错误')
            return
        length_int = int(float(length_str))
        length_hex = length_int.to_bytes(2, 'big')
        self.write_register(0x00, 0x17, length_hex)
        time.sleep(1)
        self.get_out_hex_length()
    def get_out_hex_length(self):
        self.reset_port_colors()
        self.write_plain_text_edit('查询输出的hex长度')
        self.read_register(0x00, 0x17, 0x00, 0x01)


    def write_register(self, address_hi, address_lo, value):
        """写入寄存器"""
        self.modbus_data[0:6] = [0x01, self.WRITE_SINGLE_REGISTER, address_hi, address_lo, value[0],
                                 value[1]]
        self.send_bytes(bytes(self.modbus_data[:6]))

    def read_register(self, address_hi, address_lo, add_num_hi, add_num_lo):
        """读取寄存器"""
        self.modbus_data[0:6] = [0x01, self.READ_HOLDING_REGISTERS, address_hi, address_lo, add_num_hi, add_num_lo]
        self.query_modbus_bytes = bytes(self.modbus_data[:6])

        self.send_bytes(self.query_modbus_bytes)
        self.write_plain_text_edit('查询中...')
        self.num = 0
        self.start_query_timer()

    def send_bytes(self, modbus_bytes):
        crc = self.data_processor.crc_rtu(modbus_bytes)
        modbus_bytes = modbus_bytes + crc
        modbus_send_str = self.data_processor.format_bytes_to_hexstr_space(modbus_bytes)
        self.ui.data_send(modbus_send_str)

    def validate_response(self, buffer):
        """验证响应"""
        self.write_plain_text_edit(f'开始捕获：{buffer.hex().upper()}')
        start_index = buffer.find(b'\x01\x03')
        if start_index != -1 and len(buffer) >= start_index + 5:
            self.write_plain_text_edit(f'捕获成功，开始验证：{buffer.hex().upper()}')
            byte_count = buffer[start_index + 2]
            frame_length = start_index + 5 + byte_count
            if len(buffer) >= frame_length:
                crc = self.data_processor.crc_rtu(buffer[start_index:frame_length - 2])
                if buffer[frame_length - 2:frame_length] == crc:
                    self.write_plain_text_edit('验证成功')
                    return buffer[start_index:frame_length]
                else:
                    self.write_plain_text_edit('验证失败')
                    self.write_plain_text_edit('捕获失败')
                    return False

    def check_query_value(self):
        self.num += 1
        self.write_plain_text_edit(f'执行查询次数：{self.num}', log_level='debug')
        if self.num > 10:
            self.num = 0
            timer_status = self.stop_query()
            self.write_plain_text_edit('查询超时')
            self.set_cursor_to_end()
            return False

        channel_ports = self.ui.find_selected_ser_ports()
        channel_nets = self.ui.find_selected_net_ports()
        if channel_ports is None:
            channel_ports = []
        if channel_nets is None:
            channel_nets = []

        if len(channel_ports) == 0 and len(channel_nets) == 0:
            self.stop_query()
            self.write_plain_text_edit('查询结束，在通道处查看结果')
            # 如果只有一个通道，重新勾选
            channel_all_nets = self.ui.find_all_net_ports()
            channel_all_ports = self.ui.find_all_ser_ports()
            if len(channel_all_nets) == 1:
                self.ui.select_net_channel_single(channel_all_nets[0], 2)
            if len(channel_all_ports) == 1:
                self.ui.select_ser_channel_single(channel_all_ports[0], 2)

            return False

        for port_name, buffer in self.ui.buffers.items():
            if port_name in channel_ports or port_name in channel_nets:
                buffer_valid = self.validate_response(buffer)
                if buffer_valid:
                    self.write_plain_text_edit('发射信号')
                    self.query_finished.emit(port_name, buffer_valid)
                    self.ui.buffers[port_name] = b''


    def process_query_value(self, port_name, buffer):
        """处理查询数据"""
        self.write_plain_text_edit(f'数据通道：{port_name}，数据： {buffer.hex().upper()}', log_level='debug')
        if len(buffer) > 6:
            status = self.parse_query_value(self.query_modbus_bytes, buffer)
            ind = self.ui.find_port_ser_row(port_name)
            ind_net = self.ui.find_port_net_row(port_name)

            if status:
                if ind is not None:
                    self.ui.update_ser_table_status(ind, 2, status[1], status[2])
                    self.write_plain_text_edit(f'{port_name}：{status[0]}')
                    self.ui.select_ser_channel_single(port_name, 0)
                if ind_net is not None:
                    self.ui.update_net_table_status(ind_net, 2, status[1], status[2])
                    self.write_plain_text_edit(f'{port_name}：{status[0]}')
                    self.ui.select_net_channel_single(port_name, 0)
            else:
                self.write_plain_text_edit(f'{port_name}：查询失败')

    def parse_query_value(self, modbus_bytes, buffer):
        """解析查询结果"""
        if self.query_modbus_bytes in [
            b'\x01\x03\x01\x2C\x00\x01',  # 查询环境温度标定
            b'\x01\x03\x01\x2D\x00\x01',  # 查询激光温度标定
            b'\x01\x03\x01\x2E\x00\x01'  # 查询浓度标定
        ]:
            status_map = {
                b'\x01\x03\x02\x00\x00': ('未标定', '未标定', QColor(255, 255, 255)),
                b'\x01\x03\x02\x00\x01': ('标定中', '标定中', QColor(255, 255, 0)),
                b'\x01\x03\x02\x00\x02': ('标定成功', '标定成功', QColor(0, 255, 0)),
                b'\x01\x03\x02\x00\x03': ('标定失败', '标定失败', QColor(255, 0, 0)),
                b'\x01\x03\x02\x00\x04': ('标定失败,浓度太低', '浓度太低', QColor(255, 0, 0)),
                b'\x01\x03\x02\x00\x05': ('标定失败，间距太小', '标定间隔太小', QColor(255, 0, 0))
            }
            for key, value in status_map.items():
                if key in buffer:
                    self.write_plain_text_edit(f'标定识别,查询指令：{key.hex().upper()} ，查询结果{buffer.hex().upper()}')
                    return value

            return None

        elif modbus_bytes == b'\x01\x03\x00\x16\x00\x01':
            bytes_count = buffer[2]
            query_data = buffer[3:3 + bytes_count]
            air_tempera = int.from_bytes(query_data, 'big') / 100
            return f'环境温度：查询结果：{air_tempera}', f'{air_tempera}', QColor(0, 255, 0)
        elif modbus_bytes == b'\x01\x03\x01\x2F\x00\x01':
            bytes_count = buffer[2]
            query_data = buffer[3:3 + bytes_count]
            conc_flag = int.from_bytes(query_data, 'big')
            return f'浓度标定：重置结果：{conc_flag}', f'{conc_flag}', QColor(0, 255, 0)
        elif modbus_bytes == b'\x01\x03\x00\x64\x00\x01':
            bytes_count = buffer[2]
            query_data = buffer[3:3 + bytes_count]
            conc_calib_flag = int.from_bytes(query_data, 'big')
            return f'浓度标定：查询结果：{conc_calib_flag}', f'{conc_calib_flag}', QColor(0, 255, 0)
        elif modbus_bytes == b'\x01\x03\x00\x01\x00\x01':
            bytes_count = buffer[2]
            query_data = buffer[3:3 + bytes_count]
            laser_work_tempera = int.from_bytes(query_data, 'big') / 100
            return f'激光温度：查询结果：{laser_work_tempera}', f'{laser_work_tempera}', QColor(0, 255, 0)
        elif modbus_bytes == b'\x01\x03\x00\x17\x00\x01':
            bytes_count = buffer[2]
            query_data = buffer[3:3 + bytes_count]
            laser_work_tempera = int.from_bytes(query_data, 'big')
            return f'激光温度：查询结果：{laser_work_tempera}', f'{laser_work_tempera}', QColor(0, 255, 0)

        else:
            return None

    def start_query_timer(self):
        """启动查询定时器"""
        if self.query_timer.isActive():
            self.query_timer.stop()
        self.query_timer.start(self.QUERY_INTERVAL)
        self.write_plain_text_edit('timer启动', log_level='debug')

    def stop_query(self):
        """停止查询定时器"""
        if self.query_timer and self.query_timer.isActive():
            self.query_timer.stop()
            self.write_plain_text_edit('timer已经关闭')
            return 'timer已经关闭'

    def reset_port_colors(self):
        """重置串口通道颜色"""
        for port_name in self.serial_manager.buffers.keys():
            ind = self.ui.find_port_ser_row(port_name)
            if ind is not None:
                self.ui.reset_ser_port_color(ind)

        selected_channels = self.ui.find_selected_net_ports()
        for channel_name in selected_channels:
            ind = self.ui.find_port_net_row(channel_name)
            if ind is not None:
                self.ui.reset_net_port_color(ind)

    def set_cursor_to_end(self):
        textCursor = self.ui.plainTextEditStastusCalib.textCursor()
        textCursor.movePosition(textCursor.End)
        self.ui.plainTextEditStastusCalib.setTextCursor(textCursor)