import cv2 import time from typing import List, Union import queue # # 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 .DustCleaner import DustCleaner from .Microwave import micorwaveMannager from .MessageThreads import send_logs_tool, send_images_tool, send_pointcloud_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, init_taskuuid:str = "init" ): 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.init_taskuuid = init_taskuuid self.taskuuid = init_taskuuid 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", taskuuid=self.taskuuid) send_images_tool(images=images, images_queue=self.images_queue, source=self.device_id, routing_key=self.routing_key+".images.test0", taskuuid=self.taskuuid) 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, taskuuid:str): self.routing_key = routing_key self.taskuuid = taskuuid (logs,images) = self._capture_image() # generate the msg and put it to queues self._generate_msg(logs,images) self.routing_key = self.init_routing_key self.taskuuid = self.init_taskuuid ######################################## MICROWAVE ########################################### class microwaveController(micorwaveMannager): def __init__(self, MicrowaveHost: str, MicrowavePort: int, init_packet_threshold: Union[int, None], init_timeusage_threshold: Union[int, float, None], device_id:str, logs_queue:queue.Queue, plys_queue:queue.Queue, init_routing_key:str, init_binding_key:str, init_taskuuid:str = "init" ): super().__init__(MicrowaveHost, MicrowavePort) self.device_id = device_id self.plys_queue = plys_queue self.routing_key = init_routing_key self.init_routing_key = init_routing_key self.binding_key = init_binding_key self.init_taskuuid = init_taskuuid self.taskuuid = init_taskuuid # testing the device when instance the class self._get_pointcloud(init_packet_threshold,init_timeusage_threshold) def _get_pointcloud(self, packet_threshold:Union[int, None], timeusage_threshold:Union[int, None]): self._udp_get_thread_launcher(packet_threshold,timeusage_threshold) self._udp2pointcloud_thread_launcher() self._generate_msg() def _generate_msg(self): send_pointcloud_tool(pointclouds=self.pointcloud_queue, plys_queue=self.plys_queue, source=self.device_id, routing_key=self.routing_key+".plys.test0", taskuuid=self.taskuuid) def scan(self,routing_key:str,taskuuid:str): # self.routing_key = routing_key # self.taskuuid = self.init_taskuuid # self._generate_msg() raise NotImplementedError ####################################### DUST CLEANER ######################################### class dustCleanerController(DustCleaner): def __init__(self, com, bps, timeout, logs_queue, device_id, init_routing_key:str, init_binding_key:str, init_taskuuid:str = "init"): super().__init__(com, bps, timeout) self.logs_queue = logs_queue self.device_id = device_id self.routing_key = init_routing_key self.init_routing_key = init_routing_key self.binding_key = init_binding_key self.init_taskuuid = init_taskuuid self.taskuuid = init_taskuuid 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") def clean(self,routing_key:str,taskuuid:str): raise NotImplementedError