From e839f2e83fed7eaceb677242f5f9650a79fb8c43 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期五, 04 三月 2022 14:16:57 +0800
Subject: [PATCH] temp commit
---
src/Server.py | 164 +++++++++++++++++++++++++++++++++++++-----------------
1 files changed, 112 insertions(+), 52 deletions(-)
diff --git a/src/Server.py b/src/Server.py
index b2e41a4..d4683e5 100644
--- a/src/Server.py
+++ b/src/Server.py
@@ -1,56 +1,135 @@
-from posixpath import abspath
import sys # NOQA: E402
import os
-from flask import Flask, request
+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 engine.DeviceManager import DeviceManager
-from engine.MessageQueueThread import MessageQueueThread
from protos import aerial_pb2 as pb
+from engine import MessageThreads as ths
class MyServer(Flask):
- def __init__(self, host, user, password, queue, port):
+ def __init__(self, MQhost, MQuser, MQpassword, WEBhost, WEBport, SAVEfolder):
Flask.__init__(self, __name__)
- self.host = host
- self.user = user
- self.password = password
- self.queue = queue
- self.port = port
-
+ # 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()
- self._device_manager = DeviceManager(host, user, password)
+ # 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):
- if pb.SubCommand.POWER_ON_SUB_COMMAND == req.sub_command:
- print('power on message received')
- self._device_manager.get_device(req.device_id,save_folder)
- elif pb.SubCommand.POWER_OFF_SUB_COMMAND == req.sub_command:
- device = self._device_manager.get_device(req.device_id,save_folder)
- device['power_on'] = False
- print('power off message received')
+
+ 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)
+ dispatcher(req, resp) # call the function
+
resp.checksum = 12345
response = Flask.response_class(
resp.SerializeToString(),
@@ -61,39 +140,22 @@
def hello(self):
return 'Hello, This is acquisition demo API!'
- def run_rabbitmq(self, host: str, user: str, password: str, queue: str) -> None:
- self._message_queue_thread = MessageQueueThread(
- host, user, password, queue, self._device_manager, save_folder)
- self._message_queue_thread.start()
+ 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.host, self.user, self.password, self.queue)
- self.run(debug=False, host="0.0.0.0", port=self.port)
-
- def send_controll_message(self, device_id, msg):
- device = self._device_manager.get_device(device_id,save_folder)
- if device is None:
- channel = device.controll_channel
- channel.basic_publish(
- exchange='',
- routing_key='',
- body=msg.SerializeToString())
-
- def send_dust_cleaner_cmd(self, device_id, sub_cmd, period):
- msg = pb.ControllMessage()
- msg.sys_command = pb.ControlSysCommand.DUST_CLEANER_COMMAND
- msg.sub_command = sub_cmd
- data_msg = pb.DustCleanerMessage()
- data_msg.period = period
- data1_str = data_msg.SerializeToString()
- msg.data1.length = len(data1_str)
- msg.data1.data = data1_str
+ 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] [QUEUEname] [WEBport] [SAVEfolder]")
- print("EXAMPLE: Server.py localhost chai password123 aerial_rpc 5000 ./Images")
+ 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:
@@ -102,14 +164,12 @@
MQhost = str(sys.argv[1])
MQuser = str(sys.argv[2])
MQpassword = str(sys.argv[3])
- QUEUEname = str(sys.argv[4])
+ WEBhost = str(sys.argv[4])
WEBport = int(sys.argv[5])
SAVEfolder = str(sys.argv[6])
- global save_folder
- save_folder = SAVEfolder
except Exception as e:
print(e)
help()
- server = MyServer(host="localhost", user='chai', password='password123', queue='aerial_rpc', port=5000)
+ server = MyServer(MQhost, MQuser, MQpassword, WEBhost, WEBport, SAVEfolder)
server.run_all()
--
Gitblit v1.9.1