server enhanced!

wu_xinjun

2022-06-14 d8e5c7c93ed28afa2af1d68ea20cff962599626f

 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