import sys # NOQA: E402 import os import pika import queue # from .CaptureThread import CaptureThread # from .ControllThread import ControllThread # from .MicroWaveUDPThread import MicroWaveUDPThread from . import MessageThreads as ths from .devices import cameraController, dustCleanerController,microwaveController # # 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): def __init__(self, MQhost:str, MQuser:str, MQpassword:str, deviceID:str, MICROWAVEhost:str): # device ID self.device_id = deviceID # rabbit MQ self.MQhost = MQhost self.MQuser = MQuser self.MQpassword = MQpassword self.parameters = None # camera self.IMAGE_WIDTH = IMAGE_WIDTH self.IMAGE_HEIGHT = IMAGE_HEIGHT self.CAMERA_COUNT = CAMERA_COUNT # microwave self.MICROWAVEhost = MICROWAVEhost # dust cleaner self.com = "COM11" self.bps = "115200" self.timeout = 1 # threads self.threads = [] # queue self.CommandsReceiver_queue = queue.Queue() self.LogsSender_queue = queue.Queue() self.ImagesSender_queue = queue.Queue() self.PlysSender_queue = queue.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 _receiver_Thread_Launcher(self): # launch a commands receiver thread self.CommandsReceiver_Thread = ths.receiverThread( MQparams=self.parameters, exchange_name='commands_topic', exchange_type='topic', MQqueue_name="commands_queue", binding_keys=[f"{self.device_id}"], message_queue=self.CommandsReceiver_queue ) self.CommandsReceiver_Thread.setDaemon(True) self.CommandsReceiver_Thread.setName("commands_thread") self.CommandsReceiver_Thread.start() self.threads.append(self.CommandsReceiver_Thread) def _sender_Thread_Launcher(self): # launch a logs sender thread self.LogsSender_Thread = ths.senderThread( MQparams=self.parameters, exchange_name="logs_topic", exchange_type="topic", message_queue=self.LogsSender_queue ) self.LogsSender_Thread.setDaemon(True) self.LogsSender_Thread.setName("logs_thread") self.LogsSender_Thread.start() self.threads.append(self.LogsSender_Thread) # launch a images sender thread self.ImagesSender_Thread = ths.senderThread( MQparams=self.parameters, exchange_name="images_topic", exchange_type="topic", message_queue=self.ImagesSender_queue ) self.ImagesSender_Thread.setDaemon(True) self.ImagesSender_Thread.setName("image_thread") self.ImagesSender_Thread.start() self.threads.append(self.ImagesSender_Thread) # launch a plys sender thread self.PlysSender_Thread = ths.senderThread( MQparams=self.parameters, exchange_name="plys_topic", exchange_type="topic", message_queue=self.PlysSender_queue ) self.PlysSender_Thread.setDaemon(True) self.PlysSender_Thread.setName("plys_thread") self.PlysSender_Thread.start() self.threads.append(self.PlysSender_Thread) def _connect_device(self): # connect to cameras self.cc = cameraController( IMAGE_WIDTH=self.IMAGE_WIDTH, IMAGE_HEIGHT=self.IMAGE_HEIGHT, CAMERA_COUNT=self.CAMERA_COUNT, device_id=self.device_id, logs_queue=self.LogsSender_queue, images_queue=self.ImagesSender_queue, init_binding_key="client.client.client", init_routing_key="server" ) self.dc = dustCleanerController( com = self.com, bps = self.bps, timeout=self.timeout, logs_queue=self.LogsSender_queue, device_id=self.device_id, routing_key="server" ) self.mc = microwaveController() def run_rabbitmq(self): self._connect_to_MQ() self._receiver_Thread_Launcher() self._sender_Thread_Launcher() def run_device(self): self._connect_device() 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): for t in self.threads: t.stop() print(f"stop {t.name}") def start(self): try: self.run_all() print("[*] Client is runing ...") self.CommandsReceiver_Thread.join() except Exception as e: print(e) print("running error, stop all threads") self.stop_all() 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()