from lib2to3.pytree import convert import cv2 import time 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)) # src_path = os.path.abspath(src_path) # sys.path.insert(0, src_path) from ..protos import aerial_pb2 as pb from .DustCleaner import DustCleaner from .MessageThreads import send_logs_tool, send_images_tool ######################################### CAMERAS ############################################ class cameraController(object): 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.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 _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) 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): send_logs_tool(logs=logs, logs_queue=self.logs_queue, source=self.device_id, routing_key=self.routing_key+".logs.test0") send_images_tool(images=images, images_queue=self.images_queue, source=self.device_id, routing_key=self.routing_key+".images.test0") 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.camera_work[dev_port] == True log_list.append(log) image_list.append(img) return log_list, image_list def capture(self,routing_key:str): self.routing_key = routing_key (logs,images) = self._capture_image() self._generate_msg = (logs,images) self.routing_key = self.init_routing_key ######################################## MICROWAVE ########################################### class microwaveController(object): pass ####################################### DUST CLEANER ######################################### class dustCleanerController(DustCleaner): def __init__(self, com, bps, timeout, logs_queue, device_id,routing_key): super().__init__(com, bps, timeout) self.logs_queue = logs_queue self.device_id = device_id self.routing_key = routing_key self._check_cleaner() def _check_cleaner(self): success = self.Open() log = "Unknown log." if success: log = f"The port [{self.port}] refered to the dust cleaner is open." else: log = f"The port [{self.port}] refered to the dust cleaner can not open." # display print(f"{time.asctime()} | {log}") send_logs_tool(logs=[log], logs_queue=self.logs_queue, source = self.device_id, routing_key=self.routing_key+".logs.test0") # class CaptureThread(threading.Thread): # """ # camera capture thread # """ # def __init__(self, cameras, msg_queue, interval, device_id,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 # self.device_id = device_id # def stop(self): # """stop the thread""" # self.is_stop = True # # TODO根据新的数据结构修改队消息队列。 # 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: # for i in range(len(self.cameras)): # _, frame = self.cameras[i].read() # 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数组 # # convert the image to pb.DataField # imagedata = pb.DataField() # imagedata.data = byte_array # imagedata.length = len(byte_array) # 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: # camera.release() # print("CaptureThread stopped")