Project/入口引导节点/window_screen_pyqt.py

import datetime
import threading

from PyQt5.QtWidgets import QWidget, QApplication, QMainWindow, QLabel
from PyQt5.QtGui import QPixmap, QPainter, QResizeEvent, QCloseEvent, QPaintEvent, QFont, QColor
from PyQt5.QtCore import QSize, QTimer, QRect, Qt
import async_communication as CM
import message_pb2 as message
import google.protobuf.text_format as tf
import logging

logging.basicConfig(filename='log.txt',
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s-%(funcName)s',
                    level=logging.ERROR)
MeasureStatu = {"ok": 0, "无数据": 1, "噪声": 2, "超界": 3, "故障": 4, "失去连接": 5}
ArrowType = {message.Range_status.Value('Range_correct'): "正确图片",
             message.Range_status.Value('Range_front'): "向后调整",
             message.Range_status.Value('Range_back'): "向前调整",
             message.Range_status.Value('Range_left'): "向右调整",
             message.Range_status.Value('Range_right'): "向左调整",
             message.Range_status.Value('Range_car_width'): "超宽车辆",
             message.Range_status.Value('Range_car_wheelbase'): "轴距超差",
             message.Range_status.Value('Range_angle_anti_clock'): "向右旋转",
             message.Range_status.Value('Range_angle_clock'): "向左旋转",
             message.Range_status.Value('Range_steering_wheel_nozero'): "回正方向盘",
}


class PicLabel(QLabel):
    def __init__(self, parent):
        super(PicLabel, self).__init__(parent=parent)
        self.image = None
        self.Txt = None
        self.show = False
        self.setStyleSheet("border-width:1px;border-style:solid;border-color:rgb(150,150,150)")
        self.timer = QTimer()
        self.timer.timeout.connect(self.OnTimer)

    def FlashImg(self, pixmap):
        self.image = pixmap
        if self.timer.isActive() == False:
            self.timer.start(500)

    def ShowImg(self, pixmap):
        self.timer.stop()
        self.show = True
        self.Txt = None
        self.image = pixmap
        self.repaint()


    def ShowTxt(self, txt):
        self.timer.stop()
        self.show = True
        self.image = None
        self.Txt = txt
        self.repaint()


    def OnTimer(self):
        self.show = not self.show
        self.repaint()

    def paintEvent(self, a0: QPaintEvent) -> None:
        if self.show is True:
            if self.image is not None:
                w, h = self.size().width(), self.size().height()
                iw, ih = self.image.width(), self.image.height()
                painter = QPainter(self)
                painter.drawPixmap(QRect(0, 0, w, h), self.image, QRect(0, 0, iw, ih))
            elif self.Txt is not None:
                w, h = self.size().width() / len(self.Txt), self.size().height() / 1.5
                painter = QPainter(self)
                painter.setPen(QColor(255, 0, 0))
                painter.setFont(QFont('SimSun', int(min(w, h)),QFont.Bold))
                # print(self.Txt)
                painter.drawText(a0.rect(), Qt.AlignCenter, self.Txt)


