server enhanced!

wu_xinjun

2022-06-14 d8e5c7c93ed28afa2af1d68ea20cff962599626f

server enhanced!

 scripts/mqServer.py

@@ -8,26 +8,31 @@

from src.Server import MyServer


if __name__ == "__main__":
    def help():
        print("Usage: Server.py [MQhost] [MQuser] [MQpassword] [QUEUEname] [WEBport] [SAVEfolder]")
        print("EXAMPLE: Server.py localhost chai password123 aerial_rpc 5000 ./Images")
        exit(1)

    if len(sys.argv) != 7:
    if len(sys.argv) != 8:
        help()
    try:
        MQhost = str(sys.argv[1])
        MQuser = str(sys.argv[2])
        MQpassword = str(sys.argv[3])
        QUEUEname = str(sys.argv[4])
        WEBhost = str(sys.argv[4])
        WEBport = int(sys.argv[5])
        SAVEfolder = str(sys.argv[6])
        save_folder = SAVEfolder
        save_folder = str(sys.argv[6])
        server_name = str(sys.argv[7])

    except Exception as e:
        print(e)
        help()
    server = MyServer(host=MQhost, user=MQuser, password=MQpassword, queue=QUEUEname, port=WEBport, SAVEfolder=save_folder)
    server = MyServer(MQhost=MQhost,
                    MQuser=MQuser,
                    MQpassword=MQpassword,
                    WEBhost=WEBhost,
                    WEBport=WEBport,
                    save_folder=save_folder,
                    server_name=server_name)
    server.run_all()

 scripts_launch/TEST_startServer.bat

@@ -16,20 +16,21 @@

echo change to path: %cd%

@REM set MQhost=localhost
set MQhost=47.92.33.19
set MQhost=localhost
@REM set MQhost=47.92.33.19
set MQuser=chai
set MQpassword=password123
set WEBhost=0.0.0.0
set WEBport=5000
set SAVEfolder=./Received_data_testing
set SAVEfolder=./data/test/
set SERVERname=testserver
set LOGfile=ServerTEST.log

@REM Launch the Server
echo the log file will be saved in the [%LOGfile%] file
echo Server.py is running ...
@REM -------- Usage: Server.py [MQhost] [MQuser] [MQpassword] [QUEUEname] [WEBport] [SAVEfolder]")
%aerialpython% ./scripts/mqServer.py %MQhost% %MQuser% %MQpassword% %QUEUEname% %WEBport% %SAVEfolder%
%aerialpython% ./scripts/mqServer.py %MQhost% %MQuser% %MQpassword% %WEBhost% %WEBport% %SAVEfolder% %SERVERname%
echo Server.py has been terminited.
pause


 src/Server.py

@@ -1,175 +1,142 @@
import sys              # NOQA: E402    
import os
from flask import Flask
from numpy import true_divide
import pika
import uuid
import queue
from werkzeug.wrappers import request
from threading import Thread
from datetime import datetime

from zmq import device


from .engine.DeviceManager import DeviceManager
from .engine.MessageQueueThread import MessageQueueThread
from .engine import MessageThreads as ths
from .protos import aerial_pb2 as pb




