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)
|
