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/engine/MessageThreads.py | 709 ++++------------------------------------------------------
1 files changed, 53 insertions(+), 656 deletions(-)
diff --git a/src/engine/MessageThreads.py b/src/engine/MessageThreads.py
index 2b33936..04d86e2 100644
--- a/src/engine/MessageThreads.py
+++ b/src/engine/MessageThreads.py
@@ -7,6 +7,7 @@
import sys
from datetime import datetime
from typing import List
+import numpy as np
# # add python path of src to sys.path
# src_path = os.path.join(__file__, *(['..'] * 2))
@@ -14,6 +15,7 @@
# sys.path.insert(0, src_path)
from ..protos import aerial_pb2 as pb
+from ..utils.tools import write_ply
@@ -33,7 +35,7 @@
# routing_key:str,
message_queue:queue.Queue,
):
- super(senderThread,self).__init__()
+ super().__init__()
self.MQparams = MQparams
self.exchange_name = exchange_name
@@ -78,7 +80,7 @@
)
timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
- caption = f"[x] {timestamp} | Sending message to[{msg_routing_key}]")
+ caption = f"[x] {timestamp} | Sending message to[{msg_routing_key}]"
print(caption)
else:
@@ -94,10 +96,10 @@
exchange_name: str,
exchange_type:str,
MQqueue_name:str,
- binding_keys:List(str),
+ binding_keys:List[str],
message_queue:queue.Queue,
):
- super(senderThread,self).__init__()
+ super().__init__()
self.MQparams = MQparams
self.exchange_name = exchange_name
@@ -127,7 +129,7 @@
# 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.exchange_name, queue=self.MQqueue_name, routing_key=binding_key)
# consume the message
channel.basic_qos(prefetch_count=1)
@@ -143,8 +145,8 @@
class decodeThread(threading.Thread):
def __init__(self,
- queue_list:List(queue.Queue),
- queue_tags:List(str),
+ queue_list:List[queue.Queue],
+ queue_tags:List[str],
save_path:str):
super().__init__()
@@ -155,7 +157,6 @@
self.is_stop = False
assert len(self.queue_list) == len(self.queue_tags)
-
def stop(self):
"""stop the thread"""
@@ -197,8 +198,7 @@
decode_method(queue)
# next queue
i += 1
-
-
+
def decode_select(self,queue_tags:str):
route_dic = {
"Logs":self.decode_LogsMsg,
@@ -219,6 +219,7 @@
msg_pb.ParseFromString(msg)
# save the raw pb msg
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
raw_file_name = os.path.join(self.logsmsg_folder,f"{curr_time}.LogsMsg")
with open(raw_file_name, 'wb') as file:
file.write(msg)
@@ -235,7 +236,6 @@
# next msg
count += 1
-
def decode_ImagesMsg(self,queue):
while not queue.empty():
@@ -245,7 +245,8 @@
msg_pb = pb.ImagesMessage()
msg_pb.ParseFromString(msg)
- # save the raw pb msg
+ # save the raw pb msg
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
raw_file_name = os.path.join(self.imagesmsg_folder,f"{curr_time}.ImagesMsg")
with open(raw_file_name, 'wb') as file:
file.write(msg)
@@ -261,7 +262,7 @@
image_byte = image.data
# save the images
- if (image_byte is None) | (len(image_byte) == 0):
+ if (image_byte is None) | (len(image_byte) != imagelegth ):
pass
else:
filename = f"{msg_source}_{msg_timestamp}_image{id}.jpg"
@@ -276,656 +277,52 @@
# next image item
id += 1
-
def decode_PlysMsg(self,queue):
+ while not queue.empty():
+ # extracting the msg from queue
+ msg = queue.pop()
+ msg_pb = pb.PlyMessage()
+ msg_pb.ParseFromString(msg)
-
- pass
+ # save the raw pb msg
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ raw_file_name = os.path.join(self.plysmsg_folder,f"{curr_time}.PlysMsg")
+ 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 muti ply data
+ id = 0
+ for ply in msg_pb.Plydata:
+ data_legth = ply.length
+ data_byte = ply.data
+ rows = ply.rows
+ cols = ply.cols
+ order = ply.order
+ dt = ply.d_types
+
+ # save the pointcloud array data
+ if (data_byte is None) | (len(data_byte) != data_legth):
+ pass
+ else:
+ # frame the bytes to array
+ data_array = np.frombuffer(data_byte,dtype=dt)
+ data_array = data_array.reshape(rows,cols)
+ # save the data with .ply format
+ filename = f"{msg_source}_{msg_timestamp}_pointcloud{id}"
+ write_ply(data_array,self.plysmsg_folder,filename)
+
+ # next ply item
+ id += 1
def decode_CommandsMsg(self,queue):
- pass
+ raise NotImplementedError("NOT IMPLEMENT!")
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-class device_state_senderThread(senderThread):
- def __init__(self,
- MQparameters,
- input_msg_queue,
- sender_id,
- ):
- super(device_state_senderThread,self).__init__(
- MQparameters, \
- input_msg_queue,
- sender_id,
- "device_state",
- "fanout"
- )
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-class device_state_receiverThread(receiverThread):
- def __init__(self,
- MQparameters,
- receiver_id):
- super(device_state_receiverThread,self).__init__(
- MQparameters,
- None,
- None,
- "device_state",
- "fanout",
- receiver_id
- )
-
- def callback(self, ch, method, properties, body):
-
- response_tag = f"{time.asctime()} received: {body}."
- print(response_tag)
- ch.basic_ack(delivery_tag = method.delivery_tag)
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-class alarm_state_senderThread(senderThread):
- def __init__(self,
- MQparameters,
- input_msg_queue,
- sender_id,
- ):
- super(alarm_state_senderThread,self).__init__(
- MQparameters, \
- input_msg_queue,
- sender_id,
- "alarm_state",
- "fanout"
- )
- # 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-class alarm_state_receiverThread(receiverThread):
- def __init__(self,
- MQparameters,
- receiver_id):
- super(alarm_state_receiverThread,self).__init__(
- MQparameters,
- None,
- None,
- "alarm_state",
- "fanout",
- receiver_id
- )
-
- def callback(self, ch, method, properties, body):
-
- response_tag = f"{time.asctime()} received: {body}."
- print(response_tag)
- ch.basic_ack(delivery_tag = method.delivery_tag)
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-class command_request_senderThread(senderThread):
- def __init__(self,
- MQparameters,
- input_msg_queue,
- sender_id,
- ):
- super(command_request_senderThread,self).__init__(
- MQparameters, \
- input_msg_queue,
- sender_id,
- "command_request",
- "topic"
- )
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-class command_request_receiverThread(receiverThread):
- def __init__(self,
- MQparameters,
- received_msg_queue,
- response_msg_queue,
- receiver_id):
- super(command_request_receiverThread,self).__init__(
- MQparameters,
- received_msg_queue,
- response_msg_queue,
- "command_request",
- "topic",
- receiver_id
- )
-
- def callback(self, ch, method, properties, body):
- received_msg = Message(None,None,body,None)
- self.queue.put(received_msg)
-
- response_tag = f"{time.asctime()},{self.receiver_id} received {self.exchangeName} message."
- print(response_tag)
- self.response_queue.put(Message(self.receiver_id,self.receiver_id,response_tag,''))
- ch.basic_ack(delivery_tag = method.delivery_tag)
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-class command_response_senderThread(senderThread):
- def __init__(self,
- MQparameters,
- input_msg_queue,
- sender_id,
- ):
- super(command_response_senderThread,self).__init__(
- MQparameters, \
- input_msg_queue,
- sender_id,
- "command_response",
- "fanout"
- )
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-class command_response_receiverThread(receiverThread):
- def __init__(self,
- MQparameters,
- receiver_id):
- super(command_response_receiverThread,self).__init__(
- MQparameters,
- None,
- None,
- "command_response",
- "fanout",
- receiver_id
- )
-
- def callback(self, ch, method, properties, body):
- response_tag = f"{time.asctime()} received: {body}."
- print(response_tag)
- ch.basic_ack(delivery_tag = method.delivery_tag)
-
-
-# TODO
-class image_data_senderThread(senderThread):
- def __init__(self,
- MQparameters,
- input_msg_queue,
- sender_id,
- ):
- super(command_response_senderThread,self).__init__(
- MQparameters, \
- input_msg_queue,
- sender_id,
- "image_data",
- "fanout"
- )
-
-class image_data_receiverThread(receiverThread):
- 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()
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-# class device_state_up_msg_sender_thread(threading.Thread):
-# def __init__(self, MQparameters,device_state_msg_queue):
-# super().__init__(self)
-# self.parameters = MQparameters
-# self.queue = device_state_msg_queue
-# self.connection = pika.BlockingConnection(self.parameters)
-
-# def stop(self):
-# """stop the thread"""
-# self.is_stop = True
-# self.connection.close()
-
-# def run(self):
-
-# while not self.is_stop:
-# print("device_state_up_msg_sender_thread started")
-# channel = self.connection.channel()
-# channel.exchange_declare(exchange='device_state', exchange_type='fanout')
-
-# while True:
-# try:
-# if not self.queue.empty():
-# msg = self.msg_queue.get()
-# channel.basic_publish(exchange='device_state', routing_key='', body=msg)
-# print(time.asctime(), "send device state message")
-# elif self.queue.empty():
-# time.sleep(1)
-# except Exception as e:
-# break
-
-# class device_state_up_msg_receiver_thread(threading.Thread):
-# def __init__(self, MQparameters):
-# super().__init__(self)
-# self.parameters = MQparameters
-# self.connection = pika.BlockingConnection(self.parameters)
-
-# def stop(self):
-# """stop the thread"""
-# self.is_stop = True
-# self.connection.close()
-
-# def run(self):
-# while not self.is_stop:
-# print("device_state_up_msg_receiver_thread started")
-
-# channel = self.connection.channel()
-# channel.exchange_declare(exchange='device_state', exchange_type='fanout')
-
-# result = channel.queue_declare(queue='', exclusive=True, durable=True)
-# queue_name = result.method.queue
-
-# channel.queue_bind(exchange='device_state', queue=queue_name)
-
-# def callback(ch, method, properties, body):
-# print(f"Received: {body}")
-# ch.basic_ack(delivery_tag = method.delivery_tag)
-
-# channel.basic_consume(queue=queue_name, \
-# on_message_callback=callback)
-
-# channel.basic_qos(prefetch_count=1)
-# channel.start_consuming()
-
-# class alarm_state_up_msg_sender_thread(threading.Thread):
-
-# def __init__(self, MQparameters,alarm_state_msg_queue):
-# super().__init__(self)
-# self.parameters = MQparameters
-# self.queue = alarm_state_msg_queue
-# self.connection = pika.BlockingConnection(self.parameters)
-
-# def stop(self):
-# """stop the thread"""
-# self.is_stop = True
-# self.connection.close()
-
-# def run(self):
-
-# while not self.is_stop:
-# print("alarm_state_up_msg_sender_thread started")
-# channel = self.connection.channel()
-# channel.exchange_declare(exchange='alarm_state', exchange_type='fanout')
-
-# while True:
-# try:
-# if not self.queue.empty():
-# msg = self.msg_queue.get()
-# channel.basic_publish(exchange='alarm_state', routing_key='', body=msg)
-# print(time.asctime(), "send alarm state message")
-# elif self.queue.empty():
-# time.sleep(1)
-# except Exception as e:
-# break
-
-# class alarm_state_up_msg_receiver_thread(threading.Thread):
-# def __init__(self, MQparameters):
-# super().__init__(self)
-# self.parameters = MQparameters
-# self.connection = pika.BlockingConnection(self.parameters)
-
-# def stop(self):
-# """stop the thread"""
-# self.is_stop = True
-# self.connection.close()
-
-# def run(self):
-
-# while not self.is_stop:
-# print("alarm_state_up_msg_receiver_thread started")
-# channel = self.connection.channel()
-# channel.exchange_declare(exchange='alarm_state', exchange_type='fanout')
-
-# result = channel.queue_declare(queue='', exclusive=True, durable=True)
-# queue_name = result.method.queue
-
-# channel.queue_bind(exchange='alarm_state', queue=queue_name)
-
-# def callback(ch, method, properties, body):
-# print(f"Received: {body}")
-# ch.basic_ack(delivery_tag = method.delivery_tag)
-
-# channel.basic_consume(queue=queue_name, \
-# on_message_callback=callback)
-
-# channel.basic_qos(prefetch_count=1)
-# channel.start_consuming()
-
-# class command_request_msg_sender_thread(threading.Thread):
-
-# def __init__(self, MQparameters,command_request_msg_queue,routing_key_queue):
-# super().__init__(self)
-# self.parameters = MQparameters
-# self.command_queue = command_request_msg_queue
-# self.key_queue = routing_key_queue
-# self.connection = pika.BlockingConnection(self.parameters)
-
-# def stop(self):
-# """stop the thread"""
-# self.is_stop = True
-# self.connection.close()
-
-# def run(self):
-
-# while not self.is_stop:
-# print("command_request_msg_sender_thread started")
-# channel = self.connection.channel()
-# channel.exchange_declare(exchange='command_msg', exchange_type='topic')
-
-# while True:
-# try:
-# if not self.command_queue.empty() and \
-# not self.key_queue.empty():
-# msg = self.command_queue.get()
-# routing_key = self.key_queue.get()
-# corr_id = str(uuid.uuid4())
-# channel.basic_publish(exchange='command_msg',
-# routing_key = routing_key,
-# body=msg,
-# properties=pika.BasicProperties(
-# correlation_id=corr_id
-# )
-# )
-# print(time.asctime(), f"send command request with ID:[{corr_id}]")
-# elif self.queue.empty():
-# time.sleep(1)
-# except Exception as e:
-# break
-
-# class command_request_msg_receiver_thread(threading.Thread):
-# def __init__(self, MQparameters,command_request_msg_queue, \
-# command_response_msg_queue, routing_key):
-# super().__init__(self)
-# self.parameters = MQparameters
-# self.queue = command_request_msg_queue
-# self.response_queue = command_response_msg_queue
-# self.routing_key = routing_key
-# self.connection = pika.BlockingConnection(self.parameters)
-
-# def stop(self):
-# """stop the thread"""
-# self.is_stop = True
-# self.connection.close()
-
-# def run(self):
-
-# while not self.is_stop:
-# print("command_request_msg_receiver_thread started")
-# channel = self.connection.channel()
-# channel.exchange_declare(exchange='command_msg', exchange_type='topic')
-
-# result = channel.queue_declare(queue='', exclusive=True, durable=True)
-# queue_name = result.method.queue
-
-# channel.queue_bind(exchange='command_msg', queue=queue_name, \
-# routing_key=self.routing_key)
-
-# def callback(ch, method, prop, body):
-# print(time.asctime(), f"reveive command request with ID:[{prop.corr_id}]")
-# self.queue.put(body)
-# self.response_queue.put("")
-# ch.basic_ack(delivery_tag = method.delivery_tag)
-
-# channel.basic_consume(queue=queue_name, \
-# on_message_callback=callback)
-
-# channel.basic_qos(prefetch_count=1)
-# channel.start_consuming()
-
-# class command_response_msg_sender_thread(threading.Thread):
-# def __init__(self, MQparameters,command_request_msg_queue):
-# super().__init__(self)
-# self.parameters = MQparameters
-# self.queue = command_request_msg_queue
-# self.connection = pika.BlockingConnection(self.parameters)
-
-# def stop(self):
-# """stop the thread"""
-# self.is_stop = True
-# self.connection.close()
-
-# def run(self):
-
-# while not self.is_stop:
-# print("command_response_msg_sender_thread started")
-# channel = self.connection.channel()
-# channel.exchange_declare(exchange='command_response', exchange_type='fanout')
-
-# while True:
-# try:
-# if not self.queue.empty():
-# msg = self.msg_queue.get()
-# channel.basic_publish(exchange='command_response', routing_key='', body=msg)
-# print(time.asctime(), "send device state message")
-# elif self.queue.empty():
-# time.sleep(1)
-# except Exception as e:
-# break
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-# class command_down_rpc_sender_thread(threading.Thread):
-# def __init__(self, MQparameters, command_queue, routing_key_queue):
-# super().__init__(self)
-# self.parameters = MQparameters
-# self.command_queue = command_queue
-# self.key_queue = routing_key_queue
-# self.connection = pika.BlockingConnection(self.parameters)
-
-# def stop(self):
-# """stop the thread"""
-# self.is_stop = True
-# self.connection.close()
-
-# def run(self):
-
-# while not self.is_stop:
-# # print("command_down_rpc_sender_thread started")
-
-# channel = self.connection.channel()
-
-# # declare a exchange
-# channel.exchange_declare(exchange="command_rpc", \
-# exchange_type='topic')
-
-# # declare a callback queue
-# result = channel.queue_declare(queue='', exclusive=True)
-# callback_queue = result.method.queue
-
-# # bingding the queue
-# channel.queue_bind(
-# exchange="command_rpc",
-# queue=callback_queue,
-# routing_key=routing_key
-# )
-
-# # init the command uuid
-# corr_id = str(uuid.uuid4())
-# command_response = None
-
-# # def a function to print the command respond message
-# def respond_callback(ch, method, properties, body):
-# if properties.correlation_id == corr_id:
-# command_response = body
-# print(body)
-
-# # start the cosumer
-# channel.basic_consume(
-# queue=callback_queue,
-# on_message_callback=respond_callback,
-# auto_ack=True
-# )
-
-# # publish with tropic exchange
-# channel.basic_publish(
-# exchange="command_rpc",
-# routing_key=routing_key,
-# body = request_command,
-# properties=pika.BasicProperties(
-# reply_to=callback_queue,
-# correlation_id=corr_id
-# ),
-# )
-# with command_response is None:
-# self.connection.process_data_events()
-
-# class command_down_rpc_receiver_thread(threading.Thread):
-# def __init__(self, MQparameters):
-# super().__init__(self)
-# self.parameters = MQparameters
-# self.connection = pika.BlockingConnection(self.parameters)
-
-# def stop(self):
-# """stop the thread"""
-# self.is_stop = True
-# self.connection.close()
-
-# def run(self,commands_queue):
-# while not self.is_stop:
-# print("command_down_rpc_receiver_thread started")
-# channel = self.connection.channel()
-
-# # declare a exchange
-# channel.exchange_declare(exchange="command_rpc", \
-# exchange_type='topic')
-
-# # declare a callback queue
-# result = channel.queue_declare(queue='', exclusive=True)
-# callback_queue = result.method.queue
-
-# # bingding the queue
-# channel.queue_bind(
-# exchange="command_rpc",
-# queue=callback_queue,
-# routing_key='pod*'
-# )
-
-# # def a function to reply the request from the sender
-# def on_request(ch, method, prop, body):
-# commands_queue.put(body)
-
-# response = f"{time.asctime()} {self.device_id} receive the command uuid:[{prop.correlation_id}]"
-# print(response)
-
-# ch.basic_publish(exchange="command_rpc",
-# routing_key = prop.reply_to,
-# properties = pika.BasicProperties(
-# correlation_id=prop.corr_id
-# ),
-# body = response)
-# ch.basic_ack(delivery_tag = method.delivery_tag)
-
-# # start the cosumer
-# channel.basic_qos(prefetch_count=1)
-# channel.basic_consume(
-# queue=callback_queue,
-# on_message_callback=on_request,
-# auto_ack=True
-# )
-# channel.start_consuming()
-
-# class image_data_up_msg_sender_thread(threading.Thread):
-# pass
-
-# class image_data_up_msg_receiver_thread(threading.Thread):
-# pass
-
-# class ply_data_up_msg_sender_thread(threading.Thread):
-# pass
-
-# class ply_data_up_msg_receiver_thread(threading.Thread):
-# pass
\ No newline at end of file
--
Gitblit v1.9.1