temp commit

wu_xinjun

2022-03-04 e839f2e83fed7eaceb677242f5f9650a79fb8c43

temp commit

 .vscode/launch.json

@@ -48,9 +48,9 @@
                "localhost",
                "chai",
                "password123",
                "aerial_rpc",
                "0.0.0.0",
                "5000",
                "./Images"
                "./Received_data_test"
            ],
            "console": "integratedTerminal"
        },

 TEST_startClient.bat

@@ -12,7 +12,7 @@
    call %envsetting% 
)

set MQhost=192.168.1.224
set MQhost=192.168.1.14
set MQuser=chai
set MQpassword=password123
set QUEUEname=aerial_rpc

 TEST_startServer.bat

@@ -17,17 +17,16 @@
set MQhost=localhost
set MQuser=chai
set MQpassword=password123
set QUEUEname=aerial_rpc
set WEBhost=0.0.0.0
set WEBport=5000
set SAVEfolder=./ImagesTEST
set SAVEfolder=./Received_data_testing
set LOGfile=ServerTEST.log

@REM Launch the Server
@REM -------- Usage: Server.py [MQhost] [MQuser] [MQpassword] [QUEUEname] [WEBport] [SAVEfolder]")
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% ./src/Server.py %MQhost% %MQuser% %MQpassword% %QUEUEname% %WEBport% %SAVEfolder% >> %LOGfile%
@REM -------- Usage: Server.py [MQhost] [MQuser] [MQpassword] [WEBhost] [WEBport] [SAVEfolder]")
%aerialpython% ./src/Server.py %MQhost% %MQuser% %MQpassword% %WEBhost% %WEBport% %SAVEfolder%
echo Server.py has been terminited.
pause


 src/Client.py

@@ -44,9 +44,7 @@
    client = AcquisitionMain(MQhost,
                             MQuser,
                             MQpassword, 
                             QUEUEname, 
                             DEVICEname, 
                             SERVERhost, 
                             DEVICEname,
                             MICROWAVEhost, # udp server host, 192.168.2.65 for production
                             config)
    )
    client.start()

 src/Server.py

@@ -1,56 +1,135 @@
from posixpath import abspath
import sys              # NOQA: E402    
import os
from flask import Flask, request
from flask import Flask
from numpy import true_divide
import pika
import uuid
import queue
from werkzeug.wrappers import request

from zmq import device


batfile_dir = os.path.join(__file__, "..", "protos/compile.bat" )
batfile_path = os.path.abspath(batfile_dir)
os.system(batfile_path)

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



class MyServer(Flask):
    def __init__(self, host, user, password, queue, port):
    def __init__(self, MQhost, MQuser, MQpassword, WEBhost, WEBport, SAVEfolder):
        Flask.__init__(self, __name__)
        self.host = host
        self.user = user
        self.password = password
        self.queue = queue
        self.port = port

        # rabbit MQ
        self.MQhost = MQhost
        self.MQuser = MQuser
        self.MQpassword = MQpassword
        self.parameters = None
        # Web server
        self.WEBhost = WEBhost
        self.WEBport = WEBport
        self.SAVEfolder = SAVEfolder
        self.route('/api', methods=['POST'])(self.api)
        self.route('/', methods=['GET'])(self.hello)
        self._def_routes()
        self._device_manager = DeviceManager(host, user, password)
        # 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
        # queue
        self.command_request_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)
        print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")

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

    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

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

    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

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

    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):
        if pb.SubCommand.POWER_ON_SUB_COMMAND == req.sub_command:
            print('power on message received')
            self._device_manager.get_device(req.device_id,save_folder)
        elif pb.SubCommand.POWER_OFF_SUB_COMMAND == req.sub_command:
            device = self._device_manager.get_device(req.device_id,save_folder)
            device['power_on'] = False
            print('power off message received')

        self.command_request_queue.put(ths.Message(None,None,req.command,req.target))
        return

    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)
        dispatcher(req, resp) # call the function

        resp.checksum = 12345
        response = Flask.response_class(
            resp.SerializeToString(),
@@ -61,39 +140,22 @@
    def hello(self):
        return 'Hello, This is acquisition demo API!'

    def run_rabbitmq(self, host: str, user: str, password: str, queue: str) -> None:
        self._message_queue_thread = MessageQueueThread(
            host, user, password, queue, self._device_manager, save_folder)
        self._message_queue_thread.start()
    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()

    def run_all(self):
        self.run_rabbitmq(self.host, self.user, self.password, self.queue)
        self.run(debug=False, host="0.0.0.0", port=self.port)

    def send_controll_message(self, device_id, msg):
        device = self._device_manager.get_device(device_id,save_folder)
        if device is None:
            channel = device.controll_channel
            channel.basic_publish(
                exchange='',
                routing_key='',
                body=msg.SerializeToString())

    def send_dust_cleaner_cmd(self, device_id, sub_cmd, period):
        msg = pb.ControllMessage()
        msg.sys_command = pb.ControlSysCommand.DUST_CLEANER_COMMAND
        msg.sub_command = sub_cmd
        data_msg = pb.DustCleanerMessage()
        data_msg.period = period
        data1_str = data_msg.SerializeToString()
        msg.data1.length = len(data1_str)
        msg.data1.data = data1_str
        self.run_rabbitmq()
        self.run(debug=False, host=self.WEBhost, port=self.WEBport)


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")
        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:
@@ -102,14 +164,12 @@
        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])
        global save_folder
        save_folder = SAVEfolder

    except Exception as e:
        print(e)
        help()
    server = MyServer(host="localhost", user='chai', password='password123', queue='aerial_rpc', port=5000)
    server = MyServer(MQhost, MQuser, MQpassword, WEBhost, WEBport, SAVEfolder)
    server.run_all()

 src/Server.py.backup

