Project/入口LED屏/led_control_XmSgj.py

import sys

from async_communication import TimeStatu

sys.path.append("..")
# 导入pymysql
import socket
import threading
import time
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor

import google.protobuf.text_format as tf

import message_pb2 as message
from led_protocol import LedProtocol
from led_protocol import font_library
import mytool.db_helper.db_operation as spmng


class led_control(threading.Thread):
    def __init__(self, unit, channel, communication, area_list, db_config):
        threading.Thread.__init__(self)
        self._dispatch_statu = {}
        self.led_communication = communication
        self.area_list = area_list
        self.area_1_show_str_timer = TimeStatu('', 0.1)
        self.area_2_show_str_timer = TimeStatu('', 0.1)
        self.db = spmng.DBOperation(db_config['db_ip'], db_config['db_port'], db_config['db_name'],
                                    db_config['db_user'], db_config['db_password'])
        self.unit = unit
        # if self.unit == 31 or self.unit == 32:
        #     self.en_font = font_library["EN32"]
        #     self.cn_font = font_library["CN32"]
        # else:
        #     self.en_font = font_library["EN16"]
        #     self.cn_font = font_library["CN16"]
        self.en_font = font_library["EN16"]
        self.cn_font = font_library["CN16"]
        self.channel = channel
        self.led_protocol = LedProtocol(area_list)
        self._pool = ThreadPoolExecutor()
        self.led_control_init()

    def led_control_init(self):
        font = font_library["EN32"]
        unit_char = ""
        if self.unit == 1:
            unit_char = 'A'
        elif self.unit == 2:
            unit_char = 'B'
        elif self.unit == 11:
            unit_char = 'A1'
        elif self.unit == 12:
            unit_char = 'A2'
        elif self.unit == 13:
            unit_char = 'A3'
        elif self.unit == 14:
            unit_char = 'A4'
        elif self.unit == 21:
            unit_char = 'B1'
        elif self.unit == 22:
            unit_char = 'B2'
        elif self.unit == 23:
            unit_char = 'B3'
        elif self.unit == 24:
            unit_char = 'B4'
        elif self.unit == 25:
            unit_char = 'B5'
        elif self.unit == 31:
            unit_char = 'C1'
            # font = font_library["EN48"]
        elif self.unit == 32:
            unit_char = 'C2'
            # font = font_library["EN48"]

        self.send_led_msg(0, unit_char, en_font=font, cn_font=self.cn_font)

    def receive_dispatch_statu(self, statu, ex, key):
        self._dispatch_statu[key] = statu

    def get_parking_string(self, small_space_count, big_space_count):

        led_show_string = "剩大车位:%d\\n剩小车位:%d" % (big_space_count, small_space_count)
        return led_show_string

    def get_entrance_statu_led_string(self, small_space_count, big_space_count, pick_command_count):
        key = "dispatch_%d_statu_port" % self.unit

        led_show_string = ""
        if (key in self._dispatch_statu) is False or self._dispatch_statu[key].timeout():
            print('ERROR --- 调度节点未连接----key:%s---------time:%s' % (key, datetime.now()))
            led_show_string = "故障！！"
        else:
            dispatch_node_statu = message.dispatch_node_statu()
            try:
                tf.Parse(self._dispatch_statu[key].statu, dispatch_node_statu)
                if dispatch_node_statu.plc_carrier_status == 0:
                    led_show_string = "故 障！"
                elif dispatch_node_statu.plc_carrier_status == 4:
                    led_show_string = "维护中！"
                else:
                    if big_space_count <= 0 and small_space_count <= 0:
                        led_show_string = "无位 不可进\\n取车候车:%d" % pick_command_count
                    elif big_space_count > 0 and (dispatch_node_statu.dispatch_plc_passway_status_vector[
                                                      self.channel - 1].plc_passway_enable == 0 or
                                                  dispatch_node_statu.dispatch_plc_passway_status_vector[
                                                      self.channel - 1].plc_passway_enable == 2):
                        led_show_string = "占用 不可进\\n取车候车:%d" % pick_command_count
                    elif big_space_count > 0 and (dispatch_node_statu.dispatch_plc_passway_status_vector[
                                                      self.channel - 1].plc_passway_enable == 1 or
                                                  dispatch_node_statu.dispatch_plc_passway_status_vector[
                                                      self.channel - 1].plc_passway_enable == 3):
                        led_show_string = "空闲 可进入\\n取车候车:%d" % pick_command_count
                    elif small_space_count > 0 and (dispatch_node_statu.dispatch_plc_passway_status_vector[
                                                        self.channel - 1].plc_passway_enable == 0 or
                                                    dispatch_node_statu.dispatch_plc_passway_status_vector[
                                                        self.channel - 1].plc_passway_enable == 2):
                        led_show_string = "占用 不可进\\n取车候车:%d" % pick_command_count
                    elif small_space_count > 0 and (dispatch_node_statu.dispatch_plc_passway_status_vector[
                                                        self.channel - 1].plc_passway_enable == 1 or
                                                    dispatch_node_statu.dispatch_plc_passway_status_vector[
                                                        self.channel - 1].plc_passway_enable == 3):
                        led_show_string = "空闲 可进入\\n取车候车:%d" % pick_command_count

            except Exception:
                print('ERROR --- 调度状态消息解析错误  key:%s   time:%s    message:\n%s' % (
                    key, datetime.now(), self._dispatch_statu[key].statu))
                led_show_string = "故 障！"

        return led_show_string

    def send_led_msg(self, area_index, input, en_font=font_library["EN32"], cn_font=font_library["CN32"]):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(1)
        try:
            sock.connect((self.led_communication['ip'], self.led_communication['port']))
            time.sleep(0.1)
            sock.send(self.led_protocol.string2bytes(area_index, input, en_font=en_font, cn_font=cn_font))
            print("connected led \033[0:32mSUCCEND\033[m ip:%s   port:%d\nsend msg area_index:%d   msg:%s" % (
                self.led_communication['ip'], self.led_communication['port'], area_index, input))
        except Exception as e:
            print("connected led \033[0:31mERROR\033[m: %s    time:%s   msg=%s" % (str(e.args), datetime.now(), input))
        time.sleep(0.1)
        sock.close()

    def run(self):
        i = 0

        while True:
            time.sleep(1)
            if i == 20:
                self.led_control_init()
                time.sleep(0.1)
                i=0
            i = i + 1
            small_space_res = self.db.query_space_in_height_unit_and_empty_2(1.48, 1.5, self.unit)
            big_space_res = self.db.query_space_in_height_unit_and_empty(2, self.unit)

            command_small_res = self.db.query_command_in_height_minmax_unit_and_statu(1.48, 1.5, self.unit)
            command_big_res = self.db.query_command_in_height_unit_and_statu(2, self.unit)

            pick_command_res = self.db.query_sort_pick_command(self.unit)

            if small_space_res is None or big_space_res is None or command_small_res is None or command_big_res is None or pick_command_res is None:
                continue

            small_space_count = len(small_space_res) - len(command_small_res)
            big_space_count = len(big_space_res) - len(command_big_res)
            pick_commandcount = len(pick_command_res)

            statu_string = self.get_entrance_statu_led_string(small_space_count, big_space_count, pick_commandcount)
            parking_string = self.get_parking_string(small_space_count, big_space_count)

            if self.area_1_show_str_timer.statu != statu_string or self.area_1_show_str_timer.timeout():
                self.area_1_show_str_timer = TimeStatu(statu_string, 20)
                self._pool.submit(self.send_led_msg, 1, statu_string,self.en_font,self.cn_font)

            # test
            # self._pool.submit(self.send_led_msg, 2, str(i),self.en_font,self.cn_font)
            # i = i + 1

            if self.area_2_show_str_timer.statu != parking_string or self.area_2_show_str_timer.timeout():
                self.area_2_show_str_timer = TimeStatu(parking_string,20)
                self._pool.submit(self.send_led_msg,2, parking_string,self.en_font,self.cn_font)