temp

wu_xinjun

2022-03-07 360ba8a616cb525053f288ec5dc3fc9719734282

temp

 pipeline.pptx

Binary files differ

 src/engine/AcquisitionMain.py

@@ -13,7 +13,7 @@
from .ControllThread import ControllThread
from .MicroWaveUDPThread import MicroWaveUDPThread
from . import MessageThreads as ths
from .CameraDevice import cameraController
from .devices import cameraController

# add python path of src to sys.path
src_path = os.path.join(__file__, *(['..'] * 2))
@@ -145,6 +145,19 @@
            "ply_data":self.ply_data_queue
        }

        # device
        self.cc = None # camera controller
        self.mc = None # microwave controller
        self.dc = None # dust cleaner controller
        self.rc = None # rotation device controler

        self.controller_dict = {
            "camera": self.cc,
            "microwave":self.mc,
            "dustcleaner":self.dc,
            "rotation": self.rc
        }


    def _connect_to_MQ(self):
        credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
@@ -241,7 +254,7 @@
        # connecting to camera via opencv
        cc.open_cameras()
        # capture the data, and wrap it to data queue
        # cc._start_captureThread()
        cc._start_captureThread()
        

    def connecting_microwave(self):
@@ -276,6 +289,7 @@
        self._stop_alarm_state_up_msg_sender_thread()
        self.stop
        print("main thread end.")
        return 0
 
            


 src/engine/MessageThreads.py

@@ -1,6 +1,15 @@
import os
import threading
import pika
import time
import os

# 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



@@ -36,9 +45,10 @@
            while True:
                try:
                    if not self.queue.empty():
                        msg = self.queue.get()
                        if isinstance(msg,str):
                            print(msg)
                        data= self.queue.get()
                        if isinstance(data, pb.CapturedData):
                            msg = pb.DataMessage()
                            msg.
                        else:
                            body = msg.body
                            routing_key = msg.routing_key
@@ -230,6 +240,22 @@
        ch.basic_ack(delivery_tag = method.delivery_tag)


class image_data_senderThread(senderThread):
    def __init__(self, 
        MQparameters,
        input_msg_queue,
        sender_id,
        ):
        super(command_response_senderThread,self).__init__(
                    MQparameters, \
                    input_msg_queue,
                    sender_id,
                    "image_data",
                    "fanout"
                   )







 src/engine/__init__.py

@@ -7,4 +7,4 @@
from .MessageQueueThread import *
from .MicroWaveUDPThread import *
from .MessageThreads import *
from .CameraDevice import *
from .devices import *

 src/engine/devices.py

File was renamed from src/engine/CameraDevice.py
@@ -1,3 +1,4 @@
from lib2to3.pytree import convert
import cv2
import time
import threading
@@ -57,7 +58,7 @@
        """start capture thread"""
        if self._capture_thread is None:
            self._capture_thread = \
                CaptureThread(self.cameras,self.image_data_queue,interval=10)
                CaptureThread(self.cameras,self.image_data_queue,self.device_id,interval=10)
            self._capture_thread.setDaemon(True)
            self._capture_thread.start()

@@ -68,11 +69,18 @@
            self._capture_thread = None


class microwaveController(object):
    pass

class dustCleanerController(object):
    pass 


class CaptureThread(threading.Thread):
    """
    camera capture thread
    """
    def __init__(self, cameras, msg_queue, interval, image_quality = 95):
    def __init__(self, cameras, msg_queue, interval, device_id,image_quality = 95):
        """init the thread"""
        threading.Thread.__init__(self)
        self.cameras = cameras
@@ -80,43 +88,45 @@
        self.interval = interval
        self.image_quality = image_quality
        self.is_stop = False
    
        self.device_id = device_id

    def stop(self):
        """stop the thread"""
        self.is_stop = True

    def run(self):
        print("CaptureThread started")
        encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), self.image_quality]
        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]
                    capture_time = time.time()
                    _, 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()
                        # convert the image to pb.DataField
                        imagedata = pb.DataField()
                        imagedata.data = byte_array
                        imagedata.length = len(byte_array)

                # 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)
                        data_str = imagedata.SerializeToString()

                        # create protobuf message
                        data_msg = pb.CapturedData()
                        data_msg.generate_time = capture_time
                        data_msg.generate_device = self.device_id
                        data_msg.camera_port = i
                        data_msg.image_data = data_str

                        

                        # add message to queue
                        self.msg_queue.put(data_msg)

                self.next_time += self.interval
            time.sleep(0.1)
        for camera in self.cameras:

 src/protos/aerial.proto

@@ -76,6 +76,29 @@
  int32 checksum = 3;                        // 校验码
}

message CapturedData {
  int64 capture_time = 1;                     // 采集时间
  string capture_device =2;                   // 采集设备
  int32 capture_port = 3;                     // 采集设备序号
  DataField DataField = 4;
  string data_type = 5;                         // 采集数据

}

message DataMessage{
  int64 sending_time = 1;
  string sending_device = 2;
  string target_device = 3;
  CapturedData CapturedData = 5;
  
}

message CommandMessage {

}



message ImagesMessage
{
  bytes image1  = 1;
@@ -91,6 +114,24 @@
  bytes data = 2;                            // 数据内容
}

message CommandRequest{

}

message StateMessage{
  string message_type = 1;
  int64 generate_time = 2;
  string generate_device = 3;
  string target_device = 4;
  string message_uuid = 5;
  string str_body = 6;
}






message RequestCommand {
  int32 sync_word = 1;              // 同步字
  string device_id = 2;             // 设备ID