New file
@@ -0,0 +1,115 @@
from posixpath import abspath
import sys              # NOQA: E402    
import os
from flask import Flask, request

batfile_dir = os.path.join(__file__, "..", "protos/compile.bat" )
batfile_path = os.path.abspath(batfile_dir)
os.system(batfile_path)

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



class MyServer(Flask):
    def __init__(self, host, user, password, queue, port):
        Flask.__init__(self, __name__)
        self.host = host
        self.user = user
        self.password = password
        self.queue = queue
        self.port = port

        self.route('/api', methods=['POST'])(self.api)
        self.route('/', methods=['GET'])(self.hello)
        self._def_routes()
        self._device_manager = DeviceManager(host, user, password)

    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):
        if pb.SubCommand.POWER_ON_SUB_COMMAND == req.sub_command:
            print('power on message received')
            self._device_manager.get_device(req.device_id,save_folder)
        elif pb.SubCommand.POWER_OFF_SUB_COMMAND == req.sub_command:
            device = self._device_manager.get_device(req.device_id,save_folder)
            device['power_on'] = False
            print('power off message received')
        return

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

    def api(self):
        req = pb.RequestCommand()
        req.ParseFromString(request.data)
        dispatcher = self.routes.get(req.sys_command, self._default)
        resp = pb.ResponseCommand()
        dispatcher(req, resp)
        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, host: str, user: str, password: str, queue: str) -> None:
        self._message_queue_thread = MessageQueueThread(
            host, user, password, queue, self._device_manager, save_folder)
        self._message_queue_thread.start()

    def run_all(self):
        self.run_rabbitmq(self.host, self.user, self.password, self.queue)
        self.run(debug=False, host="0.0.0.0", port=self.port)

    def send_controll_message(self, device_id, msg):
        device = self._device_manager.get_device(device_id,save_folder)
        if device is None:
            channel = device.controll_channel
            channel.basic_publish(
                exchange='',
                routing_key='',
                body=msg.SerializeToString())

    def send_dust_cleaner_cmd(self, device_id, sub_cmd, period):
        msg = pb.ControllMessage()
        msg.sys_command = pb.ControlSysCommand.DUST_CLEANER_COMMAND
        msg.sub_command = sub_cmd
        data_msg = pb.DustCleanerMessage()
        data_msg.period = period
        data1_str = data_msg.SerializeToString()
        msg.data1.length = len(data1_str)
        msg.data1.data = data1_str


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:
        help()
    try:
        MQhost = str(sys.argv[1])
        MQuser = str(sys.argv[2])
        MQpassword = str(sys.argv[3])
        QUEUEname = str(sys.argv[4])
        WEBport = int(sys.argv[5])
        SAVEfolder = str(sys.argv[6])
        global save_folder
        save_folder = SAVEfolder

    except Exception as e:
        print(e)
        help()
    server = MyServer(host="localhost", user='chai', password='password123', queue='aerial_rpc', port=5000)
    server.run_all()

 src/engine/AcquisitionMain.py

@@ -11,6 +11,8 @@
from .CaptureThread import CaptureThread
from .ControllThread import ControllThread
from .MicroWaveUDPThread import MicroWaveUDPThread
from . import MessageThreads as ths
from .CameraDevice import cameraController

# add python path of src to sys.path
src_path = os.path.join(__file__, *(['..'] * 2))
@@ -60,198 +62,383 @@
            f.close()
        return

