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