From d8e5c7c93ed28afa2af1d68ea20cff962599626f Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期二, 14 六月 2022 15:57:55 +0800
Subject: [PATCH] server enhanced!
---
src/Server.py | 217 +++++++++++++++++++++++-------------------------------
1 files changed, 92 insertions(+), 125 deletions(-)
diff --git a/src/Server.py b/src/Server.py
index 90dacf3..3b50887 100644
--- a/src/Server.py
+++ b/src/Server.py
@@ -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
--
Gitblit v1.9.1