class AcquisitionMain:
class AcquisitionMain (object):
    """
    acquisition main class
    """
    def __init__(self,
                 rabbitmq_host: str, 
                 rabbitmq_user: str, 
                 rabbitmq_password: str,
                 rabbitmq_queue: str, 
                 device_id: str, 
                 api_url: str,
                 udp_server_host: str,
                 config: Config) -> None:
        """
        init AcquisitionMain class
        Args:
            rabbitmq_host: rabbitmq server host
            rabbitmq_user: rabbitmq user
            rabbitmq_password: rabbitmq password
            rabbitmq_queue: rabbitmq upload queue
            device_id: device id
            api_url: server api url
            udp_server_host: udp server host for micro wave, localhost for testing or 192.168.2.65 for production
            config: config object             
        """
        self.rabbitmq_host: str = rabbitmq_host
        self.udp_server_host: str = udp_server_host
        self.rabbitmq_user: str = rabbitmq_user
        self.rabbitmq_password: str = rabbitmq_password
        self.rabbitmq_queue: str = rabbitmq_queue
        self.device_id: str = device_id
        self.message_send: int = 0
        self.api_url: str = api_url
        self.config = config
        self.image_quality: float = 95
        self.cameras: list = []
        self.msg_queue = queue.Queue()
        # thread members
        self._controll_thread = None
        self._capture_thread = None
        self._udp_thread = None
    # def __init__(self,
    #              rabbitmq_host: str, 
    #              rabbitmq_user: str, 
    #              rabbitmq_password: str,
    #              rabbitmq_queue: str, 
    #              device_id: str, 
    #              api_url: str,
    #              udp_server_host: str,
    #              config: Config) -> None:
    #     """
    #     init AcquisitionMain class
    #     Args:
    #         rabbitmq_host: rabbitmq server host
    #         rabbitmq_user: rabbitmq user
    #         rabbitmq_password: rabbitmq password
    #         rabbitmq_queue: rabbitmq upload queue
    #         device_id: device id
    #         api_url: server api url
    #         udp_server_host: udp server host for micro wave, localhost for testing or 192.168.2.65 for production
    #         config: config object             
    #     """
    #     self.rabbitmq_host: str = rabbitmq_host
    #     self.udp_server_host: str = udp_server_host
    #     self.rabbitmq_user: str = rabbitmq_user
    #     self.rabbitmq_password: str = rabbitmq_password
    #     self.rabbitmq_queue: str = rabbitmq_queue
    #     self.device_id: str = device_id
    #     self.message_send: int = 0
    #     self.api_url: str = api_url
    #     self.config = config
    #     self.image_quality: float = 95
    #     self.cameras: list = []
    #     self.msg_queue = queue.Queue()
    #     # thread members
    #     self._controll_thread = None
    #     self._capture_thread = None
    #     self._udp_thread = None

    def _start_udp_thread(self):
        """start udp thread"""
        if self._udp_thread is None:
            self._udp_thread = MicroWaveUDPThread((self.udp_server_host, 9911), self.msg_queue)
            self._udp_thread.setDaemon(True)
            self._udp_thread.start()

    def _stop_udp_thread(self):
        """stop udp thread"""
        if not self._udp_thread is None:
            self._udp_thread.stop()
            self._udp_thread = None
    def __init__(self, MQhost, MQuser, MQpassword, deviceID, MICROWAVEhost):

    def _start_capture_thread(self):
        """start camera capture thread"""
        if self._capture_thread is None:
            self._open_cameras()
            if len(self.cameras) == 0:
                print("No camera available")
                return
            self._capture_thread = CaptureThread(
                self.cameras, self.msg_queue, self.config.interval, self.image_quality)
            self._capture_thread.setDaemon(True)
            self._capture_thread.start()
        # device ID
        self.device_id = deviceID
        # rabbit MQ
        self.MQhost = MQhost
        self.MQuser = MQuser
        self.MQpassword = MQpassword
        self.parameters = None
        # camera
        self.IMAGE_WIDTH = 3840
        self.IMAGE_HEIGHT = 2880
        self.CAMERA_COUNT = 4
        # microwave
        self.MICROWAVEhost = MICROWAVEhost
        # threads
        self._device_state_sender_thread = None
        self._alarm_state_sender_thread = None
        self._command_request_receiver_thread = None
        self._command_response_sender_thread = None
        # queue
        self.device_state_queue = queue.Queue()
        self.alarm_state_queue = queue.Queue()
        self.command_request_queue = queue.Queue()
        self.command_response_queue = queue.Queue()
        self.image_data_queue = queue.Queue()
        self.udp_data_queue = queue.Queue()
        self.ply_data_queue = queue.Queue()

    def _stop_capture_thread(self):
        """stop camera capture thread"""
        if not self._capture_thread is None:
            self._capture_thread.stop()
            self._capture_thread = None
        self.queue_dict = {
            "device_state":self.device_state_queue,
            "alarm_state":self.alarm_state_queue,
            "command_request":self.command_request_queue,
            "command_response":self.command_response_queue,
            "image_data":self.image_data_queue,
            "udp_data":self.udp_data_queue,
            "ply_data":self.ply_data_queue
        }

    def _start_controll_thread(self):
        """start controll thread"""
        if self._controll_thread is None:
            self._controll_thread = ControllThread(
                self.rabbitmq_host, self.rabbitmq_user, self.rabbitmq_password, self.device_id + "_callback",
                self.config)
            self._controll_thread.setDaemon(True)
            self._controll_thread.start()

    def _stop_controll_thread(self):
        """stop controll thread"""
        if not self._controll_thread is None:
            self._controll_thread.stop()
            self._controll_thread = None

    def _connect_message_queue(self) -> None:
        """connect message queue of rabbitmq"""
        credentials = pika.PlainCredentials(self.rabbitmq_user, self.rabbitmq_password)
        parameters = pika.ConnectionParameters(
            credentials=credentials, host=self.rabbitmq_host)
    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(parameters)
        self.channel = self._connection.channel()
        self.channel.queue_declare(queue='', exclusive=True)
        try:
            self.connection = pika.BlockingConnection(self.parameters)
            print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
            self.connection.close()
        except Exception :
            print("can't connecting to the MQ server")
            exit(1)

    def _close_message_queue(self) -> None:
        """close message queue of rabbitmq"""
        self._connection.close()
    def _start_device_state_up_msg_sender_thread(self):
        """start device state thread"""
        if self._device_state_sender_thread is None:
            self._device_state_sender_thread = \
                ths.device_state_senderThread(self.parameters,\
                    self.device_state_queue,self.device_id)
            self._device_state_sender_thread.setDaemon(True)
            self._device_state_sender_thread.start()

    def _set_logging(self):
        """set logging"""
        logging.basicConfig(level=logging.INFO,
                            format='%(asctime)s %(levelname)s %(message)s',
                            datefmt='%Y-%m-%d %H:%M:%S')
        self.logger = logging.getLogger(__name__)
    def _stop_device_state_up_msg_sender_thread(self):
        """stop device state thread"""
        if self._device_state_sender_thread is not None:
            self._device_state_sender_thread.stop()
            self._device_state_sender_thread = None

    def _send_data(self, req: pb.RequestCommand) -> None:
        """send data to rabbitmq"""
        req.device_id = self.device_id
        body = req.SerializeToString()
        self.channel.basic_publish(
            exchange='',
            routing_key=self.rabbitmq_queue,
            body=body)
        self.message_send += 1
        return
    def _start_alarm_state_up_msg_sender_thread(self):
        """start alarm state thread"""
        if self._alarm_state_sender_thread is None:
            self._alarm_state_sender_thread = \
                ths.alarm_state_senderThread(self.parameters,\
                    self.alarm_state_queue,self.device_id)
            self._alarm_state_sender_thread.setDaemon(True)
            self._alarm_state_sender_thread.start()

    def _call(self, req: pb.RequestCommand, timeout: int = 10) -> pb.ResponseCommand:
        """call server api"""
        req.device_id = self.device_id
        self.response = None
        body = req.SerializeToString()
        response = requests.post(
            self.api_url,
            data=body,
            headers={'Content-Type': 'application/x-protobuf'},
            timeout=timeout)
        resp = pb.ResponseCommand()
        resp.ParseFromString(response.content)
        return resp
    def _stop_alarm_state_up_msg_sender_thread(self):
        """stop alarm state thread"""
        if self._alarm_state_sender_thread is not None:
            self._alarm_state_sender_thread.stop()
            self._alarm_state_sender_thread = None

    def _send_power_on_message(self) -> pb.ResponseCommand:
        """send power on message"""
        print("send power on message")
        req = pb.RequestCommand()
        req.sys_command = pb.SysCommand.SYS_COMMAND
        req.sub_command = pb.SubCommand.POWER_ON_SUB_COMMAND
        return self._call(req)

    def _send_power_off_message(self) -> pb.ResponseCommand:
        """send power off message"""
        print("send power off message")
        req = pb.RequestCommand()
        req.sys_command = pb.SysCommand.SYS_COMMAND
        req.sub_command = pb.SubCommand.POWER_OFF_SUB_COMMAND
        return self._call(req)
    def _start_command_response_sender_thread(self):
        """start command response thread"""
        if self._command_response_sender_thread is None:
            self._command_response_sender_thread = \
                ths.command_response_senderThread(self.parameters, \
                    self.command_response_queue,self.device_id)
            self._command_response_sender_thread.setDaemon(True)
            self._command_response_sender_thread.start()

    def _open_cameras(self):
        """
        Test the ports and returns a tuple with the available ports and the ones that are working.
        """
        self.cameras = []
        self.working_ports = []
        for dev_port in range(CAMERA_COUNT):
            camera = cv2.VideoCapture(dev_port, cv2.CAP_DSHOW)
            camera.set(cv2.CAP_PROP_FRAME_WIDTH, IMAGE_WIDTH)
            camera.set(cv2.CAP_PROP_FRAME_HEIGHT, IMAGE_HEIGHT)
            camera.set(cv2.CAP_PROP_FPS, 1)
            if camera.isOpened():
                is_reading, img = camera.read()
                w = camera.get(3)
                h = camera.get(4)
                if is_reading:
                    print("Port %s is working and reads images (%s x %s)" %
                          (dev_port, h, w))
                    self.working_ports.append(dev_port)
                    self.cameras.append(camera)
    def _stop_command_response_sender_thread(self):
        """stop command response thread"""
        if self._command_response_sender_thread is not None:
            self._command_response_sender_thread.stop()
            self._command_response_sender_thread = None

    def _main_loop(self):
        # main loop of the acquisition main program
        self._connect_message_queue()
        while True:
            try:
                if not self.msg_queue.empty():
                    msg = self.msg_queue.get()
                    self._send_data(msg)
                    print(time.asctime(), "send message")
                time.sleep(0.1)
            except Exception as e:
                break
        self._close_message_queue()
        self._stop_udp_thread()
        self._stop_capture_thread()
        self._stop_controll_thread()
        print("Stopped")

    def _start_command_request_receiver_thread(self):
        """start command request thread"""
        if self._command_request_receiver_thread is None:
            self._command_request_receiver_thread = \
                ths.command_request_receiverThread(self.parameters,\
                self.command_request_queue,self.command_response_queue,self.device_id)
            self._command_request_receiver_thread.setDaemon(True)
            self._command_request_receiver_thread.start()

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


    def run_rabbitmq(self):
        self._connect_to_MQ()
        self._start_device_state_up_msg_sender_thread()
        self._start_alarm_state_up_msg_sender_thread()
        self._start_command_response_sender_thread()
        self._start_command_request_receiver_thread()

    def run_device(self):
        self.connecting_camera()
        self.connecting_microwave()
        self.connecting_dustCleaner()

    def connecting_camera(self):
        # init the camera controller
        cc = cameraController(self.IMAGE_WIDTH,self.IMAGE_HEIGHT,\
            self.CAMERA_COUNT,self.device_id,self.queue_dict)
        # connecting to camera via opencv
        cc.open_cameras()
        # capture the data, and wrap it to data queue
        cc._start_captureThread()
        pass

    def connecting_microwave(self):
        pass

    def connecting_dustCleaner(self):
        pass



    def run_all(self):
        # connecting to MQ server,launching the message delivery queues
        self.run_rabbitmq()
        # connecting to devices and wrap data to the corressponding queue
        self.run_device()


    def start(self):
        self.run_all()





























    # def _start_udp_thread(self):
    #     """start udp thread"""
    #     if self._udp_thread is None:
    #         self._udp_thread = MicroWaveUDPThread((self.udp_server_host, 9911), self.msg_queue)
    #         self._udp_thread.setDaemon(True)
    #         self._udp_thread.start()

    # def _stop_udp_thread(self):
    #     """stop udp thread"""
    #     if not self._udp_thread is None:
    #         self._udp_thread.stop()
    #         self._udp_thread = None

    # def _start_capture_thread(self):
    #     """start camera capture thread"""
    #     if self._capture_thread is None:
    #         self._open_cameras()
    #         if len(self.cameras) == 0:
    #             print("No camera available")
    #             return
    #         self._capture_thread = CaptureThread(
    #             self.cameras, self.msg_queue, self.config.interval, self.image_quality)
    #         self._capture_thread.setDaemon(True)
    #         self._capture_thread.start()

    # def _stop_capture_thread(self):
    #     """stop camera capture thread"""
    #     if not self._capture_thread is None:
    #         self._capture_thread.stop()
    #         self._capture_thread = None

    # def _start_controll_thread(self):
    #     """start controll thread"""
    #     if self._controll_thread is None:
    #         self._controll_thread = ControllThread(
    #             self.rabbitmq_host, self.rabbitmq_user, self.rabbitmq_password, self.device_id + "_callback",
    #             self.config)
    #         self._controll_thread.setDaemon(True)
    #         self._controll_thread.start()

    # def _stop_controll_thread(self):
    #     """stop controll thread"""
    #     if not self._controll_thread is None:
    #         self._controll_thread.stop()
    #         self._controll_thread = None

    # def _connect_message_queue(self) -> None:
    #     """connect message queue of rabbitmq"""
    #     credentials = pika.PlainCredentials(self.rabbitmq_user, self.rabbitmq_password)
    #     parameters = pika.ConnectionParameters(
    #         credentials=credentials, host=self.rabbitmq_host)
    #     # 连接队列服务器
    #     self._connection = pika.BlockingConnection(parameters)
    #     self.channel = self._connection.channel()
    #     self.channel.queue_declare(queue='', exclusive=True)

    # def _close_message_queue(self) -> None:
    #     """close message queue of rabbitmq"""
    #     self._connection.close()

    # def _set_logging(self):
    #     """set logging"""
    #     logging.basicConfig(level=logging.INFO,
    #                         format='%(asctime)s %(levelname)s %(message)s',
    #                         datefmt='%Y-%m-%d %H:%M:%S')
    #     self.logger = logging.getLogger(__name__)

    # def _send_data(self, req: pb.RequestCommand) -> None:
    #     """send data to rabbitmq"""
    #     req.device_id = self.device_id
    #     body = req.SerializeToString()
    #     self.channel.basic_publish(
    #         exchange='',
    #         routing_key=self.rabbitmq_queue,
    #         body=body)
    #     self.message_send += 1
    #     return

    # def _call(self, req: pb.RequestCommand, timeout: int = 10) -> pb.ResponseCommand:
    #     """call server api"""
    #     req.device_id = self.device_id
    #     self.response = None
    #     body = req.SerializeToString()
    #     response = requests.post(
    #         self.api_url,
    #         data=body,
    #         headers={'Content-Type': 'application/x-protobuf'},
    #         timeout=timeout)
    #     resp = pb.ResponseCommand()
    #     resp.ParseFromString(response.content)
    #     return resp

    # def _send_power_on_message(self) -> pb.ResponseCommand:
    #     """send power on message"""
    #     print("send power on message")
    #     req = pb.RequestCommand()
    #     req.sys_command = pb.SysCommand.SYS_COMMAND
    #     req.sub_command = pb.SubCommand.POWER_ON_SUB_COMMAND
    #     return self._call(req)

    # def _send_power_off_message(self) -> pb.ResponseCommand:
    #     """send power off message"""
    #     print("send power off message")
    #     req = pb.RequestCommand()
    #     req.sys_command = pb.SysCommand.SYS_COMMAND
    #     req.sub_command = pb.SubCommand.POWER_OFF_SUB_COMMAND
    #     return self._call(req)

    # def _open_cameras(self):
    #     """
    #     Test the ports and returns a tuple with the available ports and the ones that are working.
    #     """
    #     self.cameras = []
    #     self.working_ports = []
    #     for dev_port in range(CAMERA_COUNT):
    #         camera = cv2.VideoCapture(dev_port, cv2.CAP_DSHOW)
    #         camera.set(cv2.CAP_PROP_FRAME_WIDTH, IMAGE_WIDTH)
    #         camera.set(cv2.CAP_PROP_FRAME_HEIGHT, IMAGE_HEIGHT)
    #         camera.set(cv2.CAP_PROP_FPS, 1)
    #         if camera.isOpened():
    #             is_reading, img = camera.read()
    #             w = camera.get(3)
    #             h = camera.get(4)
    #             if is_reading:
    #                 print("Port %s is working and reads images (%s x %s)" %
    #                       (dev_port, h, w))
    #                 self.working_ports.append(dev_port)
    #                 self.cameras.append(camera)

    # def _main_loop(self):
    #     # main loop of the acquisition main program
    #     self._connect_message_queue()
    #     while True:
    #         try:
    #             if not self.msg_queue.empty():
    #                 msg = self.msg_queue.get()
    #                 self._send_data(msg)
    #                 print(time.asctime(), "send message")
    #             time.sleep(0.1)
    #         except Exception as e:
    #             break
    #     self._close_message_queue()
    #     self._stop_udp_thread()
    #     self._stop_capture_thread()
    #     self._stop_controll_thread()
    #     print("Stopped")
        
    def start(self) -> None:
        """start all threads"""
        self._start_controll_thread()
        self._start_capture_thread()
        self._start_udp_thread()
        self._main_loop()
    # def start(self) -> None:
    #     """start all threads"""
    #     self._start_controll_thread()
    #     self._start_capture_thread()
    #     self._start_udp_thread()
    #     self._main_loop()

 src/engine/AcquisitionMain.py.backup

