client enhanced&camera contorller

wu_xinjun

2022-06-15 0dc337197311d0fec373bf479f8ee0ac26b4cad6

client enhanced&camera contorller

 scripts/mqClient.py

@@ -8,7 +8,7 @@


from src.engine.AcquisitionMain import AcquisitionMain
from src.engine.AcquisitionMain import Config
# from src.engine.AcquisitionMain import Config

def quit(signum, frame):
    print('You choose to stop me.')
@@ -20,16 +20,14 @@
        print("EXAMPLE: Client.py 172.17.17.206 chai password123 aerial_rpc pod1 http://172.17.17.206/api 192.168.2.65")
        exit(1)

    if len(sys.argv) != 8:
    if len(sys.argv) != 6:
        help()
    try:
        MQhost = str(sys.argv[1])
        MQuser = str(sys.argv[2])
        MQpassword = str(sys.argv[3])
        QUEUEname = str(sys.argv[4])
        DEVICEname = str(sys.argv[5])
        SERVERhost = str(sys.argv[6])
        MICROWAVEhost = str(sys.argv[7])
        DEVICEname = str(sys.argv[4])
        MICROWAVEhost = str(sys.argv[5])

    except Exception as e:
        print(e)
@@ -40,15 +38,13 @@
    signal.signal(signal.SIGTERM, quit)
    # client = AcquisitionMain('192.168.3.54', 'chai',
    #                       'password123', 'aerial_rpc', 'pod1', 'http://192.168.3.54:5000/api', 10)
    config_dir = os.path.join(__file__, "..",  "../src/config.json" )
    config_path = os.path.abspath(config_dir)
    config = Config(config_path)
    # config_dir = os.path.join(__file__, "..",  "../src/config.json" )
    # config_path = os.path.abspath(config_dir)
    # config = Config(config_path)
    client = AcquisitionMain(MQhost,
                             MQuser,
                             MQpassword, 
                             QUEUEname, 
                             DEVICEname, 
                             SERVERhost, 
                             MICROWAVEhost, # udp server host, 192.168.2.65 for production
                             config)
                            )
    client.start()

 scripts_launch/TEST_startClient.bat

@@ -20,7 +20,7 @@
set MQuser=chai
set MQpassword=password123
set QUEUEname=aerial_rpc
set DEVICEname=pod1
set DEVICEname=client.client.client
set SERVERhost=http://192.168.1.224/api
set MICROWAVEhost=localhost
set LOGfile=ClientTEST.log
@@ -30,6 +30,6 @@
echo Client.py is running ...
@REM -------- Usage: Client.py [MQhost] [MQuser] [MQpassword] [QUEUEname] [DEVICEname] [SERVERhost] [MICROWAVEhost]
@REM %aerialpython% ./src/Client.py %MQhost% %MQuser% %MQpassword% %QUEUEname% %DEVICEname% %SERVERhost% %MICROWAVEhost% >> %LOGfile%
%aerialpython% ./scripts/mqClient.py %MQhost% %MQuser% %MQpassword% %QUEUEname% %DEVICEname% %SERVERhost% %MICROWAVEhost%
%aerialpython% ./scripts/mqClient.py %MQhost% %MQuser% %MQpassword% %DEVICEname% %MICROWAVEhost%
echo Client.py has been terminited.
pause

 src/Server.py

@@ -34,7 +34,7 @@
        self.WEBhost = WEBhost
        self.WEBport = WEBport
        self.save_folder = save_folder
        self.server_name = "server0"
        self.server_name = server_name
        

        # self.route('/api', methods=['POST'])(self.api)
@@ -53,9 +53,14 @@
        self.parameters = pika.ConnectionParameters(
            credentials=credentials, host=self.MQhost)
        # 连接队列服务器
        connection = pika.BlockingConnection(self.parameters)
        print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
        connection.close()
        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 _receiver_Thread_Launcher(self):
@@ -66,7 +71,7 @@
                                    exchange_name='logs_topic',
                                    exchange_type='topic',
                                    MQqueue_name=f'logs_queue',
                                    binding_keys='server.logs.test0',
                                    binding_keys=['server.logs.test0'],
                                    message_queue=self.LogsReceiver_queue
                                    )
        self.LogsReceiver_Thread.setDaemon(True)
@@ -79,7 +84,7 @@
                                    exchange_name='images_topic',
                                    exchange_type='topic',
                                    MQqueue_name=f'images_queue',
                                    binding_keys="server.images.test0",
                                    binding_keys=["server.images.test0"],
                                    message_queue=self.ImagesReceiver_queue
                                    )
        self.ImagesReceiver_Thread.setDaemon(True)
