这是使用PyQT5作为UI的首次提交,将后端和UI合并到1个工程中，统一使用了Python，没有使用JS和HTML

mhi_plc.py

@@ -0,0 +1,322 @@
+import array
+import asyncio
+import sys
+import time
+import datetime
+
+import keyboard
+import serial
+import requests
+import threading
+import argparse
+import re
+import utils
+
+class MitsubishiPLC:
+    def __init__(self, plc_port, comm_config):
+        global print_condition, api_condition, all_plc_recv
+        port_opened = False
+        self.event_format_long = False
+        self.plc_port = plc_port
+        self.baudrate = comm_config['baudrate']
+        try:
+            self.ser = serial.Serial(port=plc_port,
+                                     baudrate=comm_config['baudrate'],
+                                     parity=comm_config['parity'],
+                                     bytesize=comm_config['bytesize'],
+                                     stopbits=comm_config['stopbits'],
+                                     timeout=1, rtscts=False)
+            # condition_print(f"DevicePLC open {plc_port} success")  # cyx
+            print(f"Mhi DevicePLC open {plc_port} {comm_config['baudrate']} {comm_config['bytesize']} success")  # cyx
+        except Exception as error:
+            print("Mhi DevicePLC open error", error)  # cyx
+            utils.condition_print("\033[91m {}\033[00m".format("plc  " + plc_port + " open error"))
+            try:
+                self.ser = serial.Serial(port="com3",
+                                         baudrate=comm_config['baudrate'],
+                                         parity=comm_config['parity'],
+                                         bytesize=comm_config['bytesize'],
+                                         stopbits=comm_config['stopbits'],
+                                         timeout=1, rtscts=False)
+            except:
+                print("\033[91m {}\033[00m".format("plc " + "com1" + " open error"))
+            else:
+                port_opened = True
+                print("\033[92m {}\033[00m".format("plc " + "com1" + " open success"))
+        else:
+            port_opened = True
+            print("\033[92m {}\033[00m".format("plc " + plc_port + " open success--1"))
+        finally:
+            pass
+
+        if port_opened:
+            self.link_down_timer = threading.Timer(5.0, self.link_down_timeout)  # 创建一个5秒后执行的定时器
+            # self.link_down_timer.start()
+            self.link_down_event = threading.Event()
+
+        if self.port_is_opened():
+            # self.recv_thread.start();
+            pass
+
+    def conv_to_plc_address(self, type_char, value):
+        # 检查type_char是否为'M'或'D'，value是否为整数
+        if type_char not in ('M', 'D') or not isinstance(value, int):
+            raise ValueError("Type must be 'M' or 'D' and value must be an integer.")
+
+        # 使用str.format()方法格式化字符串
+        formatted = "{type}{value:04d}".format(type=type_char, value=value)
+        return formatted
+
+    def readWithTimeOut(self, size, timeout):
+        origin_timeout = self.ser.timeout
+        self.ser.timeout = timeout
+        data = None
+        success = True
+        message = ""
+        start_time = time.time()
+        try:
+            data = self.ser.read(size)
+        except serial.serialTimeoutException:
+            success = False
+            message = "serial read timeout"
+        finally:
+            self.ser.timeout = origin_timeout
+        real_time = time.time() - start_time
+        # print(f"time={real_time}", message,f"timeout setting={timeout}") #cyx
+        return success, data, message
+
+    def port_is_opened(self):
+        if hasattr(self, 'ser') and self.ser:
+            return True
+        else:
+            return False
+        return False
+
+    def get_connect_state(self, connect_state_address, slave):
+        if self.port_is_opened() is False:
+            return 0
+        try:
+            # current_time = datetime.datetime.now().strftime('%H:%M:%S.%f')[:-3]
+            # print(f'{current_time} :mhi plc {self.plc_port} get_connect_state begin') #cyx
+            result = self.plc_read('WR', slave, 0xff, connect_state_address, 0x01)
+            if result and 'status' in result and result['status'] == 'success' and 'data' in result:
+                frame = result['data']
+                data_len = (len(frame) - 8) / 4
+                if data_len == 0x1:
+                    return 1
+            else:
+                return 0
+        except Exception as e:
+            return 0
+        finally:
+            # current_time = datetime.datetime.now().strftime('%H:%M:%S.%f')[:-3]
+            # print(f'{current_time} :mhi plc {self.plc_port} get_connect_state end') # cyx
+            pass
+
+    def link_down_timeout(self):
+        self.link_down_event.set()
+        self.link_down_timer = threading.Timer(5.0, self.link_down_timeout)  # 创建一个5秒后执行的定时器
+        self.link_down_timer.start()
+
+    def reset_link_down_timer(self):
+        if self.link_down_timer:
+            self.link_down_timer.cancel()
+        self.link_down_timer = threading.Timer(5.0, self.link_down_timeout)  # 创建一个5秒后执行的定时器
+        self.link_down_timer.start()
+
+    def send_heartBit(self):
+        if self.port_is_opened():
+            pass
+
+    def crc16_ccitt(self, data: bytearray):  # CRC16/CCITT
+        crcval = 0x0000
+        for c in data:
+            q = (crcval ^ c) & 0x0F
+            crcval = (crcval >> 4) ^ (q * 0o10201)
+            q = (crcval ^ (c >> 4)) & 0x0F
+            crcval = (crcval >> 4) ^ (q * 0o10201)
+        return crcval
+
+    def plc_read_frame(self, command, plc_no=0x01, pc_no=0xFF, address='D0160', lenght=0x02):
+        frame = []
+        frame.append(0x05)
+        frame.extend(utils.hex_to_ascii_arr(plc_no, 2))
+        frame.extend(utils.hex_to_ascii_arr(pc_no, 2))
+        frame.extend([ord(c) for c in command])  # word read command
+        frame.append(0x30)
+        frame.extend(utils.str_to_ascii_array(address, 5))  # eg.D0160
+        frame.extend(utils.int_to_ascii_array(lenght, 2))
+        checksum = sum(frame[1:]) & 0xFF
+        frame.extend(utils.hex_to_ascii_arr(checksum, 2))
+        # condition_print("frame=",frame,type(frame))
+        utils.condition_print('frame intent to read plc: [', ', '.join('{:02x}'.format(x) for x in frame), ']')
+        return frame
+
+    def plc_write_frame(self, command, plc_no=0x01, pc_no=0xFF, address='D0160', date_len=0x02, data=[]):
+        frame = [0x05]
+        frame.extend(utils.hex_to_ascii_arr(plc_no, 2))
+        frame.extend(utils.hex_to_ascii_arr(pc_no, 2))
+        frame.extend([ord(c) for c in command])  # word write command bit write command
+        frame.append(0x30)
+        frame.extend(utils.str_to_ascii_array(address, 5))  # eg.D0160
+        frame.extend(utils.int_to_ascii_array(date_len, 2))
+        data_index = 0
+        while date_len > 0:
+            value = data[data_index]
+            data_index = data_index + 1
+            if command == 'BW':
+                arr_len = 1
+            else:
+                arr_len = 4
+            word_ascii_array = utils.hex_to_ascii_arr(value, arr_len)
+            # condition_print(word,'[', ', '.join('{:02x}'.format(x) for x in word_ascii_array), ']')
+            frame.extend(word_ascii_array)
+            date_len = date_len - 1
+        checksum = sum(frame[1:]) & 0xFF
+        frame.extend(utils.hex_to_ascii_arr(checksum, 2))
+        utils.condition_print('frame intend to write plc: [', ', '.join('{:02x}'.format(x) for x in frame), ']')
+        return frame
+
+    # Symbol name   Description	               Code (hexadecimal)
+    # STX	         Start of Text	               02H
+    # ETX	         End of Text	               03H
+    # EOT	        End of Transmission      04H
+    # ENQ	        Enquiry	              05H
+    # ACK	        Acknowledge	             06H
+    # LF	        Line Feed	             0AH
+    # CL	        Clear	0CH
+    # CR	       Carriage Return	             ODH
+    # NAK	       Negative Acknowledge	15H
+    def frame_is_valid(self, frame):
+        # 检查第一个字节是否为控制代码
+        if len(frame) < 5:
+            return False
+        try:
+            control_code = frame[0]
+            ETX: bool = False
+            if control_code not in [0x05, 0x02, 0x03, 0x4, 0x06, 0x15]:
+                utils.condition_print("First byte is not a control code.")
+                return False
+            # 检查最后两个字节是否为校验和代码
+            hex_string = ''.join(chr(char) for char in frame[-2:])
+            # 然后，你可以使用int函数将这个16进制的字符串转换为一个整数，指定基数为16
+            checksum = int(hex_string, 16)
+            if frame[0] == 0x02 and frame[-3] != 0x03:
+                return False
+            if frame[0] == 0x06 or frame[0] == 0x15:
+                return True
+            if frame[-3] == 0x03:
+                ETX = True
+            end_position = -3 if ETX else -2
+            checksum_calc = sum(frame[1:-2]) & 0xFF  # 计算校验和并取第八位
+            if checksum != checksum_calc:
+                utils.condition_print("Last two bytes do not match checksum.", checksum, checksum_calc, frame[-2] - 0x30,
+                                frame[-1] - 0x30)
+                return False
+            # 创建一个包含所有有效字符的字符串
+            valid_chars = '0123456789abcdefABCDEF'
+
+            # 定义一个函数来检查bytes对象中的每个元素
+            for i in frame[1:end_position]:
+                char = chr(i)
+                if char not in valid_chars:
+                    utils.condition_print(f'{char} ({i}) is not a valid character.')
+                    # 在此处添加其他的逻辑处理代码，例如抛出异常
+                    # raise ValueError(f'{char} ({i}) is not a valid character.')
+                    return False
+                else:
+                    # condition_print(f'{char} ({i}) is a valid character.')
+                    pass
+        except Exception as e:
+            return False
+        return True
+
+    def plc_write(self, command='WW', plc_no=0x01, pc_no=0xFF, address='D0160', lenght=0x02, data=0x00000000):
+        frame = self.plc_write_frame(command, plc_no, pc_no, address, lenght, data)
+        timeout = 0.05  # 设置超时时间为50毫秒
+        try:
+            if hasattr(self, 'ser') and self.ser:
+                self.ser.write(frame)
+                read_bytes = 20
+                timeout = read_bytes * 11 * 1 / self.baudrate * 3
+                success, recv_frame, message = self.readWithTimeOut(read_bytes, timeout)
+                if not success:
+                    print("mhi plc write error", plc_no, address, lenght, data)  # cyx
+                    return {"status": "error", "msg": message}
+                else:
+                    if not self.frame_is_valid(recv_frame):
+                        utils.condition_print(f"recv frame is not valid recv frame len={len(recv_frame)}")
+                        return {"status": "error", "msg": "recv frame is not valid"}
+                    else:
+                        utils.condition_print('--read data return: [', ', '.join('{:02x}'.format(x) for x in recv_frame), ']')
+                        return {"status": "success"}  # {"status": "success", "data": recv_frame}
+            else:
+                return {"status": "error", "msg": "serial com is not open!"}
+            # 记录开始时间
+        except  Exception as e:
+            return {"status": "error", "msg": "serial com is not open!"}
+
+    def plc_read(self, command, plc_no=0x01, pc_no=0xFF, address='D0160', length=0x01):
+        frame = self.plc_read_frame(command, plc_no, pc_no, address, length)
+        try:
+            if hasattr(self, 'ser') and self.ser:
+                self.ser.write(frame)
+                read_bytes = 20
+                timeout = read_bytes * 11 * 1 / self.baudrate * 3
+                success, recv_frame, message = self.readWithTimeOut(read_bytes, timeout)
+                if not success:
+                    return {"status": "error", "msg": message}
+                else:
+                    if not self.frame_is_valid(recv_frame):
+                        utils.condition_print(f"recv frame is not valid frame len={len(recv_frame)}")
+                        return {"status": "error", "msg": "recv frame is not valid"}
+                    else:
+                        utils.condition_print('--read data return: [', ', '.join('{:02x}'.format(x) for x in recv_frame), ']')
+                        return {"status": "success", "data": recv_frame}
+            else:
+                return {"status": "error", "msg": "serial com is not open!"}
+        except  Exception as e:
+            return {"status": "error", "msg": "serial com is not open!"}
+
+    def plc_read_words(self, plc_no=0x01, pc_no=0xFF, address='D0160', length=0x01):
+        utils.condition_print(f'mhi read words {address} {length} ')
+        resp = {'status': 'error'}
+        result = self.plc_read('WR', plc_no, pc_no, address, length)
+        if result and 'status' in result and result['status'] == 'success' and 'data' in result:
+            frame = result['data']
+            data_len = (len(frame) - 8) / 4
+            if data_len == length:
+                value = [utils.ascii_array_to_word(frame[5 + 4 * i:5 + 4 * (i + 1)]) for i in range(length)]
+                resp = {'status': 'success', 'length': data_len, 'data': value}
+        return resp
+
+    def plc_read_bits(self, plc_no=0x01, pc_no=0xFF, address='D0160', length=0x01):
+        utils.condition_print(f'mhi read bits {address} {length} ')
+        resp = {'status': 'error'}
+        result = self.plc_read('BR', plc_no, pc_no, address, length)
+        utils.condition_print('plc_read_bits result', result)  # cyx
+        if 'status' in result and result['status'] == 'success' and 'data' in result:
+            frame = result['data']
+            data_len = (len(frame) - 8) / 1
+            if data_len == length:
+                value = [(frame[5 + 1 * i] - 0x30) for i in range(length)]
+                resp = {'status': 'success', 'length': data_len, 'data': value}
+        return resp
+
+    def plc_write_word(self, plc_no=0x01, pc_no=0xFF, address='D0160', data=0x0000):
+        utils.condition_print(f'mhi write word {address} {data} ')
+        return self.plc_write('WW', plc_no, pc_no, address, 0x01, list([data]))
+
+    def plc_write_dword(self, plc_no=0x01, pc_no=0xFF, address='D0160', data=0x00000000):
+        utils.condition_print(f'mhi write dword {address} {data} ')
+        return self.plc_write('WW', plc_no, pc_no, address, 0x02, list([data >> 16, data & 0xFFFF]))
+
+    def plc_write_words(self, plc_no=0x01, pc_no=0xFF, start_addr='D0160', length=0x04, data=[]):
+        utils.condition_print(f'mhi write words {start_addr} {length} ', data)
+        return self.plc_write('WW', plc_no, pc_no, start_addr, length, data)
+
+    def plc_write_bit(self, plc_no=0x01, pc_no=0xFF, address='D0160', data=0x00):
+        utils.condition_print(f'mhi write bit {address} {data} ')
+        return self.plc_write('BW', plc_no, pc_no, address, 0x1, list([data]))
+
+