From baee79553738a212d82a12a1966fe57ce4acb937 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期五, 10 六月 2022 18:09:30 +0800
Subject: [PATCH] -_-
---
src/Server1.py | 134 ++++++++++++++
src/protos/aerial.proto | 36 +++
src/engine/MessageThreads.py | 350 +++++++++++++++++++++++++++++++++-----
src/engine/devices.py | 10
4 files changed, 465 insertions(+), 65 deletions(-)
diff --git a/src/Server1.py b/src/Server1.py
new file mode 100644
index 0000000..1ba60f2
--- /dev/null
+++ b/src/Server1.py
@@ -0,0 +1,134 @@
+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)
+
diff --git a/src/engine/MessageThreads.py b/src/engine/MessageThreads.py
index 903dea8..2b33936 100644
--- a/src/engine/MessageThreads.py
+++ b/src/engine/MessageThreads.py
@@ -1,15 +1,19 @@
import os
+import queue
import threading
import pika
import time
-import os
+from time import sleep
+import sys
+from datetime import datetime
+from typing import List
-# add python path of src to sys.path
-src_path = os.path.join(__file__, *(['..'] * 2))
-src_path = os.path.abspath(src_path)
-sys.path.insert(0, src_path)
+# # add python path of src to sys.path
+# src_path = os.path.join(__file__, *(['..'] * 2))
+# src_path = os.path.abspath(src_path)
+# sys.path.insert(0, src_path)
-from protos import aerial_pb2 as pb
+from ..protos import aerial_pb2 as pb
@@ -19,17 +23,27 @@
# self.sender_key = sender_key
# self.body = body
# self.routing_key = routing_key
-# 鏍规嵁鏂扮殑鏁版嵁缁撴瀯鏀瑰啓run
+
class senderThread(threading.Thread):
- def __init__(self, MQparameters,input_msg_queue,sender_id,exchangeName,exchangeType):
- super(senderThread,self).__init__()
- self.parameters = MQparameters
- self.queue = input_msg_queue
- self.exchangeName = exchangeName
- self.exchangeType = exchangeType
- self.sender_id = sender_id
- self.is_stop = False
- self.connection = pika.BlockingConnection(self.parameters)
+
+ def __init__(self,
+ MQparams: pika.ConnectionParameters,
+ exchange_name: str,
+ exchange_type:str,
+ # routing_key:str,
+ message_queue:queue.Queue,
+ ):
+ super(senderThread,self).__init__()
+
+ self.MQparams = MQparams
+ self.exchange_name = exchange_name
+ self.exchange_type = exchange_type
+ # self.routing_key = routing_key
+ self.message_queue = message_queue
+
+ # connect to RabbitMQ
+ self.is_stop = False
+ self.connection = pika.BlockingConnection(self.MQparams)
def stop(self):
"""stop the thread"""
@@ -39,42 +53,62 @@
def run(self):
while not self.is_stop:
+ # declare a exchange
channel = self.connection.channel()
- channel.exchange_declare(exchange=self.exchangeName, exchange_type=self.exchangeType)
+ channel.exchange_declare(exchange=self.exchange_name,
+ exchange_type=self.exchange_type)
while True:
try:
- if not self.queue.empty():
- data= self.queue.get()
- if isinstance(data, pb.CapturedData):
- msg = pb.DataMessage()
- else:
- body = msg.body
- routing_key = msg.routing_key
- channel.basic_publish(exchange=self.exchangeName, routing_key=routing_key, body=body)
-
- response_tag = f"{time.asctime()},{self.sender_id} send {self.exchangeName} message to {routing_key}"
+ if not self.message_queue.empty():
- print(response_tag)
+ # decode the routing key and message body from tuple list
+ message = self.message_queue.get()
+ msg_routing_key = message[0]
+ msg_body = message[1]
- elif self.queue.empty():
+ # publish a PERSISTENT message
+ channel.basic_publish(
+ exchange = self.exchange_name,
+ routing_key = msg_routing_key,
+ body = msg_body,
+ properties=pika.BasicProperties(
+ delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
+ )
+ )
+
+ timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ caption = f"[x] {timestamp} | Sending message to[{msg_routing_key}]")
+ print(caption)
+
+ else:
time.sleep(1)
+
except Exception as e:
print(e)
break
-# 鏍规嵁鏂扮殑鏁版嵁缁撴瀯鏀瑰啓run
class receiverThread(threading.Thread):
- def __init__(self, MQparameters,received_msg_queue,output_response_queue,exchangeName,exchangeType,receiver_id):
- super(receiverThread,self).__init__()
- self.parameters = MQparameters
- self.queue = received_msg_queue
- self.response_queue = output_response_queue
- self.exchangeName = exchangeName
- self.exchangeType = exchangeType
- self.receiver_id = receiver_id
- self.is_stop = False
- self.connection = pika.BlockingConnection(self.parameters)
+ def __init__(self,
+ MQparams: pika.ConnectionParameters,
+ exchange_name: str,
+ exchange_type:str,
+ MQqueue_name:str,
+ binding_keys:List(str),
+ message_queue:queue.Queue,
+ ):
+ super(senderThread,self).__init__()
+
+ self.MQparams = MQparams
+ self.exchange_name = exchange_name
+ self.exchange_type = exchange_type
+ self.MQqueue_name = MQqueue_name
+ self.binding_keys = binding_keys
+ self.message_queue = message_queue
+
+ # connect to RabbitMQ
+ self.is_stop = False
+ self.connection = pika.BlockingConnection(self.MQparams)
def stop(self):
"""stop the thread"""
@@ -83,25 +117,178 @@
def run(self):
+
while not self.is_stop:
-
+ # declare a exchange
channel = self.connection.channel()
- channel.exchange_declare(exchange=self.exchangeName, exchange_type=self.exchangeType)
+ channel.exchange_declare(exchange=self.exchange_name,
+ exchange_type=self.exchange_type)
- result = channel.queue_declare(queue='', exclusive=True, durable=True)
- queue_name = result.method.queue
+ # declare the queue and bind it with keys
+ channel.queue_declare(queue=self.MQqueue_name, durable=True)
+ for binding_key in self.binding_keys:
+ channel.queue_bind(exchange=self.exchangeName, queue=self.MQqueue_name, routing_key=binding_key)
- channel.queue_bind(exchange=self.exchangeName, queue=queue_name, routing_key=self.receiver_id)
-
- channel.basic_consume(queue=queue_name, \
- on_message_callback=self.callback)
-
+ # consume the message
channel.basic_qos(prefetch_count=1)
+ channel.basic_consume(
+ queue=self.MQqueue_name,
+ on_message_callback=self.callback
+ )
channel.start_consuming()
- def callback(ch, method, properties, body):
- raise NotImplementedError(
- "Need to implement the callback function")
+ def callback(self,ch, method, properties, body):
+ self.message_queue.put(body)
+ ch.basic_ack(delivery_tag = method.delivery_tag)
+
+class decodeThread(threading.Thread):
+ def __init__(self,
+ queue_list:List(queue.Queue),
+ queue_tags:List(str),
+ save_path:str):
+ super().__init__()
+
+ self.queue_list = queue_list
+ self.queue_tags = queue_tags
+ self.queue_num = 0
+ self.save_path = save_path
+ self.is_stop = False
+
+ assert len(self.queue_list) == len(self.queue_tags)
+
+
+ def stop(self):
+ """stop the thread"""
+ self.is_stop = True
+
+ def run(self) -> None:
+
+ # creating a path for saving data
+ time_tag = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ self.save_root_folder = os.path.join(self.save_path,time_tag)
+ os.makedirs(self.save_root_folder)
+
+ self.logsmsg_folder = os.path.join(self.save_root_folder,"LogsMsg")
+ self.imagesmsg_folder = os.path.join(self.save_root_folder,"ImagesMsg")
+ self.plysmsg_folder = os.path.join(self.save_root_folder,"PlysMsg")
+ self.commandsmsg_folder = os.path.join(self.save_root_folder,"CommandsMsg")
+
+ sub_folder_list = [self.logsmsg_folder,self.imagesmsg_folder,
+ self.plysmsg_folder,self.commandsmsg_folder]
+
+ for folder in sub_folder_list:
+ os.mkdir(folder)
+ print(f"[NOTICE] Creating folder: {folder}")
+
+ # process the queue list
+ self.queue_num = len(self.queue_list)
+ while self.queue_num:
+ i = 0
+ while True:
+ # select a queue
+ i_tag = i % self.queue_num
+ queue = self.queue_list[i_tag]
+
+ # make sure the queue is not empty
+ while not queue.empty():
+ # decode the msg using cooresponding method
+ queue_tag = self.queue_tags[i_tag]
+ decode_method = self.decode_select(queue_tag)
+ decode_method(queue)
+ # next queue
+ i += 1
+
+
+ def decode_select(self,queue_tags:str):
+ route_dic = {
+ "Logs":self.decode_LogsMsg,
+ "Images":self.decode_ImagesMsg,
+ "Plys":self.decode_PlysMsg,
+ "Commands":self.decode_CommandsMsg
+ }
+ return route_dic[queue_tags]
+
+ def decode_LogsMsg(self,queue):
+ nums_limit = 10
+ count = 0
+ while not queue.empty() and count < nums_limit:
+
+ # extracting the msg from queue
+ msg = queue.pop()
+ msg_pb = pb.LogsMessage()
+ msg_pb.ParseFromString(msg)
+
+ # save the raw pb msg
+ raw_file_name = os.path.join(self.logsmsg_folder,f"{curr_time}.LogsMsg")
+ with open(raw_file_name, 'wb') as file:
+ file.write(msg)
+
+ # process the msg
+ msg_source = msg_pb.source
+ msg_timestamp = msg_pb.timestamp
+ msg_context = msg_pb.context
+
+ # display the info to the console
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ print(f"[x] {curr_time}|(From {msg_source}): {msg_context}" )
+
+ # next msg
+ count += 1
+
+
+ def decode_ImagesMsg(self,queue):
+
+ while not queue.empty():
+
+ # extracting the msg from queue
+ msg = queue.pop()
+ msg_pb = pb.ImagesMessage()
+ msg_pb.ParseFromString(msg)
+
+ # save the raw pb msg
+ raw_file_name = os.path.join(self.imagesmsg_folder,f"{curr_time}.ImagesMsg")
+ with open(raw_file_name, 'wb') as file:
+ file.write(msg)
+
+ # process the msg
+ msg_source = msg_pb.source
+ msg_timestamp = msg_pb.timestamp
+
+ # process the images
+ id = 0
+ for image in msg_pb.Image:
+ imagelegth = image.length
+ image_byte = image.data
+
+ # save the images
+ if (image_byte is None) | (len(image_byte) == 0):
+ pass
+ else:
+ filename = f"{msg_source}_{msg_timestamp}_image{id}.jpg"
+ filepath = os.path.join(self.imagesmsg_folder,filename)
+ with open(filepath, 'wb') as f:
+ f.write(image)
+
+ # display the info to the console
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ print(f"[x] {curr_time}|(From {msg_source}): image{id} saved!" )
+
+ # next image item
+ id += 1
+
+
+ def decode_PlysMsg(self,queue):
+
+
+
+
+ pass
+
+ def decode_CommandsMsg(self,queue):
+
+ pass
+
+
# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
class device_state_senderThread(senderThread):
@@ -260,6 +447,65 @@
pass
+def message_sender(MQparams: pika.ConnectionParameters,
+ exchange_name: str,
+ exchange_type:str,
+ routing_key:str,
+ message:str,
+ ):
+ # connect to RabbitMQ
+ connection = pika.BlockingConnection(MQparams)
+
+ # declare a exchange
+ channel = connection.channel()
+ channel.exchange_declare(exchange=exchange_name,
+ exchange_type=exchange_type)
+ # publish a PERSISTENT message
+ channel.basic_publish(
+ exchange=exchange_name,
+ routing_key=routing_key,
+ body=message,
+ properties=pika.BasicProperties(
+ delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
+ )
+ )
+ # close the connection
+ connection.close()
+
+def message_receiver(MQparams: pika.ConnectionParameters,
+ exchange_name: str,
+ exchange_type:str,
+ queue_name:str,
+ binding_keys:List[str],
+ callback_fn:callable
+ ):
+ # connect to RabbitMQ
+ connection = pika.BlockingConnection(MQparams)
+
+ # declare a exchange
+ channel = connection.channel()
+ channel.exchange_declare(exchange=exchange_name,
+ exchange_type=exchange_type)
+
+ # declare a durable queue
+ channel.queue_declare(queue=queue_name,durable=True)
+
+ # literate the binding keys and bind the queue
+ for binding_key in binding_keys:
+ channel.queue_bind(queue=queue_name,exchange=exchange_name,
+ routing_key=binding_key)
+
+ # waiting for consuming the messages
+ channel.basic_qos(prefetch_count=1)
+ channel.basic_consume(queue=queue_name,
+ auto_ack=False,
+ on_message_callback=callback_fn
+ )
+ channel.start_consuming()
+
+
+
+
diff --git a/src/engine/devices.py b/src/engine/devices.py
index cb1b5e2..aacc2ac 100644
--- a/src/engine/devices.py
+++ b/src/engine/devices.py
@@ -5,12 +5,12 @@
import numpy as np
import os, sys
-# add python path of src to sys.path
-src_path = os.path.join(__file__, *(['..'] * 2))
-src_path = os.path.abspath(src_path)
-sys.path.insert(0, src_path)
+# # add python path of src to sys.path
+# src_path = os.path.join(__file__, *(['..'] * 2))
+# src_path = os.path.abspath(src_path)
+# sys.path.insert(0, src_path)
-from protos import aerial_pb2 as pb
+from ..protos import aerial_pb2 as pb
class cameraController(object):
diff --git a/src/protos/aerial.proto b/src/protos/aerial.proto
index 6929336..1a81118 100644
--- a/src/protos/aerial.proto
+++ b/src/protos/aerial.proto
@@ -116,18 +116,38 @@
CommandData CommandData =4;
}
+message LogsMessage{
+ string source = 1;
+ string timestamp = 2;
+ string context = 3;
+}
+
+
+message ImagesMessage
+{
+ string source = 1;
+ string timestamp = 2;
+
+ message ImageItems
+ {
+ int32 length = 1; // 鏁版嵁闀垮害
+ bytes data = 2; // 鏁版嵁鍐呭
+ }
+
+ repeated ImageItems Image = 3;
+}
+
+
+message PlyMessage
+{
+ string source = 1;
+ string timestamp = 2;
+ bytes plydata = 3;
+}
-// message ImagesMessage
-// {
-// bytes image1 = 1;
-// bytes image2 = 2;
-// bytes image3 = 3;
-// bytes image4 = 4;
-// bytes image5 = 5;
-// }
// message MicroWaveMessage
// {
--
Gitblit v1.9.1