Project/指令检查节点/node.py

import sys

import json
sys.path.append("..")
import datetime
import os
import threading
import time
import async_communication as CM
import message_pb2 as message
import google.protobuf.text_format as tf
import check_command_XmSgj as CHE_XmSgj
import check_command_GyBhhy as CHE_GyBhhy
import check_command_ZKXY as CHE_ZKXY

import mytool.json_helper.parse_json as parse_json
from http.server import HTTPServer, BaseHTTPRequestHandler
from mytool.log_helper.logger import HandleLog
def check_park_command(body):
    # print("接收停车指令 %s  message:%s" % (datetime.datetime.now(), body))
    log.info("接收停车指令 %s    message:%s" % (datetime.datetime.now(),body[0:260]))
    cmd = checker.entrance_isOK(body)
    table = message.park_table()
    table.CopyFrom(cmd)
    cmd.car_number_info.plate_full_image = ''
    cmd.car_number_info.plate_clip_image = ''
    response_port = "user_park_command_%d_response_port" % cmd.terminal_id
    dispatch_key = ""
    entrance_key = "entrance_update_%d_port" % cmd.terminal_id
    g_rabbitmq.publish(mq_config['mq_command_exchange_name'], response_port, tf.MessageToString(cmd, as_utf8=True))
    if cmd.statu.execute_statu == message.eNormal:
        # 指令检查正常
        dispatch_key = "dispatch_park_command_%d_request_port" % cmd.unit_id
        g_rabbitmq.publish(mq_config['mq_command_exchange_name'], dispatch_key, tf.MessageToString(table, as_utf8=True))
        log.info("停车指令 %s OK:%s    检查成功!\n" % (datetime.datetime.now(), tf.MessageToString(table, as_utf8=True)))
    else:
        print("---------------刷新投影-----------------" + entrance_key)
        g_rabbitmq.publish(mq_config['mq_statu_exchange_name'], entrance_key,
                           tf.MessageToString(cmd, as_utf8=True))
        if cmd.statu.execute_statu == message.eWarning:
            # 给终端发送指令，刷新投影
            log.warning("停车指令 %s  车牌号:%s    检查失败!\n 发送反馈    交换机:%s  发送反馈端口:%s    反馈表单：%s" % (
                    datetime.datetime.now(), cmd.car_number,
                    mq_config['mq_command_exchange_name'], response_port + "," + dispatch_key,
                    tf.MessageToString(cmd, as_utf8=True)))

        elif cmd.statu.execute_statu == message.eError:
            log.error("停车指令 %s  车牌号:%s    检查失败!\n 发送反馈    交换机:%s  发送反馈端口:%s    反馈表单：%s" % (
                datetime.datetime.now(), cmd.car_number,
                mq_config['mq_command_exchange_name'], response_port + "," + dispatch_key,
                tf.MessageToString(cmd, as_utf8=True)))
        else:
            log.critical("停车指令 %s  车牌号:%s    检查失败!\n 发送反馈    交换机:%s  发送反馈端口:%s    反馈表单：%s" % (
                datetime.datetime.now(), cmd.car_number,
                mq_config['mq_command_exchange_name'], response_port + "," + dispatch_key,
                tf.MessageToString(cmd, as_utf8=True)))


def check_pick_command(body):
    log.info("接收取车指令 %s    message:%s" % (datetime.datetime.now(), body))
    cmd = checker.exit_isOK(body)
    response_port = "user_pick_command_%d_response_port" % cmd.terminal_id
    dispatch_key = ""
    if cmd.statu.execute_statu == message.eNormal and cmd.statu.table_process_mod != message.PROCESS_ONLY_TO_PAY:
        # 指令检查正常
        log.info("取车指令 %s OK :%s    检查成功!\n" % (datetime.datetime.now(), tf.MessageToString(cmd, as_utf8=True)))
        dispatch_key = "dispatch_pick_command_%d_request_port" % cmd.unit_id
        g_rabbitmq.publish(mq_config['mq_command_exchange_name'], dispatch_key, tf.MessageToString(cmd, as_utf8=True))
    else:
        if cmd.statu.execute_statu == message.eWarning:
            log.warning("取车指令 %s  唯一码:%s    检查失败!\n 发送反馈    交换机:%s  发送反馈端口:%s    反馈表单：%s" % (
                    datetime.datetime.now(), cmd.primary_key,
                    mq_config['mq_command_exchange_name'], response_port + "," + dispatch_key,
                    tf.MessageToString(cmd, as_utf8=True)))
        elif cmd.statu.execute_statu == message.eError:
            log.error("取车指令 %s  唯一码:%s    检查失败!\n 发送反馈    交换机:%s  发送反馈端口:%s    反馈表单：%s" % (
                datetime.datetime.now(), cmd.primary_key,
                mq_config['mq_command_exchange_name'], response_port + "," + dispatch_key,
                tf.MessageToString(cmd, as_utf8=True)))
        else:
            log.critical("取车指令 %s  唯一码:%s    检查失败!\n 发送反馈    交换机:%s  发送反馈端口:%s    反馈表单：%s" % (
                    datetime.datetime.now(), cmd.primary_key,
                    mq_config['mq_command_exchange_name'], response_port + "," + dispatch_key,
                    tf.MessageToString(cmd, as_utf8=True)))
    if cmd.terminal_id != 0:
        g_rabbitmq.publish(mq_config['mq_command_exchange_name'], response_port, tf.MessageToString(cmd, as_utf8=True))
    else:
        return cmd