New file
@@ -0,0 +1,257 @@
import sys              # NOQA: E402
import os
import cv2
import logging
import json
import pika
import queue
import requests
import time

from .CaptureThread import CaptureThread
from .ControllThread import ControllThread
from .MicroWaveUDPThread import MicroWaveUDPThread

# 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


IMAGE_WIDTH = 3840
IMAGE_HEIGHT = 2880
CAMERA_COUNT = 4

class Config:
    """
    config class
    """
    def __init__(self, config_file: str):
        self.interval: float = 10
        self.resolution = '4k'

        self.clean_lens_start_angle = 10
        self.clean_lens_end_angle = 90
        self.clean_lens_num = 2
        self.clean_lens_period = 1200
        self.clean_self_start_angle = 10
        self.clean_self_end_angle = 55
        self.clean_self_num = 2
        self.clean_self_period = 240

        # load config file
        with open(config_file, 'r') as f:
            self.config = json.load(f)
            self.device_id = self.config['Device_ID']
            self.interval = self.config['Captue_Image_para']['capture_interval']
            self.resolution = self.config['Captue_Image_para']['resolution']

            Clean_Brush_para = self.config['Clean_Brush_para']
            self.clean_lens_start_angle = Clean_Brush_para['clean_lens_start_angle']
            self.clean_lens_end_angle = Clean_Brush_para['clean_lens_end_angle']
            self.clean_lens_num = Clean_Brush_para['clean_lens_num']
            self.clean_lens_period = Clean_Brush_para['clean_lens_period']
            self.clean_self_start_angle = Clean_Brush_para['clean_self_start_angle']
            self.clean_self_end_angle = Clean_Brush_para['clean_self_end_angle']
            self.clean_self_num = Clean_Brush_para['clean_self_num']
            self.clean_self_period = Clean_Brush_para['clean_self_period']
            f.close()
        return

