import sys # NOQA: E402 import os from flask import Flask, request import pika import uuid import queue from threading import Thread from datetime import datetime 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 class MyServer(Flask): def __init__(self, MQhost:str, MQuser:str, MQpassword:str, WEBhost:str, WEBport:int, save_folder:str, server_name:str ): Flask.__init__(self, __name__) self.MQhost = MQhost self.MQuser = MQuser self.MQpassword = MQpassword self.WEBhost = WEBhost self.WEBport = WEBport self.save_folder = save_folder self.server_name = server_name self.replay_host = "http://192.168.1.17:8001/admin/business/colletmeta/getDeviceMsg" 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() # queue self.LogsReceiver_queue = queue.Queue() self.ImagesReceiver_queue = queue.Queue() self.PlysReceiver_queue = queue.Queue() self.CommandsSender_queue = queue.Queue() def _connect_to_MQ(self): credentials = pika.PlainCredentials(self.MQuser, self.MQpassword) self.parameters = pika.ConnectionParameters( credentials=credentials, host=self.MQhost,heartbeat=0) # 连接队列服务器 try: self.connection = pika.BlockingConnection(self.parameters) print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}") self.connection.close() except Exception : print("Can't connecting to the MQ server") exit(1) def _receiver_Thread_Launcher(self): # launch logs receiver thread self.LogsReceiver_Thread = ths.receiverThread( MQparams=self.parameters, exchange_name='logs_topic', exchange_type='topic', MQqueue_name=f'{self.server_name}_logs_queue', binding_keys=[f'{self.server_name}.logs.test0'], message_queue=self.LogsReceiver_queue ) self.LogsReceiver_Thread.setDaemon(True) self.LogsReceiver_Thread.setName("Thread-logsReceiver") self.LogsReceiver_Thread.start() # launch images receiver thread self.ImagesReceiver_Thread = ths.receiverThread( MQparams=self.parameters, exchange_name='images_topic', exchange_type='topic', MQqueue_name=f'{self.server_name}_images_queue', binding_keys=[f"{self.server_name}.images.test0"], message_queue=self.ImagesReceiver_queue ) self.ImagesReceiver_Thread.setDaemon(True) self.ImagesReceiver_Thread.setName("Thread-imagesReceiver") self.ImagesReceiver_Thread.start() # launch plys receiver thread self.PlysReceiver_Thread = ths.receiverThread( MQparams=self.parameters, exchange_name='plys_topic', exchange_type='topic', MQqueue_name=f'{self.server_name}_plys_queue', binding_keys=[f'{self.server_name}.plys.test0'], message_queue=self.PlysReceiver_queue ) self.PlysReceiver_Thread.setDaemon(True) self.PlysReceiver_Thread.setName("Thread-plysReceiver") self.PlysReceiver_Thread.start() # launch decode thread queue_list = [self.LogsReceiver_queue,self.ImagesReceiver_queue,self.PlysReceiver_queue] queue_tags = ["Logs", "Images", "Plys"] self.MsgDecoder_Thread = ths.decodeThread( queue_list= queue_list, queue_tags= queue_tags, task_folder= self.task_uuid_folder , replay_host=self.replay_host ) self.MsgDecoder_Thread.setDaemon(True) self.MsgDecoder_Thread.setName("Thread-dataDecoder") self.MsgDecoder_Thread.start() def _sender_Thread_Launcher(self): # launcher commands sender thread self.CommandsSender_Thread = ths.senderThread( MQparams=self.parameters, exchange_name='commands_topic', exchange_type='topic', message_queue=self.CommandsSender_queue ) self.CommandsSender_Thread.setDaemon(True) self.CommandsSender_Thread.setName("Thread-commandsSender") self.CommandsSender_Thread.start() 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() self._sender_Thread_Launcher() def run_all(self): self.run_rabbitmq() self.run(debug=False, host=self.WEBhost, port=self.WEBport)