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, host, user, password, queue, port, SAVEfolder):
|
Flask.__init__(self, __name__)
|
self.host = host
|
self.user = user
|
self.password = password
|
self.queue = queue
|
self.port = port
|
global save_folder
|
save_folder = SAVEfolder
|
|
|
self.route('/api', methods=['POST'])(self.api)
|
self.route('/', methods=['GET'])(self.hello)
|
self._def_routes()
|
# threads
|
|
# queue
|
self.command_request_queue = queue.Queue()
|
|
def _connect_to_MQ(self):
|
credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
|
self.parameters = pika.ConnectionParameters(
|
credentials=credentials, host=self.MQhost)
|
# 连接队列服务器
|
connection = pika.BlockingConnection(self.parameters)
|
print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
|
connection.close()
|
|
|
|
def _message_receiver_Thread_launcher(self):
|
|
def callback(ch,method,properties,body):
|
timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
|
print(f"[x] {timestamp} | [{method.routing_key}]:{body}")
|
|
target_args = (self.parameters,
|
"message_topic",
|
"topic",
|
"message_queue",
|
["server.server.message", "server.server0.message"],
|
callback)
|
|
self.message_receiver_thread = Thread(target=ths.message_receiver,
|
args=target_args)
|
self.message_receiver_thread.setDaemon(True)
|
self.message_receiver_thread.start()
|
|
|
def _image_receiver_Thread_launcher(self):
|
|
def callback(ch,method,properties,body):
|
|
timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
|
print(f"[x] {timestamp} | [{method.routing_key}]:{body}")
|
|
target_args = (self.parameters,
|
"message_topic",
|
"topic",
|
"message_queue",
|
["server.server.message", "server.server0.message"],
|
callback)
|
|
self.message_receiver_thread = Thread(target=ths.message_receiver,
|
args=target_args)
|
self.message_receiver_thread.setDaemon(True)
|
self.message_receiver_thread.start()
|
|
|
|
def _def_routes(self):
|
self.routes = {
|
pb.SysCommand.SYS_COMMAND: self._on_sys_command,
|
}
|
|
def _on_sys_command(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
|
|
self.command_request_queue.put(ths.Message(None,None,req.command,req.target))
|
return
|
|
def _default(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
|
print('onknown command')
|
|
def api(self):
|
|
req = pb.RequestCommand()
|
req.ParseFromString(request.data)
|
|
# a function: _on_sys_command, refer to self._on_sys_command
|
dispatcher = self.routes.get(req.sys_command, self._default)
|
|
resp = pb.ResponseCommand()
|
dispatcher(req, resp) # call the function
|
|
resp.checksum = 12345
|
response = Flask.response_class(
|
resp.SerializeToString(),
|
mimetype='application/x-protobuf'
|
)
|
return response, 200
|
|
def hello(self):
|
return 'Hello, This is acquisition demo API!'
|
|
def run_rabbitmq(self):
|
self._connect_to_MQ()
|
self._start_device_state_up_msg_receiver_thread()
|
self._start_alarm_state_up_msg_receiver_thread()
|
self._start_command_response_receiver_thread()
|
self._start_command_request_sender_thread()
|
|
def run_all(self):
|
self.run_rabbitmq()
|
self.run(debug=False, host=self.WEBhost, port=self.WEBport)
|