class AcquisitionMain:
    """
    acquisition main class
    """
    def __init__(self,
                 rabbitmq_host: str, 
                 rabbitmq_user: str, 
                 rabbitmq_password: str,
                 rabbitmq_queue: str, 
                 device_id: str, 
                 api_url: str,
                 udp_server_host: str,
                 config: Config) -> None:
        """
        init AcquisitionMain class
        Args:
            rabbitmq_host: rabbitmq server host
            rabbitmq_user: rabbitmq user
            rabbitmq_password: rabbitmq password
            rabbitmq_queue: rabbitmq upload queue
            device_id: device id
            api_url: server api url
            udp_server_host: udp server host for micro wave, localhost for testing or 192.168.2.65 for production
            config: config object             
        """
        self.rabbitmq_host: str = rabbitmq_host
        self.udp_server_host: str = udp_server_host
        self.rabbitmq_user: str = rabbitmq_user
        self.rabbitmq_password: str = rabbitmq_password
        self.rabbitmq_queue: str = rabbitmq_queue
        self.device_id: str = device_id
        self.message_send: int = 0
        self.api_url: str = api_url
        self.config = config
        self.image_quality: float = 95
        self.cameras: list = []
        self.msg_queue = queue.Queue()
        # thread members
        self._controll_thread = None
        self._capture_thread = None
        self._udp_thread = None

    def _start_udp_thread(self):
        """start udp thread"""
        if self._udp_thread is None:
            self._udp_thread = MicroWaveUDPThread((self.udp_server_host, 9911), self.msg_queue)
            self._udp_thread.setDaemon(True)
            self._udp_thread.start()

    def _stop_udp_thread(self):
        """stop udp thread"""
        if not self._udp_thread is None:
            self._udp_thread.stop()
            self._udp_thread = None

    def _start_capture_thread(self):
        """start camera capture thread"""
        if self._capture_thread is None:
            self._open_cameras()
            if len(self.cameras) == 0:
                print("No camera available")
                return
            self._capture_thread = CaptureThread(
                self.cameras, self.msg_queue, self.config.interval, self.image_quality)
            self._capture_thread.setDaemon(True)
            self._capture_thread.start()

    def _stop_capture_thread(self):
        """stop camera capture thread"""
        if not self._capture_thread is None:
            self._capture_thread.stop()
            self._capture_thread = None

    def _start_controll_thread(self):
        """start controll thread"""
        if self._controll_thread is None:
            self._controll_thread = ControllThread(
                self.rabbitmq_host, self.rabbitmq_user, self.rabbitmq_password, self.device_id + "_callback",
                self.config)
            self._controll_thread.setDaemon(True)
            self._controll_thread.start()

    def _stop_controll_thread(self):
        """stop controll thread"""
        if not self._controll_thread is None:
            self._controll_thread.stop()
            self._controll_thread = None

    def _connect_message_queue(self) -> None:
        """connect message queue of rabbitmq"""
        credentials = pika.PlainCredentials(self.rabbitmq_user, self.rabbitmq_password)
        parameters = pika.ConnectionParameters(
            credentials=credentials, host=self.rabbitmq_host)
        # 连接队列服务器
        self._connection = pika.BlockingConnection(parameters)
        self.channel = self._connection.channel()
        self.channel.queue_declare(queue='', exclusive=True)

    def _close_message_queue(self) -> None:
        """close message queue of rabbitmq"""
        self._connection.close()

    def _set_logging(self):
        """set logging"""
        logging.basicConfig(level=logging.INFO,
                            format='%(asctime)s %(levelname)s %(message)s',
                            datefmt='%Y-%m-%d %H:%M:%S')
        self.logger = logging.getLogger(__name__)

    def _send_data(self, req: pb.RequestCommand) -> None:
        """send data to rabbitmq"""
        req.device_id = self.device_id
        body = req.SerializeToString()
        self.channel.basic_publish(
            exchange='',
            routing_key=self.rabbitmq_queue,
            body=body)
        self.message_send += 1
        return

    def _call(self, req: pb.RequestCommand, timeout: int = 10) -> pb.ResponseCommand:
        """call server api"""
        req.device_id = self.device_id
        self.response = None
        body = req.SerializeToString()
        response = requests.post(
            self.api_url,
            data=body,
            headers={'Content-Type': 'application/x-protobuf'},
            timeout=timeout)
        resp = pb.ResponseCommand()
        resp.ParseFromString(response.content)
        return resp

    def _send_power_on_message(self) -> pb.ResponseCommand:
        """send power on message"""
        print("send power on message")
        req = pb.RequestCommand()
        req.sys_command = pb.SysCommand.SYS_COMMAND
        req.sub_command = pb.SubCommand.POWER_ON_SUB_COMMAND
        return self._call(req)

    def _send_power_off_message(self) -> pb.ResponseCommand:
        """send power off message"""
        print("send power off message")
        req = pb.RequestCommand()
        req.sys_command = pb.SysCommand.SYS_COMMAND
        req.sub_command = pb.SubCommand.POWER_OFF_SUB_COMMAND
        return self._call(req)

    def _open_cameras(self):
        """
        Test the ports and returns a tuple with the available ports and the ones that are working.
        """
        self.cameras = []
        self.working_ports = []
        for dev_port in range(CAMERA_COUNT):
            camera = cv2.VideoCapture(dev_port, cv2.CAP_DSHOW)
            camera.set(cv2.CAP_PROP_FRAME_WIDTH, IMAGE_WIDTH)
            camera.set(cv2.CAP_PROP_FRAME_HEIGHT, IMAGE_HEIGHT)
            camera.set(cv2.CAP_PROP_FPS, 1)
            if camera.isOpened():
                is_reading, img = camera.read()
                w = camera.get(3)
                h = camera.get(4)
                if is_reading:
                    print("Port %s is working and reads images (%s x %s)" %
                          (dev_port, h, w))
                    self.working_ports.append(dev_port)
                    self.cameras.append(camera)

    def _main_loop(self):
        # main loop of the acquisition main program
        self._connect_message_queue()
        while True:
            try:
                if not self.msg_queue.empty():
                    msg = self.msg_queue.get()
                    self._send_data(msg)
                    print(time.asctime(), "send message")
                time.sleep(0.1)
            except Exception as e:
                break
        self._close_message_queue()
        self._stop_udp_thread()
        self._stop_capture_thread()
        self._stop_controll_thread()
        print("Stopped")
        
    def start(self) -> None:
        """start all threads"""
        self._start_controll_thread()
        self._start_capture_thread()
        self._start_udp_thread()
        self._main_loop()

 src/engine/CameraDevice.py