@@ -92,7 +97,7 @@
                                    exchange_name='plys_topic',
                                    exchange_type='topic',
                                    MQqueue_name=f'plys_queue',
                                    binding_keys='server.plys.test0',
                                    binding_keys=['server.plys.test0'],
                                    message_queue=self.PlysReceiver_queue
                                    )
        self.PlysReceiver_Thread.setDaemon(True)
@@ -115,7 +120,7 @@
    
    def _sender_Thread_Launcher(self):

        # launcher command sender thread
        # launcher commands sender thread
        self.CommandsSender_Thread = ths.senderThread(
                                    MQparams=self.parameters,
                                    exchange_name='commands_topic',

 src/engine/AcquisitionMain.py

@@ -1,112 +1,71 @@
import sys              # NOQA: E402
import os
import cv2
import logging
import json
import pika
import queue
import requests
import time
import threading

from .CaptureThread import CaptureThread
from .ControllThread import ControllThread
from .MicroWaveUDPThread import MicroWaveUDPThread
# from .CaptureThread import CaptureThread
# from .ControllThread import ControllThread
# from .MicroWaveUDPThread import MicroWaveUDPThread
from . import MessageThreads as ths
from .devices import cameraController
from .devices import cameraController, dustCleanerController

# # add python path of src to sys.path
# src_path = os.path.join(__file__, *(['..'] * 2))
# src_path = os.path.abspath(src_path)
# sys.path.insert(0, src_path)

from ..protos import aerial_pb2 as pb
# from ..protos import aerial_pb2 as pb


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'
# 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
#         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']
#         # 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
#             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 (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, MQhost, MQuser, MQpassword, deviceID, MICROWAVEhost):
    def __init__(self,
                MQhost:str,
                MQuser:str,
                MQpassword:str,
                deviceID:str,
                MICROWAVEhost:str):

        # device ID
        self.device_id = deviceID
@@ -122,28 +81,14 @@
        # 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()
        self.threads = []

        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
        }
        # queue
        self.CommandsReceiver_queue = queue.Queue()

        self.LogsSender_queue = queue.Queue()
        self.ImagesSender_queue = queue.Queue()
        self.PlysSender_queue = queue.Queue()

        # device
        self.cc = None # camera controller
