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/engine/MessageThreads.py | 350 +++++++++++++++++++++++++++++++++++++++++++++++++--------
1 files changed, 298 insertions(+), 52 deletions(-)
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()
+
+
+
+
--
Gitblit v1.9.1