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 . import MessageThreads as ths from .devices import cameraController # # 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 (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): # device ID self.device_id = deviceID # rabbit MQ self.MQhost = MQhost self.MQuser = MQuser self.MQpassword = MQpassword self.parameters = None # camera self.IMAGE_WIDTH = 3840 self.IMAGE_HEIGHT = 2880 self.CAMERA_COUNT = 4 # 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.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 } # 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) self.parameters = pika.ConnectionParameters( credentials=credentials, host=self.MQhost) # 连接队列服务器 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 _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 _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 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() 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 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() 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 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 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() 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 def run_all(self): # connecting to MQ server,launching the message delivery queues self.run_rabbitmq() # connecting to devices and wrap data to the corressponding queue 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() def start(self): try: self.run_all() self._device_state_sender_thread.join() except Exception as e: print("running error, stop all threads") self._stop_alarm_state_up_msg_sender_thread() self.stop print("main thread end.") return 0 # 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()