class MyHttpRequestHandler(BaseHTTPRequestHandler):
    def _send_cors_headers(self):
        """ Sets headers required for CORS """
        self.send_header('Content-type', 'application/json')
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "*")
        self.send_header("Access-Control-Allow-Headers", "Authorization, Content-Type")

    def do_POST(self):
        print('do_post:')
        datas = self.rfile.read(int(self.headers['content-length']))  # 固定格式，获取表单提交的数据
        self.send_response(200)  # 返回状态码
        self.end_headers()  # 返回响应头结束
        try:
            st = datas.decode('utf-8')
            res = json.loads(st)
            print(res)
            if ('data' in res.keys()) and ('InRecordId' in res['data'].keys()):
                primary_key = res['data']['InRecordId']
                table = message.pick_table()
                table.primary_key = primary_key
                # table.statu.table_process_mod = message.PROCESS_ONLY_TO_DISPATCH
                cmd = check_pick_command(tf.MessageToString(table, as_utf8=True))
                if cmd.statu.execute_statu == message.eNormal:
                    buf = {"Message": "取车成功", "Tag": 1}
                else:
                    buf = {"Message": cmd.statu.statu_description, "Tag": 0}
            else:
                buf = {"Message": "取车指令缺少关键字段Data或InRecordId!", "Tag": 0}
                # 以下是返回报文
        except Exception as e:
            buf = {"Message": "请求解析失败!", "Tag": 0}
        self.wfile.write(json.dumps(buf, ensure_ascii=False).encode('utf-8'))  # 发送json格式的返回包体
        print(buf)


def recv_post():
    ts = HTTPServer(('192.168.2.101', 8899), MyHttpRequestHandler)
    ts.serve_forever()

if __name__ == '__main__':
    config = parse_json.parse_json_with_comments('./config.json')
    log = HandleLog.get_logger(os.getcwd() + '_logs')

    # 数据库配置
    db_config = config['db_config']
    # rabbitmq配置
    mq_config = config['mq_config']
    # 日志记录
    # 消费指令消息
    cmd_callbacks = [["user_park_command_request_queue", check_park_command],
                     ["user_pick_command_request_queue", check_pick_command]]
    if config['project_name'] == 'xm_sgj':
        checker = CHE_XmSgj.CommandChecker(db_config)
    elif config['project_name'] == 'gy_bhhy':
        checker = CHE_GyBhhy.CommandChecker(db_config)
    elif config['project_name'] == 'zkxy':
        checker = CHE_ZKXY.CommandChecker(db_config)

    g_rabbitmq = CM.RabbitAsyncCommunicator(mq_config['mq_ip'], mq_config['mq_port'], mq_config['mq_user'], mq_config['mq_password'])
    g_rabbitmq.Init(cmd_callbacks, mq_config['mq_statu_exchange_keys'],"checker")
    for ex,key in mq_config['mq_statu_exchange_keys']:
        if key.find("dispatch") >= 0:
            g_rabbitmq.bind_statu_callback(ex,key,checker.receive_dispatch_statu)
        if key.find("measure") >= 0:
            g_rabbitmq.bind_statu_callback(ex,key,checker.receive_measure_info)

    # t1 = threading.Thread(target=recv_post)
    g_rabbitmq.start()
    # t1.start()
    g_rabbitmq.join()
    # t1.join()