add taskuuid

wu_xinjun

2022-06-24 74c96c72625aff6613d69c7168e1a7d76da22870

 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)