import sys # NOQA: E402
|
import os
|
from flask import Flask
|
from numpy import true_divide
|
import pika
|
import uuid
|
import queue
|
from werkzeug.wrappers import request
|
|
from zmq import device
|
|
|
batfile_dir = os.path.join(__file__, "..", "protos/compile.bat" )
|
batfile_path = os.path.abspath(batfile_dir)
|
os.system(batfile_path)
|
|
from protos import aerial_pb2 as pb
|
from engine import MessageThreads as ths
|
|
|
|
class MyServer(Flask):
|
def __init__(self, MQhost, MQuser, MQpassword, WEBhost, WEBport, SAVEfolder):
|
Flask.__init__(self, __name__)
|
# rabbit MQ
|
self.MQhost = MQhost
|
self.MQuser = MQuser
|
self.MQpassword = MQpassword
|
self.parameters = None
|
# Web server
|
self.WEBhost = WEBhost
|
self.WEBport = WEBport
|
self.SAVEfolder = SAVEfolder
|
self.route('/api', methods=['POST'])(self.api)
|
self.route('/', methods=['GET'])(self.hello)
|
self._def_routes()
|
# threads
|
self._device_state_reveiver_thread = None
|
self._alarm_state_receiver_thread = None
|
self._command_request_sender_thread = None
|
self._command_response_receiver_thread = None
|
# 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)
|
# 连接队列服务器
|
self.connection = pika.BlockingConnection(self.parameters)
|
print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
|
|
def _start_device_state_up_msg_receiver_thread(self):
|
"""start device state thread"""
|
if self._device_state_reveiver_thread is None:
|
self._device_state_reveiver_thread = \
|
ths.device_state_receiverThread(self.parameters,"server")
|
self._device_state_reveiver_thread.setDaemon(True)
|
self._device_state_reveiver_thread.start()
|
|
def _stop_device_state_up_msg_receiver_thread(self):
|
"""stop device state thread"""
|
if self._device_state_reveiver_thread is not None:
|
self._device_state_reveiver_thread.stop()
|
self._device_state_reveiver_thread = None
|
|
def _start_alarm_state_up_msg_receiver_thread(self):
|
"""start alarm state thread"""
|
if self._alarm_state_receiver_thread is None:
|
self._alarm_state_receiver_thread = \
|
ths.alarm_state_receiverThread(self.parameters,"server")
|
self._alarm_state_receiver_thread.setDaemon(True)
|
self._alarm_state_receiver_thread.start()
|
|
def _stop_alarm_state_up_msg_receiver_thread(self):
|
"""stop alarm state thread"""
|
if self._alarm_state_receiver_thread is not None:
|
self._alarm_state_receiver_thread.stop()
|
self._alarm_state_receiver_thread = None
|
|
def _start_command_response_receiver_thread(self):
|
"""start command response thread"""
|
if self._command_response_receiver_thread is None:
|
self._command_response_receiver_thread = \
|
ths.command_response_receiverThread(self.parameters,"server")
|
self._command_response_receiver_thread.setDaemon(True)
|
self._command_response_receiver_thread.start()
|
|
def _stop_command_response_receiver_thread(self):
|
"""stop command response thread"""
|
if self._command_response_receiver_thread is not None:
|
self._command_response_receiver_thread.stop()
|
self._command_response_receiver_thread = None
|
|
def _start_command_request_sender_thread(self):
|
"""start command request thread"""
|
if self._command_request_sender_thread is None:
|
self._command_request_sender_thread = \
|
ths.command_request_senderThread(self.parameters,self.command_request_queue,"server")
|
self._command_request_sender_thread.setDaemon(True)
|
self._command_request_sender_thread.start()
|
|
def _stop_command_request_sender_thread(self):
|
"""stop command request thread"""
|
if self._command_request_sender_thread is not None:
|
self._command_request_sender_thread.stop()
|
self._command_request_sender_thread = None
|
|
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)
|
|
|
if __name__ == "__main__":
|
def help():
|
print("Usage: Server.py [MQhost] [MQuser] [MQpassword] [WEBhost] [WEBport] [SAVEfolder]")
|
print("EXAMPLE: Server.py localhost chai password123 0.0.0.0 5000 ./Received_data/")
|
exit(1)
|
|
if len(sys.argv) != 7:
|
help()
|
try:
|
MQhost = str(sys.argv[1])
|
MQuser = str(sys.argv[2])
|
MQpassword = str(sys.argv[3])
|
WEBhost = str(sys.argv[4])
|
WEBport = int(sys.argv[5])
|
SAVEfolder = str(sys.argv[6])
|
|
except Exception as e:
|
print(e)
|
help()
|
server = MyServer(MQhost, MQuser, MQpassword, WEBhost, WEBport, SAVEfolder)
|
server.run_all()
|
