From 74c96c72625aff6613d69c7168e1a7d76da22870 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期五, 24 六月 2022 17:18:31 +0800
Subject: [PATCH] add taskuuid
---
src/engine/MessageThreads.py | 147 +++++++++++++++++++++++++++++--------------------
1 files changed, 87 insertions(+), 60 deletions(-)
diff --git a/src/engine/MessageThreads.py b/src/engine/MessageThreads.py
index 998a3e2..8cc2344 100644
--- a/src/engine/MessageThreads.py
+++ b/src/engine/MessageThreads.py
@@ -67,20 +67,20 @@
message = self.message_queue.get()
msg_routing_key = message[0]
msg_body = message[1]
+ if msg_body is not False:
+ # 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
+ )
+ )
- # 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)
+ 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)
@@ -143,20 +143,21 @@
if self.is_stop:
ch.close()
- self.message_queue.put(body)
+ if body is not False:
+ 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):
+ task_folder:dict[str,str]):
super().__init__()
self.queue_list = queue_list
self.queue_tags = queue_tags
self.queue_num = 0
- self.save_path = save_path
+ self.task_folder = task_folder
self.is_stop = False
assert len(self.queue_list) == len(self.queue_tags)
@@ -167,22 +168,22 @@
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)
+ # # 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.task_folder,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")
+ # 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]
+ # 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}")
+ # 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)
@@ -221,16 +222,19 @@
msg_pb = pb.LogsMessage()
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)
-
# process the msg
msg_source = msg_pb.source
msg_timestamp = msg_pb.timestamp
msg_context = msg_pb.context
+ taskuuid = msg_pb.taskuuid
+
+ save_folder = self.task_folder[taskuuid]
+
+ # save the raw pb msg
+ # curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ raw_file_name = os.path.join(save_folder,f"{msg_timestamp}-{msg_source}.LogsMsg")
+ with open(raw_file_name, 'wb') as file:
+ file.write(msg)
# display the info to the console
curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
@@ -248,15 +252,18 @@
msg_pb = pb.ImagesMessage()
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.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
+ taskuuid = msg_pb.taskuuid
+ save_folder = self.task_folder[taskuuid]
+
+ # save the raw pb msg
+ # curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ raw_file_name = os.path.join(save_folder,f"{msg_timestamp}-{msg_source}.ImagesMsg")
+ with open(raw_file_name, 'wb') as file:
+ file.write(msg)
+
# process the images
id = 0
@@ -268,8 +275,8 @@
if (image_byte is None) | (len(image_byte) != imagelegth ):
pass
else:
- filename = f"{msg_source}_{msg_timestamp}_image{id}.jpg"
- filepath = os.path.join(self.imagesmsg_folder,filename)
+ filename = f"{msg_timestamp}-{msg_source}-image{id}.jpg"
+ filepath = os.path.join(save_folder,filename)
with open(filepath, 'wb') as f:
f.write(image_byte)
@@ -289,15 +296,18 @@
msg_pb = pb.PlysMessage()
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.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
+ taskuuid = msg_pb.taskuuid
+
+ save_folder = self.task_folder[taskuuid]
+
+ # save the raw pb msg
+ # curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ raw_file_name = os.path.join(save_folder,f"{msg_timestamp}-{msg_source}.PlysMsg")
+ with open(raw_file_name, 'wb') as file:
+ file.write(msg)
# process the muti ply data
id = 0
@@ -317,13 +327,14 @@
data_array = np.frombuffer(data_byte,dtype=dt)
data_array = data_array.reshape(rows,cols,order=order)
# save the data with .ply format
- filename = f"{msg_source}_{msg_timestamp}_pointcloud{id}"
- write_ply(data_array,self.plysmsg_folder,filename)
+ filename = f"{msg_timestamp}-{msg_source}-pointcloud{id}"
+ write_ply(data_array,save_folder,filename)
# next ply item
id += 1
def decode_CommandsMsg(self,queue):
+
raise NotImplementedError("NOT IMPLEMENT!")
@@ -332,15 +343,18 @@
def send_logs_tool(logs:List[str],
logs_queue:queue.Queue,
source:str,
- routing_key:str):
+ routing_key:str,
+ taskuuid:str = "init"):
for i, log in enumerate(logs):
if log is not None:
# constract a logs message
logsMsg = pb.LogsMessage()
logsMsg.source = source
- logsMsg.timestamp = f"{time.asctime()}"
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ logsMsg.timestamp = curr_time
logsMsg.context = log
+ logsMsg.taskuuid = taskuuid
# convert the proto object to string
logsMsg_bytes = logsMsg.SerializeToString()
# add logsmsg to queue
@@ -350,13 +364,15 @@
def send_images_tool(images:List[np.array],
images_queue:queue.Queue,
source:str,
- routing_key:str):
+ routing_key:str,
+ taskuuid:str = "init"):
# constract a images message
imagesMsg = pb.ImagesMessage()
imagesMsg.source = source
curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
imagesMsg.timestamp = curr_time
+ imagesMsg.taskuuid = taskuuid
for img in images:
if img is not None:
@@ -377,13 +393,17 @@
def send_pointcloud_tool(pointclouds:queue.Queue,
plys_queue:queue.Queue,
source:str,
- routing_key:str):
+ routing_key:str,
+ taskuuid:str = "init"):
# constract a plys message
plysMsg = pb.PlysMessage()
plysMsg.source = source
curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
plysMsg.timestamp = curr_time
+ plysMsg.taskuuid = taskuuid
+
+ empty = True
run =True
@@ -393,6 +413,7 @@
value = pointclouds.get()
if value is not False:
+ empty = False
pointcloud = np.array(value)
# get the attributes of the PointCloudItem
rows = pointcloud.shape[0]
@@ -412,13 +433,19 @@
item.data = data
else:
- # converting the pb ply msg to bytes
- plysMsg_bytes = plysMsg.SerializeToString()
- # add the plysMsg to the queue
- plys_queue.put([routing_key,plysMsg_bytes])
- # end the outer while loop and triminate this function
+ if not empty:
+ # converting the pb ply msg to bytes
+ plysMsg_bytes = plysMsg.SerializeToString()
+ # add the plysMsg to the queue
+ plys_queue.put([routing_key,plysMsg_bytes])
+ # end the outer while loop and triminate this function
+
+ else:
+ print("[x] NO poindcloud has been captured!")
run = False
+ else:
+ time.sleep(3)
--
Gitblit v1.9.1