hn3d-server-control.git

			@@ -1,175 +1,142 @@
			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 threading import Thread
			from datetime import datetime

			from zmq import device


			from .engine.DeviceManager import DeviceManager
			from .engine.MessageQueueThread import MessageQueueThread
			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):
			def __init__(self,
			MQhost:str,
			MQuser:str,
			MQpassword:str,
			WEBhost:str,
			WEBport:int,
			save_folder:str,
			server_name:str
			):
			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.MQhost = MQhost
			self.MQuser = MQuser
			self.MQpassword = MQpassword
			self.WEBhost = WEBhost
			self.WEBport = WEBport
			self.save_folder = save_folder
			self.server_name = "server0"


			self.route('/api', methods=['POST'])(self.api)
			# 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
			# self._def_routes()

			# queue
			self.command_request_queue = queue.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)
			# 连接队列服务器
			self.connection = pika.BlockingConnection(self.parameters)
			connection = pika.BlockingConnection(self.parameters)
			print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
			connection.close()

			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 _receiver_Thread_Launcher(self):

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

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

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

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

			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()
			self.MsgDecoder_Thread.setDaemon(True)
			self.MsgDecoder_Thread.setName("decode_thread")
			self.MsgDecoder_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):
			def _sender_Thread_Launcher(self):

			self.command_request_queue.put(ths.Message(None,None,req.command,req.target))
			return
			# launcher command 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 _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()
			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)

			<<<<<<< HEAD

			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()
			=======
			>>>>>>> aa8a0ba3a86ff5dbbdfcd68c72a5e70e1c52e877