New file
@@ -0,0 +1,121 @@
import cv2
import time
import threading
import numpy as np
import os, sys

# 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

class cameraController(object):

    def __init__(self,IMAGE_WIDTH,IMAGE_HEIGHT,CAMERA_COUNT,device_id,queue_dict):
        self.image_width = IMAGE_WIDTH
        self.image_height = IMAGE_HEIGHT
        self.camera_count = CAMERA_COUNT
        self.device_id = device_id
        self.queue_dict = queue_dict
        self.device_state_queue = self.queue_dict["device_state"]
        self.image_data_queue = self.queue_dict["image_data"]
        self._capture_thread = None

    def open_cameras(self):
        """
        Test the ports and returns a tuple with the available ports and the ones that are working.
        """
        self.cameras = []
        self.working_ports = []
        for dev_port in range(self.CAMERA_COUNT):
            camera = cv2.VideoCapture(dev_port, cv2.CAP_DSHOW)
            camera.set(cv2.CAP_PROP_FRAME_WIDTH, self.IMAGE_WIDTH)
            camera.set(cv2.CAP_PROP_FRAME_HEIGHT, self.IMAGE_HEIGHT)
            camera.set(cv2.CAP_PROP_FPS, 1)
            if camera.isOpened():
                is_reading, img = camera.read()
                w = camera.get(3)
                h = camera.get(4)
                if is_reading:
                    self.working_ports.append(dev_port)
                    self.cameras.append(camera)
                    msg = f"{time.asctime()} {self.device_id} camera port {dev_port} is working and reads images ({h} x {w})"
                else:
                    msg = f"{time.asctime()} {self.device_id} camera port {dev_port} is not working"

                # add msg to queue
                self.device_state_queue.put(msg)
                print(msg)


    def _start_captureThread(self):
        """start capture thread"""
        if self._capture_thread is None:
            self._capture_thread = \
                CaptureThread(self.cameras,self.image_data_queue,interval=10)
            self._capture_thread.setDaemon(True)
            self._capture_thread.start()

    def _stop_captureThread(self):
        """stop capture thread"""
        if self._capture_thread is not None:
            self._capture_thread.stop()
            self._capture_thread = None


