import sys # NOQA: E402
|
import os
|
from flask import Flask
|
import pika
|
import uuid
|
import queue
|
from werkzeug.wrappers import request
|
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
|
|
|
|
|
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.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)
|
# 连接队列服务器
|
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'logs_queue',
|
binding_keys=['server.logs.test0'],
|
message_queue=self.LogsReceiver_queue
|
)
|
self.LogsReceiver_Thread.setDaemon(True)
|
self.LogsReceiver_Thread.setName("logs_thread")
|
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'images_queue',
|
binding_keys=["server.images.test0"],
|
message_queue=self.ImagesReceiver_queue
|
)
|
self.ImagesReceiver_Thread.setDaemon(True)
|
self.ImagesReceiver_Thread.setName("images_thread")
|
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'plys_queue',
|
binding_keys=['server.plys.test0'],
|
message_queue=self.PlysReceiver_queue
|
)
|
self.PlysReceiver_Thread.setDaemon(True)
|
self.PlysReceiver_Thread.setName("plys_thread")
|
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,
|
save_path= self.save_folder
|
)
|
|
self.MsgDecoder_Thread.setDaemon(True)
|
self.MsgDecoder_Thread.setName("decode_thread")
|
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("commands_thread")
|
self.CommandsSender_Thread.start()
|
|
|
def hello(self):
|
return 'Hello, This is acquisition demo API!'
|
|
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)
|
