-_-

wu_xinjun

2022-06-10 baee79553738a212d82a12a1966fe57ce4acb937

 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()