class CaptureThread(threading.Thread):
    """
    camera capture thread
    """
    def __init__(self, cameras, msg_queue, interval, image_quality = 95):
        """init the thread"""
        threading.Thread.__init__(self)
        self.cameras = cameras
        self.msg_queue = msg_queue
        self.interval = interval
        self.image_quality = image_quality
        self.is_stop = False
    
    def stop(self):
        """stop the thread"""
        self.is_stop = True

    def run(self):
        print("CaptureThread started")
        self.next_time = time.time() + self.interval
        while not self.is_stop:
            this_time = time.time()
            if this_time > self.next_time:
                data1 = pb.ImagesMessage()
                for i in range(len(self.cameras)):
                    _, frame = self.cameras[i].read()
                    encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), self.image_quality]
                    _, encimg = cv2.imencode('.jpg', frame, encode_param)
                    if frame is not None:
                        np_array = np.asarray(encimg)  # frame 转换为numpy数组
                        byte_array = np_array.tobytes()  # numpy数组转换为byte数组
                        if i == 0:
                            data1.image1 = byte_array
                        elif i == 1:
                            data1.image2 = byte_array
                        elif i == 2:
                            data1.image3 = byte_array
                        elif i == 3:
                            data1.image4 = byte_array
                data1_str = data1.SerializeToString()

                # create protobuf message
                req = pb.RequestCommand()
                req.sys_command = pb.SysCommand.FIVE_G_COMMAND
                req.sub_command = pb.SubCommand.FIVE_G_SEND_IMAGE_SUB_COMMAND
                req.data1.length = len(data1_str)
                req.data1.data = data1_str
                # add message to queue
                self.msg_queue.put(req)
                self.next_time += self.interval
            time.sleep(0.1)
        for camera in self.cameras:
            camera.release()
        print("CaptureThread stopped")

 src/engine/MessageThreads.py