class MyServer(Flask):
    def __init__(self, host, user, password, queue, port, SAVEfolder):
    def __init__(self,
                MQhost:str,
                MQuser:str,
                MQpassword:str,
                WEBhost:str,
                WEBport:int,
                save_folder:str,
                server_name:str
                ):
        Flask.__init__(self, __name__)
        self.host = host
        self.user = user
        self.password = password
        self.queue = queue
        self.port = port
        global save_folder
        save_folder = SAVEfolder
        self.MQhost = MQhost
        self.MQuser = MQuser
        self.MQpassword = MQpassword
        self.WEBhost = WEBhost
        self.WEBport = WEBport
        self.save_folder = save_folder
        self.server_name = "server0"
        

        self.route('/api', methods=['POST'])(self.api)
        # self.route('/api', methods=['POST'])(self.api)
        self.route('/', methods=['GET'])(self.hello)
        self._def_routes()
        # threads
        self._device_state_reveiver_thread = None
        self._alarm_state_receiver_thread = None
        self._command_request_sender_thread = None
        self._command_response_receiver_thread = None
        # self._def_routes()

        # queue
        self.command_request_queue = queue.Queue()
        self.LogsReceiver_queue = queue.Queue()
        self.ImagesReceiver_queue = queue.Queue()
        self.PlysReceiver_queue = queue.Queue()

        self.CommandsSender_queue = queue.Queue()

    def _connect_to_MQ(self):
        credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
        self.parameters = pika.ConnectionParameters(
            credentials=credentials, host=self.MQhost)
        # 连接队列服务器
        self.connection = pika.BlockingConnection(self.parameters)
        connection = pika.BlockingConnection(self.parameters)
        print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
        connection.close()

    def _start_device_state_up_msg_receiver_thread(self):
        """start device state thread"""
        if self._device_state_reveiver_thread is None:
            self._device_state_reveiver_thread = \
                ths.device_state_receiverThread(self.parameters,"server")
            self._device_state_reveiver_thread.setDaemon(True)
            self._device_state_reveiver_thread.start()

    def _stop_device_state_up_msg_receiver_thread(self):
        """stop device state thread"""
        if self._device_state_reveiver_thread is not None:
            self._device_state_reveiver_thread.stop()
            self._device_state_reveiver_thread = None
    def _receiver_Thread_Launcher(self):

    def _start_alarm_state_up_msg_receiver_thread(self):
        """start alarm state thread"""
        if self._alarm_state_receiver_thread is None:
            self._alarm_state_receiver_thread = \
                ths.alarm_state_receiverThread(self.parameters,"server")
            self._alarm_state_receiver_thread.setDaemon(True)
            self._alarm_state_receiver_thread.start()
        # launch logs receiver thread
        self.LogsReceiver_Thread = ths.receiverThread(
                                    MQparams=self.parameters,
                                    exchange_name='logs_topic',
                                    exchange_type='topic',
                                    MQqueue_name=f'logs_queue',
                                    binding_keys='server.logs.test0',
                                    message_queue=self.LogsReceiver_queue
                                    )
        self.LogsReceiver_Thread.setDaemon(True)
        self.LogsReceiver_Thread.setName("logs_thread")
        self.LogsReceiver_Thread.start()

    def _stop_alarm_state_up_msg_receiver_thread(self):
        """stop alarm state thread"""
        if self._alarm_state_receiver_thread is not None:
            self._alarm_state_receiver_thread.stop()
            self._alarm_state_receiver_thread = None
        # launch images receiver thread
        self.ImagesReceiver_Thread = ths.receiverThread(
                                    MQparams=self.parameters,
                                    exchange_name='images_topic',
                                    exchange_type='topic',
                                    MQqueue_name=f'images_queue',
                                    binding_keys="server.images.test0",
                                    message_queue=self.ImagesReceiver_queue
                                    )
        self.ImagesReceiver_Thread.setDaemon(True)
        self.ImagesReceiver_Thread.setName("images_thread")
        self.ImagesReceiver_Thread.start()

    def _start_command_response_receiver_thread(self):
        """start command response thread"""
        if self._command_response_receiver_thread is None:
            self._command_response_receiver_thread = \
                ths.command_response_receiverThread(self.parameters,"server")
            self._command_response_receiver_thread.setDaemon(True)
            self._command_response_receiver_thread.start()
        # launch plys receiver thread
        self.PlysReceiver_Thread = ths.receiverThread(
                                    MQparams=self.parameters,
                                    exchange_name='plys_topic',
                                    exchange_type='topic',
                                    MQqueue_name=f'plys_queue',
                                    binding_keys='server.plys.test0',
                                    message_queue=self.PlysReceiver_queue
                                    )
        self.PlysReceiver_Thread.setDaemon(True)
        self.PlysReceiver_Thread.setName("plys_thread")
        self.PlysReceiver_Thread.start()

    def _stop_command_response_receiver_thread(self):
        """stop command response thread"""
        if self._command_response_receiver_thread is not None:
            self._command_response_receiver_thread.stop()
            self._command_response_receiver_thread = None
        # launch decode thread
        queue_list = [self.LogsReceiver_queue,self.ImagesReceiver_queue,self.PlysReceiver_queue]
        queue_tags = ["Logs", "Images", "Plys"]
        self.MsgDecoder_Thread = ths.decodeThread(
                                    queue_list= queue_list,
                                    queue_tags= queue_tags,
                                    save_path= self.save_folder           
                                    )

    def _start_command_request_sender_thread(self):
        """start command request thread"""
        if self._command_request_sender_thread is None:
            self._command_request_sender_thread = \
                ths.command_request_senderThread(self.parameters,self.command_request_queue,"server")
            self._command_request_sender_thread.setDaemon(True)
            self._command_request_sender_thread.start()
        self.MsgDecoder_Thread.setDaemon(True)
        self.MsgDecoder_Thread.setName("decode_thread")
        self.MsgDecoder_Thread.start()

    def _stop_command_request_sender_thread(self):
        """stop command request thread"""
        if self._command_request_sender_thread is not None:
            self._command_request_sender_thread.stop()
            self._command_request_sender_thread = None

    def _def_routes(self):
        self.routes = {
            pb.SysCommand.SYS_COMMAND: self._on_sys_command,
        }
    
    def _on_sys_command(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
    def _sender_Thread_Launcher(self):

        self.command_request_queue.put(ths.Message(None,None,req.command,req.target))
        return
        # launcher command sender thread
        self.CommandsSender_Thread = ths.senderThread(
                                    MQparams=self.parameters,
                                    exchange_name='commands_topic',
                                    exchange_type='topic',
                                    message_queue=self.CommandsSender_queue
                        )
        self.CommandsSender_Thread.setDaemon(True)
        self.CommandsSender_Thread.setName("commands_thread")
        self.CommandsSender_Thread.start()

    def _default(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
        print('onknown command')

    def api(self):

        req = pb.RequestCommand()
        req.ParseFromString(request.data)

        # a function: _on_sys_command, refer to self._on_sys_command
        dispatcher = self.routes.get(req.sys_command, self._default)

        resp = pb.ResponseCommand()
        dispatcher(req, resp) # call the function

        resp.checksum = 12345
        response = Flask.response_class(
            resp.SerializeToString(),
            mimetype='application/x-protobuf'
        )
        return response, 200

    def hello(self):
        return 'Hello, This is acquisition demo API!'

    def run_rabbitmq(self):
        self._connect_to_MQ()
        self._start_device_state_up_msg_receiver_thread()
        self._start_alarm_state_up_msg_receiver_thread()
        self._start_command_response_receiver_thread()
        self._start_command_request_sender_thread()
        self._receiver_Thread_Launcher()
        self._sender_Thread_Launcher()


    def run_all(self):
        self.run_rabbitmq()
        self.run(debug=False, host=self.WEBhost, port=self.WEBport)

<<<<<<< HEAD

if __name__ == "__main__":
    def help():
        print("Usage: Server.py [MQhost] [MQuser] [MQpassword] [WEBhost] [WEBport] [SAVEfolder]")
        print("EXAMPLE: Server.py localhost chai password123 0.0.0.0 5000 ./Received_data/")
        exit(1)

    if len(sys.argv) != 7:
        help()
    try:
        MQhost = str(sys.argv[1])
        MQuser = str(sys.argv[2])
        MQpassword = str(sys.argv[3])
        WEBhost = str(sys.argv[4])
        WEBport = int(sys.argv[5])
        SAVEfolder = str(sys.argv[6])

    except Exception as e:
        print(e)
        help()
    server = MyServer(MQhost, MQuser, MQpassword, WEBhost, WEBport, SAVEfolder)
    server.run_all()
=======
>>>>>>> aa8a0ba3a86ff5dbbdfcd68c72a5e70e1c52e877

 src/Server1.py

File was deleted

 src/__init__.py

@@ -1,5 +1,5 @@
from .engine import *
from .microwave import *
from .protos import *
from .slam import *
from .utils import *
# from .engine import *
# from .microwave import *
# from .protos import *
# from .slam import *
# from .utils import *

 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

 src/engine/__init__.py

@@ -1,10 +1,10 @@
from .AcquisitionMain import *
from .CaptureThread import *
from .ControllThread import *
from .DeviceManager import *
from .DustCleaner import *
from .DustCleanerThread import *
from .MessageQueueThread import *
from .MicroWaveUDPThread import *
from .MessageThreads import *
from .devices import *
# from .AcquisitionMain import *
# from .CaptureThread import *
# from .ControllThread import *
# from .DeviceManager import *
# from .DustCleaner import *
# from .DustCleanerThread import *
# from .MessageQueueThread import *
# from .MicroWaveUDPThread import *
# from .MessageThreads import *
# from .devices import *

 src/microwave/__init__.py

@@ -1,5 +1,5 @@
from .components import *
from  .constants import *
from  .parsers import *
from  .servers import *
from .deviceTools import *
# from .components import *
# from  .constants import *
# from  .parsers import *
# from  .servers import *
# from .deviceTools import *

 src/protos/aerial.proto

@@ -138,17 +138,50 @@
}


enum DataType
{
  c_int16 = 1;
  c_int32 = 2;
  c_int64 = 3;
  c_float16 = 4;
  c_float32 = 5;
  c_float = 6;
}


message PlyMessage
{
  string source = 1;
  string timestamp = 2;
  bytes plydata = 3;

  message PointCloudItem
  {
    int32 length = 1;
    int32 rows = 2;
    int32 cols = 3;
    string order = 4;
    DataType d_types = 5;
    bytes data = 6
    
  }

  repeated PointCloudItem Plydata = 3;

  // enum ref_system
  // {
  //   MICROWAVE_SYSTEM = 0;
  //   DEVICE_SYSTEM = 1;
  // }

  // ref_system reference = 4
}







// message MicroWaveMessage
// {
//   int32 length = 1;                           // 数据长度

 src/slam/__init__.py

@@ -1,3 +1,3 @@
from .core import *
from .manualMode import *
from .autoMode import *
# from .core import *
# from .manualMode import *
# from .autoMode import *

 src/utils/__init__.py

@@ -1 +1 @@
from .tools import *
# from .tools import *