import sys # NOQA: E402 import os from time import sleep import pika import queue from datetime import datetime # 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 # self.MICROWAVEPort = MICROWAVEhost self.MICROWAVEport = 9911 # 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,heartbeat=0) # 连接队列服务器 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=f"{self.device_id}_commands_queue", binding_keys=[f"{self.device_id}"], message_queue=self.CommandsReceiver_queue ) self.CommandsReceiver_Thread.setDaemon(True) self.CommandsReceiver_Thread.setName("Thread-commandsReceiver") 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("Thread-logsSender") 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("Thread-imageSender") 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("Thread-plysSender") 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=self.device_id, init_routing_key="server0" ) self.dc = dustCleanerController( com = self.com, bps = self.bps, timeout=self.timeout, logs_queue=self.LogsSender_queue, device_id=self.device_id, init_binding_key=self.device_id, init_routing_key="server0" ) self.mc = microwaveController( MicrowaveHost=self.MICROWAVEhost, MicrowavePort=self.MICROWAVEport, init_packet_threshold=100, init_timeusage_threshold=0.2, device_id=self.device_id, logs_queue=self.LogsSender_queue, plys_queue=self.PlysSender_queue, init_binding_key=self.device_id, init_routing_key="server0" ) def command_control(self): while True: try: while not self.CommandsReceiver_queue.empty(): command_bytes= self.CommandsReceiver_queue.get() # parse the command from bytes to the pb object req_command = pb.CommandMessage() req_command.ParseFromString(command_bytes) # extracting some info for use source = req_command.source generatetime = req_command.generatetime syscommand = req_command.syscommand subcommand = req_command.subcommand targetclient = req_command.targetclient returnserver = req_command.returnserver taskuuid = req_command.taskuuid curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S') print(f"[x] {curr_time}|(From {source}): task [{taskuuid}] received!" ) # command for camera if syscommand == pb.SysCommand.CAMERA_COMMAND: # capture image if subcommand == pb.SubCommand.ONE_TIME_CAPTURE: self.cc.capture(routing_key=returnserver,taskuuid=taskuuid) # unknown command else: pass # command for microwave device elif syscommand == pb.SysCommand.MICROWAVE_COMMAND: if subcommand == pb.SubCommand.ONE_CIRCLE_SCAN: self.mc.scan(routing_key=returnserver,taskuuid=taskuuid) # unknown command else: pass # command for dust cleaner elif syscommand == pb.Syscommand.DUSTCLEANER_COMMAND: if subcommand == pb.SubCommand.ONE_CLEAN_SCAN: self.dc.clean(routing_key=returnserver,taskuuid=taskuuid) # unknown command else: pass # unknown command else: pass else: sleep(1) except: break 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() sleep(10) self.command_control() 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.") self.run_all() 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()