add taskuuid

wu_xinjun

2022-06-24 74c96c72625aff6613d69c7168e1a7d76da22870

 src/Server.py

@@ -1,10 +1,9 @@
import sys              # NOQA: E402    
import os
from flask import Flask
from flask import Flask, request
import pika
import uuid
import queue
from werkzeug.wrappers import request
from threading import Thread
from datetime import datetime

@@ -13,6 +12,7 @@
from .engine.DeviceManager import DeviceManager
from .engine import MessageThreads as ths
from .protos import aerial_pb2 as pb
from .utils.tools import update_folder_for_today



@@ -35,9 +35,22 @@
        self.WEBport = WEBport
        self.save_folder = save_folder
        self.server_name = server_name
        

        # self.route('/api', methods=['POST'])(self.api)
        today = datetime.now().strftime('%Y-%m-%d')
        today_folder = os.path.join(self.save_folder,today)

        self.today = {"today":today,
                    "today_folder":today_folder,
                    "sup_folder":self.save_folder,
                    "sub_folder": ""}

        self.task_uuid_folder = {"init":os.path.join(self.today["today_folder"],"init")}

        if not os.path.exists(self.task_uuid_folder["init"]):
            os.makedirs(self.task_uuid_folder["init"])


        self.route('/api', methods=['POST'])(self.api)
        self.route('/', methods=['GET'])(self.hello)
        # self._def_routes()

@@ -51,7 +64,7 @@
    def _connect_to_MQ(self):
        credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
        self.parameters = pika.ConnectionParameters(
            credentials=credentials, host=self.MQhost)
            credentials=credentials, host=self.MQhost,heartbeat=0)
        # 连接队列服务器
        try:
            self.connection = pika.BlockingConnection(self.parameters)
@@ -110,7 +123,7 @@
        self.MsgDecoder_Thread = ths.decodeThread(
                                    queue_list= queue_list,
                                    queue_tags= queue_tags,
                                    save_path= self.save_folder           
                                    task_folder= self.task_uuid_folder        
                                    )

        self.MsgDecoder_Thread.setDaemon(True)
@@ -135,6 +148,52 @@
    def hello(self):
        return 'Hello, This is acquisition demo API!'

    def api(self):

        try:
            success, response_bytes = self.verify_commands(request.data)
        except:
            response_bytes = b'command error!'

        response = Flask.response_class(
            response_bytes,
            mimetype='application/x-bytes'
        )
        return response,200


    def verify_commands(self,input_command:bytes):

        # parse the command from bytes to the pb object
        req_command = pb.CommandMessage()
        req_command.ParseFromString(input_command)

        # extracting some info for use
        source = req_command.source
        # generatetime = req_command.generatetime
        # syscommand = req_command.syscommand
        # subcommand = req_command.subcommand
        targetclient = req_command.targetclient
        # returnserver = req_command.returnserver
        taskuuid = req_command.taskuuid

        update_folder_for_today(self.today)
        task_path = os.path.join(self.today["today_folder"],taskuuid)

        self.task_uuid_folder[taskuuid] = task_path

        if not os.path.exists(task_path):
            os.makedirs(task_path)

        curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
        print(f"[x] {curr_time}|(From {source}): task [{taskuuid}] received and sending it to [{targetclient}]." )

        self.CommandsSender_queue.put([targetclient,input_command])
        response_bytes = bytes(f"Receive command success! The data will be save in [{os.path.abspath(task_path)}]",\
                        encoding="utf8")

        return True, response_bytes

    def run_rabbitmq(self):
        self._connect_to_MQ()
        self._receiver_Thread_Launcher()