New file
@@ -0,0 +1,650 @@
import threading
import pika
import time



class Message(object):
    def __init__(self,sender_id,sender_key,body,routing_key):
        self.sender_id = sender_id
        self.sender_key = sender_key
        self.body = body
        self.routing_key = routing_key

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 stop(self):
        """stop the thread"""
        self.is_stop = True
        self.connection.close()

    def run(self):
    
        while not self.is_stop:
            channel = self.connection.channel()
            channel.exchange_declare(exchange=self.exchangeName, exchange_type=self.exchangeType)

            while True:
                try:
                    if not self.queue.empty():
                        msg = self.msg_queue.get()
                        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}"
                        
                        print(response_tag)

                    elif self.queue.empty():
                        time.sleep(1)
                except Exception as e:
                    break

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 stop(self):
        """stop the thread"""
        self.is_stop = True
        self.connection.close()


    def run(self):
        while not self.is_stop:

            channel = self.connection.channel()
            channel.exchange_declare(exchange=self.exchangeName, exchange_type=self.exchangeType)

            result = channel.queue_declare(queue='', exclusive=True, durable=True)
            queue_name = result.method.queue

            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)

            channel.basic_qos(prefetch_count=1)
            channel.start_consuming()

    def callback(ch, method, properties, body):
        raise NotImplementedError(
            "Need to implement the callback function")

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

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)

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

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

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)

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

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)


























# 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

@@ -5,4 +5,6 @@
from .DustCleaner import *
from .DustCleanerThread import *
from .MessageQueueThread import *
from .MicroWaveUDPThread import *
from .MicroWaveUDPThread import *
from .MessageThreads import *
from .CameraDevice import *

 src/utils/tools.py

@@ -38,6 +38,8 @@
    return ply_array




if __name__ == "__main__":
    ply_array = read_ply('src\\microwave\\demo\\demoData\\2522.ply')