@@ -159,6 +104,7 @@
        }



    def _connect_to_MQ(self):
        credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
        self.parameters = pika.ConnectionParameters(
@@ -169,100 +115,92 @@
            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")
            print("Can't connecting to the MQ server")
            exit(1)

    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 _receiver_Thread_Launcher(self):

    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
        # launch a commands receiver thread
        self.CommandsReceiver_Thread = ths.receiverThread(
                                    MQparams=self.parameters,
                                    exchange_name='commands_topic',
                                    exchange_type='topic',
                                    MQqueue_name="commands_queue",
                                    binding_keys=[f"{self.device_id}"],
                                    message_queue=self.CommandsReceiver_queue
                                    )
        self.CommandsReceiver_Thread.setDaemon(True)
        self.CommandsReceiver_Thread.setName("commands_thread")
        self.CommandsReceiver_Thread.start()

    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()
        self.threads.append(self.CommandsReceiver_Thread)
        
    def _sender_Thread_Launcher(self):

    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
        # launch a logs sender thread
        self.LogsSender_Thread = ths.senderThread(
                                MQparams=self.parameters,
                                exchange_name="logs_topic",
                                exchange_type="topic",
                                message_queue=self.LogsSender_queue
                        )
        self.LogsSender_Thread.setDaemon(True)
        self.LogsSender_Thread.setName("logs_thread")
        self.LogsSender_Thread.start()

        self.threads.append(self.LogsSender_Thread)


    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()
        # launch a images sender thread
        self.ImagesSender_Thread = ths.senderThread(
                                MQparams=self.parameters,
                                exchange_name="images_topic",
                                exchange_type="topic",
                                message_queue=self.ImagesSender_queue
                        )
        self.ImagesSender_Thread.setDaemon(True)
        self.ImagesSender_Thread.setName("image_thread")
        self.ImagesSender_Thread.start()

    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
        self.threads.append(self.ImagesSender_Thread)

        # launch a plys sender thread
        self.PlysSender_Thread = ths.senderThread(
                                MQparams=self.parameters,
                                exchange_name="plys_topic",
                                exchange_type="topic",
                                message_queue=self.PlysSender_queue
                        )
        self.PlysSender_Thread.setDaemon(True)
        self.PlysSender_Thread.setName("plys_thread")
        self.PlysSender_Thread.start()

        self.threads.append(self.PlysSender_Thread)


    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 _connect_device(self):
        # connect to cameras
        self.cc = cameraController(
                        IMAGE_WIDTH=IMAGE_WIDTH,
                        IMAGE_HEIGHT=IMAGE_HEIGHT,
                        CAMERA_COUNT=CAMERA_COUNT,
                        device_id=self.device_id,
                        logs_queue=self.LogsSender_queue,
                        images_queue=self.ImagesSender_queue,
                        init_binding_key="client.client.client",
                        init_routing_key="server"

        )
        # self.dc = dustCleanerController()

    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()
        self._receiver_Thread_Launcher()
        self._sender_Thread_Launcher()

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

    def connecting_microwave(self):
        pass

    def connecting_dustCleaner(self):
        pass

        self._connect_device()


    def run_all(self):
@@ -273,21 +211,20 @@
        self.run_device()

    def stop_all(self):
        self._stop_alarm_state_up_msg_sender_thread()
        self._stop_device_state_up_msg_sender_thread()
        self._stop_command_request_receiver_thread()
        self._start_command_response_sender_thread()

        for t in self.threads:
            t.stop()
            print(f"stop {t.name}")

    def start(self):
        try:
            self.run_all()
            self._device_state_sender_thread.join()
            print("[*] Client is runing ...")
            self.CommandsReceiver_Thread.join()
        except Exception as e:
            print(e)
            print("running error, stop all threads")
        
        self._stop_alarm_state_up_msg_sender_thread()
        self.stop
            self.stop_all()
        print("main thread end.")
        return 0
 

 src/engine/MessageThreads.py

@@ -50,7 +50,7 @@
    def stop(self):
        """stop the thread"""
        self.is_stop = True
        self.connection.close()
    #     self.connection.close()

    def run(self):
    
@@ -62,7 +62,7 @@

            while True:
                try:
                    if not self.message_queue.empty():
                    if not self.message_queue.empty() and not self.is_stop:
                        
                        # decode the routing key and message body from tuple list
                        message = self.message_queue.get()
@@ -115,7 +115,7 @@
    def stop(self):
        """stop the thread"""
        self.is_stop = True
        self.connection.close()
    #     self.connection.close()


    def run(self):
@@ -130,6 +130,7 @@
            channel.queue_declare(queue=self.MQqueue_name, durable=True)
            for binding_key in self.binding_keys:
                channel.queue_bind(exchange=self.exchange_name, queue=self.MQqueue_name, routing_key=binding_key)
                print(f"[*] Using [{binding_key}] bind [{self.MQqueue_name}] on [{self.exchange_name}]")

            # consume the message
            channel.basic_qos(prefetch_count=1)
@@ -140,6 +141,9 @@
            channel.start_consuming()

    def callback(self,ch, method, properties, body):

        if self.is_stop:
            ch.close()
        self.message_queue.put(body)
        ch.basic_ack(delivery_tag = method.delivery_tag)

@@ -191,7 +195,7 @@
                queue = self.queue_list[i_tag]

                # make sure the queue is not empty
                while not queue.empty():
                while not queue.empty() and not self.is_stop:
                    # decode the msg using cooresponding method
                    queue_tag = self.queue_tags[i_tag]
                    decode_method = self.decode_select(queue_tag)
@@ -214,7 +218,7 @@
        while not queue.empty() and count < nums_limit:

            # extracting the msg from queue
            msg = queue.pop()
            msg = queue.get()
            msg_pb = pb.LogsMessage()
            msg_pb.ParseFromString(msg)

@@ -241,7 +245,7 @@
        while not queue.empty():

            # extracting the msg from queue
            msg = queue.pop()
            msg = queue.get()
            msg_pb = pb.ImagesMessage()
            msg_pb.ParseFromString(msg)

@@ -268,7 +272,7 @@
                    filename = f"{msg_source}_{msg_timestamp}_image{id}.jpg"
                    filepath = os.path.join(self.imagesmsg_folder,filename)
                    with open(filepath, 'wb') as f:
                        f.write(image)
                        f.write(image_byte)

                    # display the info to the console
                    curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
@@ -282,7 +286,7 @@
        while not queue.empty():

            # extracting the msg from queue
            msg = queue.pop()
            msg = queue.get()
            msg_pb = pb.PlyMessage()
            msg_pb.ParseFromString(msg)


 src/engine/devices.py

@@ -4,6 +4,9 @@
import threading
import numpy as np
import os, sys
from typing import List
import queue
from datetime import datetime

# # add python path of src to sys.path
# src_path = os.path.join(__file__, *(['..'] * 2))
@@ -14,65 +17,162 @@

class cameraController(object):

    def __init__(self,IMAGE_WIDTH,IMAGE_HEIGHT,CAMERA_COUNT,device_id,queue_dict):
    def __init__(self,
            IMAGE_WIDTH:int, 
            IMAGE_HEIGHT:int,
            CAMERA_COUNT:int,
            device_id:str,
            logs_queue:queue.Queue,
            images_queue:queue.Queue,
            init_routing_key:str,
            init_binding_key:str
            ):

        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
        self.logs_queue = logs_queue
        self.images_queue = images_queue
        self.routing_key = init_routing_key
        self.init_routing_key = init_routing_key
        self.binding_key = init_binding_key
        self._check_cameras()

    def open_cameras(self):
        """
        Test the ports and returns a tuple with the available ports and the ones that are working.
        """

    def _check_cameras(self):
        self.cameras = []
        self.camera_work = []
        self.working_ports = []
        test_logs = []
        test_images = []

        # init the cameras
        for dev_port in range(self.CAMERA_COUNT):
            # set the props of the camera by cv2 lib
            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)
            self.cameras.append(camera)
            self.camera_work.append(False)

        # test the cameras
        (test_logs, test_images) = self._capture_image(mode = "test")

        # generate the msg and put it to queues
        self._generate_msg(test_logs,test_images)

        
    def _generate_msg(self,logs:List[str],images:List):

        # constract a images message
        imagesMsg = pb.ImagesMessage()
        imagesMsg.source = self.device_id
        curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
        imagesMsg.timestamp = curr_time 
        log_rk = self.routing_key+".logs.test0"
        image_rk = self.routing_key+".images.test0"

        for log, img in zip(logs,images):

            # encoding the images to the bytes
            # encode_params = [int(cv2.IMWRITE_JPEG_QUALITY), self.image_quality]
            # image_encoded = cv2.imencode(".jpg",single_img,encode_params)
            if img is not None:
                success,image_encoded = cv2.imencode(".jpg",img)
                if success:
                    image_bytes = image_encoded.tobytes()
                    image_length = len(image_bytes)
                    # add the encoded image to msg
                    image_item = imagesMsg.Image.add()
                    image_item.length = image_length
                    image_item.data = image_bytes

            if log is not None:
                # constract a logs message
                logsMsg = pb.LogsMessage()
                logsMsg.source = self.device_id
                logsMsg.timestamp = f"{time.asctime()}"
                logsMsg.context = log
                # convert the proto object to string
                logsMsg_bytes = logsMsg.SerializeToString()
                # add logsmsg to queue
                self.logs_queue.put([log_rk,logsMsg_bytes])


        # converting the pb image msg to bytes
        imageMsg_bytes = imagesMsg.SerializeToString()
        # add the imageMsg to the queue
        self.images_queue.put([image_rk,imageMsg_bytes])


    def _capture_image(self,mode:str = "capture"):
        """
        mode -> "test" or "capture"
        """
        TEST = False
        if mode == "test":
            TEST = True     
        else:
            pass

        log_list = []
        image_list = []

        # iterate the cameras
        for dev_port, camera in enumerate(self.cameras):
            # init the flags
            is_reading = False
            log = None
            img = None

            if not TEST and self.camera_work[dev_port] is not True:
                pass
            else:
                # capture the image
                if camera.isOpened():
                    is_reading, img = camera.read()
                    w = camera.get(3)
                    h = camera.get(4)
                    if is_reading:
                        log = f"camera port {dev_port} is working and reads images ({h} x {w})"
                    else:
                        log = f"camera port {dev_port} can not work"
                else:  
                    log = f"camera port {dev_port} can not found"

                # display
                print(f"{time.asctime()} | {log}")

            # add some tags for camera when test
            if TEST:
                self.working_ports.append(dev_port)
                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})"
                    self.camera_work[dev_port] == True

            
                else:
                    msg = f"{time.asctime()} {self.device_id} camera port {dev_port} can not work"
            else:  
                msg = f"{time.asctime()} {self.device_id} camera port {dev_port} can not found"
            log_list.append(log)
            image_list.append(img)

            # add msg to queue
            self.device_state_queue.put(msg)
            print(msg)
        return log_list, image_list


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

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

        (logs,images) = self._capture_image()
        self._generate_msg = (logs,images)

        self.routing_key = self.init_routing_key


class microwaveController(object):
    pass

class dustCleanerController(object):
    
    pass 



 src/protos/aerial.proto

@@ -140,12 +140,12 @@

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


@@ -161,7 +161,7 @@
    int32 cols = 3;
    string order = 4;
    DataType d_types = 5;
    bytes data = 6
    bytes data = 6;
    
  }