class Frame(QMainWindow):
    def __init__(self, import_id, images, prj_parameter, second_screen=False):
        super(Frame, self).__init__()
        self.flag = None
        self.last_back_io = None
        self.images = images
        self.images = {}
        self.import_id = import_id
        self.dispatch_statu = CM.TimeStatu(timeout=0.1)
        self.measure_statu = CM.TimeStatu(timeout=0.1)
        self.terminal_statu = CM.TimeStatu(timeout=0.1)

        for key, file in images.items():
            self.images[key] = pix = QPixmap(file)

        self.InitUI()

        # self.last_door_statu=None
        self.last_moving_statu = None
        # self.last_show = None
        # self.last_back_io = "正常"
        # self.flag = None
        if second_screen == True:
            desktop = QApplication.desktop()
            screen_count = desktop.screenCount()
            displays = [desktop.availableGeometry(i) for i in range(desktop.screenCount())]
            second_offx = 0
            second_offy = 0

            if len(displays) == 2:
                second_offx = displays[1].width()
                second_offy = displays[1].height()
            self.setGeometry(second_offx, second_offy, 200, 500)


        self.timer = QTimer()
        self.timer.timeout.connect(self.Switch)
        self.timer.start(200)
        self.lock = threading.Lock()

    def InitUI(self):
        self.measure_info_txt = QLabel(self)
        self.car_number_txt = QLabel(self)
        self.car_number_txt.setStyleSheet("border-width:1px;border-style:solid;border-color:rgb(150,150,150)")
        self.car_number_txt.setAlignment(Qt.AlignCenter)
        self.measure_info_txt.setStyleSheet("color:blue")
        self.panel_txt = PicLabel(self)
        self.panel_arrow = PicLabel(self)
    def receive_dispatch(self, statu):
        self.dispatch_statu = statu

    def receive_measureInfo(self, statu):
        self.measure_statu = statu

    def receive_terminal(self, statu):
        self.terminal_statu = statu

    def receive_entrance_update(self,statu):
        with self.lock:
            print("---------------刷新投影-----------------")
            self.last_moving_statu = None

    def resizeEvent(self, a0: QResizeEvent) -> None:
        w, h = a0.size().width(), a0.size().height()
        w, h = self.size().width(), self.size().height()
        top_w, top_h = w, int(h * 0.15)
        car_number_w, car_number_h = w, int(h * 0.10)
        txt_w, txt_h = w, int(h * 0.20)
        arrow_w, arrow_h = w, int(h * 0.55)
        font = QFont()
        font.setBold(True)
        font.setPixelSize(int(w / 40))
        self.measure_info_txt.setFont(font)
        self.measure_info_txt.setGeometry(0, 0, top_w, top_h)
        font.setPixelSize(int(w / 20))
        self.car_number_txt.setFont(font)
        self.car_number_txt.setGeometry(0, top_h, car_number_w, car_number_h)

        self.panel_txt.setGeometry(0, top_h + car_number_h, txt_w, txt_h)
        self.panel_arrow.setGeometry(0, top_h + car_number_h + txt_h, arrow_w, arrow_h)
    def Switch(self):
        try:
            if self.terminal_statu.timeout() is False:
                terminal_statu = message.terminal_node_statu()
                tf.Parse(self.terminal_statu.statu, terminal_statu)
                self.car_number_txt.setText(terminal_statu.car_number)
                self.car_number_txt.setStyleSheet(
                    "border-width:1px;border-style:solid;border-color:rgb(150,150,150);color:green")
            else:
                self.car_number_txt.setText("车牌号为空！")
                self.car_number_txt.setStyleSheet(
                    "border-width:1px;border-style:solid;border-color:rgb(150,150,150);color:red")

            # 先判断连接状态
            if self.dispatch_statu.timeout():
                self.panel_txt.ShowImg(self.images['传感器超时'])
                self.panel_arrow.FlashImg(self.images['传感器超时'])
                return

            if self.measure_statu.timeout():
                self.panel_txt.ShowImg(self.images['测绘超时'])
                self.panel_arrow.FlashImg(self.images['测绘超时'])
                return

            # 有车才显示
            dispatch_statu = message.dispatch_node_statu()
            tf.Parse(self.dispatch_statu.statu, dispatch_statu)

            measure_buffer = message.measure_buffer()
            measure_info = message.measure_info()
            tf.Parse(self.measure_statu.statu, measure_buffer)
            measure_info.CopyFrom(measure_buffer.measure_info_to_terminal)
            # 测量状态
            lidar_statu = measure_info.ground_status  # 测量状态（正常,无数据、噪声、超界）
            # 传感器状态
            dispatch_plc_passway_status = dispatch_statu.dispatch_plc_passway_status_vector[self.import_id - 1]
            sensor_1 = dispatch_plc_passway_status.sensor_1



            if lidar_statu == MeasureStatu['故障']:
                self.panel_txt.ShowImg(self.images['测绘超时'])
                self.panel_arrow.FlashImg(self.images['测绘超时'])
                return
            # 获取传感器数据
            # print('---------------------------------')
            # print('PLC数据:')
            # print(dispatch_plc_passway_status)
            # print('----------------------------------')
            # print('====================================')
            # print('雷达数据:')
            # print(measure_info)
            # print('====================================')

            # 判断超界信息
            if (sensor_1 >> 3) & 0x01 == 0:
                if not self.last_back_io == "超界":
                    self.last_back_io = "超界"
                    self.flag = False
            else:
                if not self.last_back_io == "正常":
                    self.last_back_io = "正常"
                    self.flag = True

            # if dispatch_plc_passway_status.car_height == 0:
            #     self.panel_txt.ShowImg(self.images["空闲"])
            #     self.panel_arrow.ShowImg(self.images["空闲"])
            #     return

            if dispatch_plc_passway_status.car_height == 4:
                self.panel_txt.ShowImg(self.images["超高车辆"])
                self.panel_arrow.ShowImg(self.images["超高车辆"])
                return


            # 以下是有车的处理
            # 1,显示测量数据（实时和静态）

            static_info_txt = ''
            realtime_info_txt = "(实时) 中心(%.3f,%.3f) 角度:%.2f° 宽:%.3f 轴距:%.3f 前轮角:%.2f°" % \
                                (measure_info.cx, measure_info.cy, measure_info.theta, measure_info.width,
                                 measure_info.wheelbase, measure_info.front_theta)
            if not self.last_moving_statu == None:
                static_info_txt = "(静止) 中心(%.3f,%.3f) 角度:%.2f° 宽:%.3f 轴距:%.3f 前轮角:%.2f°" % \
                                  (self.last_moving_statu.cx, self.last_moving_statu.cy,
                                   self.last_moving_statu.theta, self.last_moving_statu.width,
                                   self.last_moving_statu.wheelbase, self.last_moving_statu.front_theta)
            self.measure_info_txt.setText(realtime_info_txt + "\n" + static_info_txt)

            # 2,获取车辆运动状态（静止or动态）
            border_statu = measure_info.border_statu
            is_moving = measure_info.motion_statu == 0
            print(is_moving)
            # 当前运动
            if is_moving:
                with self.lock:
                    self.last_moving_statu = None
            else:
                # 当前静止
                if lidar_statu == MeasureStatu["ok"]:
                    with self.lock:
                        self.last_moving_statu = measure_info
                elif lidar_statu == MeasureStatu["超界"]:
                    if not self.last_moving_statu == None:
                        # 上一刻静止且数据正确，当前静止，不可能出现超界，将超界清除
                        new_border = border_statu
                        new_border = (new_border & (~(0x01 << 0)))
                        new_border = (new_border & (~(0x01 << 1)))
                        new_border = (new_border & (~(0x01 << 2)))
                        new_border = (new_border & (~(0x01 << 3)))
                        new_border = (new_border & (~(0x01 << 6)))
                        new_border = (new_border & (~(0x01 << 7)))
                        new_border = (new_border & (~(0x01 << 8)))
                        new_border = (new_border & (~(0x01 << 9)))
                        border_statu = new_border
                elif lidar_statu == MeasureStatu["无数据"]:
                    # 当前静止无车，清除上一时刻数据
                    with self.lock:
                        self.last_moving_statu = None
                # elif lidar_statu == MeasureStatu["噪声"]:
                    # if not self.last_moving_statu == None:
                        # 上一时刻静止且正确，当前噪声，不显示当前数据，显示上一正确数据
                        # measure_info.CopyFrom(self.last_moving_statu)

            # print((sensor_1 >> 4) & 0x01)
            # 先判断光电
            if (sensor_1 >> 3) & 0x01 == 0:
                self.panel_txt.ShowImg(self.images["请调整"])
                self.panel_arrow.FlashImg(self.images["向前调整"])
                self.last_show = "调整"
                return
            elif (sensor_1 >> 4) & 0x01 == 0:
                self.panel_txt.ShowImg(self.images["请调整"])
                self.panel_arrow.FlashImg(self.images["向后调整"])
                self.last_show = "调整"
                # print(111)
                return
            # elif (sensor_1 >> 5) & 0x01 == 0:
            #     self.panel_txt.ShowImg(self.images["请调整"])
            #     self.panel_arrow.FlashImg(self.images["向右调整"])
            #     self.last_show = "调整"
            #     return
            # elif (sensor_1 >> 6) & 0x01 == 0:
            #     self.panel_txt.ShowImg(self.images["请调整"])
            #     self.panel_arrow.FlashImg(self.images["向左调整"])
            #     self.last_show = "调整"
            #     return
            lidar_statu = measure_info.ground_status  # 测量状态（正常,无数据、噪声、超界）

            # 光电正常
            if lidar_statu == MeasureStatu["无数据"]:
                self.panel_txt.ShowImg(self.images["空闲"])
                self.panel_arrow.ShowImg(self.images["空闲"])
                return
            elif lidar_statu == MeasureStatu["失去连接"]:
                self.panel_txt.ShowImg(self.images['测绘超时'])
                self.panel_arrow.FlashImg(self.images['测绘超时'])
                return
            elif lidar_statu == MeasureStatu["噪声"]:
                    self.panel_txt.ShowImg(self.images['请调整'])
                    self.panel_arrow.ShowImg(self.images["请调整"])
                    return
            elif lidar_statu == MeasureStatu["ok"]:
                self.panel_txt.ShowImg(self.images['正确文字'])
                self.panel_arrow.ShowImg(self.images['正确图片'])
                return
            # 查看是否超界
            elif lidar_statu == MeasureStatu["超界"]:
                border = border_statu & 0xf7ff
                if (border in ArrowType.keys()) is False:
                    self.panel_txt.ShowImg(self.images['检查杂物'])
                    self.panel_arrow.FlashImg(self.images['检查杂物'])
                else:
                    if ArrowType[border] == "正确图片":
                        self.panel_arrow.ShowImg(self.images[ArrowType[border]])
                        self.panel_txt.ShowImg(self.images['正确文字'])
                    elif ArrowType[border] == "向前调整":
                        if measure_info.move_distance > 0.01:
                            self.panel_txt.ShowTxt("请向前" + '%.2f'%(measure_info.move_distance * 100) + "厘米")
                        else:
                            self.panel_txt.ShowImg(self.images['请调整'])
                        self.panel_arrow.FlashImg(self.images["向前调整"])
                    else:
                        self.panel_arrow.FlashImg(self.images[ArrowType[border]])
                        self.panel_txt.ShowImg(self.images['请调整'])


        except Exception as e:
            print(str(e.args))
            logging.error(str(e.args))

    def Switch1(self):
        try:
            if self.terminal_statu.timeout() is False:
                terminal_statu = message.terminal_node_statu()
                tf.Parse(self.terminal_statu.statu, terminal_statu)
                # print(terminal_statu.car_number)
                self.car_number_txt.setText(terminal_statu.car_number)
                self.car_number_txt.setStyleSheet(
                    "border-width:1px;border-style:solid;border-color:rgb(150,150,150);color:green")
            else:
                self.car_number_txt.setText("车牌号为空！")
                self.car_number_txt.setStyleSheet(
                    "border-width:1px;border-style:solid;border-color:rgb(150,150,150);color:red")

            # 先判断连接状态
            if self.dispatch_statu.timeout():
                self.panel_txt.ShowImg(self.images['传感器超时'])
                self.panel_arrow.FlashImg(self.images['传感器超时'])
                return

            if self.measure_statu.timeout():
                self.panel_txt.ShowImg(self.images['测绘超时'])
                self.panel_arrow.FlashImg(self.images['测绘超时'])
                return

            # 有车才显示
            dispatch_statu = message.dispatch_node_statu()
            tf.Parse(self.dispatch_statu.statu, dispatch_statu)

            measure_buffer = message.measure_buffer()
            measure_info = message.measure_info()
            tf.Parse(self.measure_statu.statu, measure_buffer)
            measure_info.CopyFrom(measure_buffer.measure_info_to_terminal)
            # 测量状态
            lidar_statu = measure_info.ground_status  # 测量状态（正常,无数据、噪声、超界）
            # 传感器状态
            dispatch_plc_passway_status = dispatch_statu.dispatch_plc_passway_status_vector[self.import_id - 1]
            print(dispatch_plc_passway_status.car_height)
            sensor_1 = dispatch_plc_passway_status.sensor_1




            if lidar_statu == MeasureStatu['故障']:
                self.panel_txt.ShowImg(self.images['测绘超时'])
                self.panel_arrow.FlashImg(self.images['测绘超时'])
                return
            # 获取传感器数据
            # print('---------------------------------')
            # print('PLC数据:')
            # print(dispatch_plc_passway_status)
            # print('----------------------------------')
            # print('====================================')
            # print('雷达数据:')
            # print(measure_info)
            # print('====================================')

            # 关门清除上一帧数据
            if dispatch_plc_passway_status.outside_door_status == 2:
                with self.lock:
                    self.last_moving_statu = None
            # 判断超界信息
            if (sensor_1 >> 3) & 0x01 == 0:
                if not self.last_back_io == "超界":
                    self.last_back_io = "超界"
                    self.flag = False
            else:
                if not self.last_back_io == "正常":
                    self.last_back_io = "正常"
                    self.flag = True


            if dispatch_plc_passway_status.car_height == 0:
                self.panel_txt.ShowImg(self.images["空闲"])
                self.panel_arrow.ShowImg(self.images["空闲"])
                return

            if dispatch_plc_passway_status.car_height == 4:
                self.panel_txt.ShowImg(self.images["超高车辆"])
                self.panel_arrow.ShowImg(self.images["超高车辆"])
                return




            # 以下是有车的处理
            # 1,显示测量数据（实时和静态）

            realtime_info_txt = ''
            static_info_txt = ''
            realtime_info_txt = "(实时) 中心(%.3f,%.3f) 角度:%.2f° 宽:%.3f 轴距:%.3f 前轮角:%.2f°" % \
                                (measure_info.cx, measure_info.cy, measure_info.theta, measure_info.width,
                                 measure_info.wheelbase, measure_info.front_theta)
            if not self.last_moving_statu == None:
                static_info_txt = "(静止) 中心(%.3f,%.3f) 角度:%.2f° 宽:%.3f 轴距:%.3f 前轮角:%.2f°" % \
                                  (self.last_moving_statu.cx, self.last_moving_statu.cy,
                                   self.last_moving_statu.theta, self.last_moving_statu.width,
                                   self.last_moving_statu.wheelbase, self.last_moving_statu.front_theta)
            self.measure_info_txt.setText(realtime_info_txt + "\n" + static_info_txt)

            # 2,获取车辆运动状态（静止or动态）
            border_statu = measure_info.border_statu
            is_moving = measure_info.motion_statu == 0

            # 当前运动
            if is_moving:
                with self.lock:
                    self.last_moving_statu = None
            else:
                # 当前静止
                if lidar_statu == MeasureStatu["ok"]:
                    with self.lock:
                        self.last_moving_statu = measure_info
                elif lidar_statu == MeasureStatu["超界"]:
                    if not self.last_moving_statu == None:
                        # 上一刻静止且数据正确，当前静止，不可能出现超界，将超界清除
                        new_border = border_statu
                        new_border = (new_border & (~(0x01 << 0)))
                        new_border = (new_border & (~(0x01 << 1)))
                        new_border = (new_border & (~(0x01 << 2)))
                        new_border = (new_border & (~(0x01 << 3)))
                        new_border = (new_border & (~(0x01 << 6)))
                        new_border = (new_border & (~(0x01 << 7)))
                        new_border = (new_border & (~(0x01 << 8)))
                        new_border = (new_border & (~(0x01 << 9)))
                        border_statu = new_border
                elif lidar_statu == MeasureStatu["无数据"]:
                    # 当前静止无车，清除上一时刻数据
                    with self.lock:
                        self.last_moving_statu = None
                elif lidar_statu == MeasureStatu["噪声"]:
                    if not self.last_moving_statu == None:
                        # 上一时刻静止且正确，当前噪声，不显示当前数据，显示上一正确数据
                        measure_info.CopyFrom(self.last_moving_statu)

            # print((sensor_1 >> 4) & 0x01)
            # 先判断光电
            if (sensor_1 >> 3) & 0x01 == 0:
                self.panel_txt.ShowImg(self.images["请调整"])
                self.panel_arrow.FlashImg(self.images["向前调整"])
                self.last_show = "调整"
                return
            elif (sensor_1 >> 4) & 0x01 == 0:
                self.panel_txt.ShowImg(self.images["请调整"])
                self.panel_arrow.FlashImg(self.images["向后调整"])
                self.last_show = "调整"
                # print(111)
                return
            # elif (sensor_1 >> 5) & 0x01 == 0:
            #     self.panel_txt.ShowImg(self.images["请调整"])
            #     self.panel_arrow.FlashImg(self.images["向右调整"])
            #     self.last_show = "调整"
            #     return
            # elif (sensor_1 >> 6) & 0x01 == 0:
            #     self.panel_txt.ShowImg(self.images["请调整"])
            #     self.panel_arrow.FlashImg(self.images["向左调整"])
            #     self.last_show = "调整"
            #     return
            lidar_statu = measure_info.ground_status  # 测量状态（正常,无数据、噪声、超界）

            # 光电正常
            if lidar_statu == MeasureStatu["无数据"]:
                self.panel_txt.ShowImg(self.images["空闲"])
                self.panel_arrow.ShowImg(self.images["空闲"])
                self.last_show = "空闲"
                return
            elif lidar_statu == MeasureStatu["失去连接"]:
                self.panel_txt.ShowImg(self.images['测绘超时'])
                self.panel_arrow.FlashImg(self.images['测绘超时'])
                self.last_show = "超时"
                return
            elif lidar_statu == MeasureStatu["噪声"]:
                if self.last_moving_statu is None and self.flag is True:
                    self.panel_txt.ShowImg(self.images['请调整'])
                    self.panel_arrow.ShowImg(self.images["请调整"])
                    self.last_show = "请调整"
                    return
                if self.last_show == "超时":
                    self.panel_txt.ShowImg(self.images["空闲"])
                    self.panel_arrow.ShowImg(self.images["空闲"])
                    self.last_show = "空闲"
                    return
            elif lidar_statu == MeasureStatu["ok"]:
                self.panel_txt.ShowImg(self.images['正确文字'])
                self.panel_arrow.ShowImg(self.images['正确图片'])
                return
            # 查看是否超界
            elif lidar_statu == MeasureStatu["超界"]:
                border = border_statu & 0xf7ff
                if (border in ArrowType.keys()) is False:
                    self.panel_txt.ShowImg(self.images['检查杂物'])
                    self.panel_arrow.FlashImg(self.images['检查杂物'])
                    self.last_show = "检查杂物"
                else:
                    if ArrowType[border] == "正确图片":
                        self.panel_arrow.ShowImg(self.images[ArrowType[border]])
                        self.panel_txt.ShowImg(self.images['正确文字'])
                        self.last_show = "正确"
                    elif ArrowType[border] == "向前调整":
                        if (measure_info.move_distance is not None):
                            self.panel_txt.ShowTxt("请向前" + '%.2f'%(measure_info.move_distance * 100) + "厘米")
                        else:
                            self.panel_txt.ShowImg(self.images['请调整'])
                        self.panel_arrow.FlashImg(self.images["向前调整"])
                        self.last_show = "请调整"
                    else:
                        self.panel_arrow.FlashImg(self.images[ArrowType[border]])
                        self.panel_txt.ShowImg(self.images['请调整'])
                        self.last_show = "请调整"


        except Exception as e:
            print(str(e.args))
            logging.error(str(e.args))