import os import queue import threading import pika import time from time import sleep import sys from datetime import datetime from typing import List # # 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 # class Message(object): # def __init__(self,sender_id,sender_key,body,routing_key): # self.sender_id = sender_id # self.sender_key = sender_key # self.body = body # self.routing_key = routing_key class senderThread(threading.Thread): def __init__(self, MQparams: pika.ConnectionParameters, exchange_name: str, exchange_type:str, # routing_key:str, message_queue:queue.Queue, ): super(senderThread,self).__init__() self.MQparams = MQparams self.exchange_name = exchange_name self.exchange_type = exchange_type # self.routing_key = routing_key self.message_queue = message_queue # connect to RabbitMQ self.is_stop = False self.connection = pika.BlockingConnection(self.MQparams) def stop(self): """stop the thread""" self.is_stop = True self.connection.close() def run(self): while not self.is_stop: # declare a exchange channel = self.connection.channel() channel.exchange_declare(exchange=self.exchange_name, exchange_type=self.exchange_type) while True: try: if not self.message_queue.empty(): # decode the routing key and message body from tuple list message = self.message_queue.get() msg_routing_key = message[0] msg_body = message[1] # publish a PERSISTENT message channel.basic_publish( exchange = self.exchange_name, routing_key = msg_routing_key, body = msg_body, properties=pika.BasicProperties( delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE ) ) timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S') caption = f"[x] {timestamp} | Sending message to[{msg_routing_key}]") print(caption) else: time.sleep(1) except Exception as e: print(e) break class receiverThread(threading.Thread): def __init__(self, MQparams: pika.ConnectionParameters, exchange_name: str, exchange_type:str, MQqueue_name:str, binding_keys:List(str), message_queue:queue.Queue, ): super(senderThread,self).__init__() self.MQparams = MQparams self.exchange_name = exchange_name self.exchange_type = exchange_type self.MQqueue_name = MQqueue_name self.binding_keys = binding_keys self.message_queue = message_queue # connect to RabbitMQ self.is_stop = False self.connection = pika.BlockingConnection(self.MQparams) def stop(self): """stop the thread""" self.is_stop = True self.connection.close() def run(self): while not self.is_stop: # declare a exchange channel = self.connection.channel() channel.exchange_declare(exchange=self.exchange_name, exchange_type=self.exchange_type) # declare the queue and bind it with keys channel.queue_declare(queue=self.MQqueue_name, durable=True) for binding_key in self.binding_keys: channel.queue_bind(exchange=self.exchangeName, queue=self.MQqueue_name, routing_key=binding_key) # consume the message channel.basic_qos(prefetch_count=1) channel.basic_consume( queue=self.MQqueue_name, on_message_callback=self.callback ) channel.start_consuming() def callback(self,ch, method, properties, body): self.message_queue.put(body) ch.basic_ack(delivery_tag = method.delivery_tag) class decodeThread(threading.Thread): def __init__(self, queue_list:List(queue.Queue), queue_tags:List(str), save_path:str): super().__init__() self.queue_list = queue_list self.queue_tags = queue_tags self.queue_num = 0 self.save_path = save_path self.is_stop = False assert len(self.queue_list) == len(self.queue_tags) def stop(self): """stop the thread""" self.is_stop = True def run(self) -> None: # creating a path for saving data time_tag = datetime.now().strftime('%Y-%m-%d-%H-%M-%S') self.save_root_folder = os.path.join(self.save_path,time_tag) os.makedirs(self.save_root_folder) self.logsmsg_folder = os.path.join(self.save_root_folder,"LogsMsg") self.imagesmsg_folder = os.path.join(self.save_root_folder,"ImagesMsg") self.plysmsg_folder = os.path.join(self.save_root_folder,"PlysMsg") self.commandsmsg_folder = os.path.join(self.save_root_folder,"CommandsMsg") sub_folder_list = [self.logsmsg_folder,self.imagesmsg_folder, self.plysmsg_folder,self.commandsmsg_folder] for folder in sub_folder_list: os.mkdir(folder) print(f"[NOTICE] Creating folder: {folder}") # process the queue list self.queue_num = len(self.queue_list) while self.queue_num: i = 0 while True: # select a queue i_tag = i % self.queue_num queue = self.queue_list[i_tag] # make sure the queue is not empty while not queue.empty(): # decode the msg using cooresponding method queue_tag = self.queue_tags[i_tag] decode_method = self.decode_select(queue_tag) decode_method(queue) # next queue i += 1 def decode_select(self,queue_tags:str): route_dic = { "Logs":self.decode_LogsMsg, "Images":self.decode_ImagesMsg, "Plys":self.decode_PlysMsg, "Commands":self.decode_CommandsMsg } return route_dic[queue_tags] def decode_LogsMsg(self,queue): nums_limit = 10 count = 0 while not queue.empty() and count < nums_limit: # extracting the msg from queue msg = queue.pop() msg_pb = pb.LogsMessage() msg_pb.ParseFromString(msg) # save the raw pb msg raw_file_name = os.path.join(self.logsmsg_folder,f"{curr_time}.LogsMsg") with open(raw_file_name, 'wb') as file: file.write(msg) # process the msg msg_source = msg_pb.source msg_timestamp = msg_pb.timestamp msg_context = msg_pb.context # display the info to the console curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S') print(f"[x] {curr_time}|(From {msg_source}): {msg_context}" ) # next msg count += 1 def decode_ImagesMsg(self,queue): while not queue.empty(): # extracting the msg from queue msg = queue.pop() msg_pb = pb.ImagesMessage() msg_pb.ParseFromString(msg) # save the raw pb msg raw_file_name = os.path.join(self.imagesmsg_folder,f"{curr_time}.ImagesMsg") with open(raw_file_name, 'wb') as file: file.write(msg) # process the msg msg_source = msg_pb.source msg_timestamp = msg_pb.timestamp # process the images id = 0 for image in msg_pb.Image: imagelegth = image.length image_byte = image.data # save the images if (image_byte is None) | (len(image_byte) == 0): pass else: filename = f"{msg_source}_{msg_timestamp}_image{id}.jpg" filepath = os.path.join(self.imagesmsg_folder,filename) with open(filepath, 'wb') as f: f.write(image) # display the info to the console curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S') print(f"[x] {curr_time}|(From {msg_source}): image{id} saved!" ) # next image item id += 1 def decode_PlysMsg(self,queue): pass def decode_CommandsMsg(self,queue): pass # 根据rabbitmq重新设置消息参数 class device_state_senderThread(senderThread): def __init__(self, MQparameters, input_msg_queue, sender_id, ): super(device_state_senderThread,self).__init__( MQparameters, \ input_msg_queue, sender_id, "device_state", "fanout" ) # 根据rabbitmq重新设置消息参数 class device_state_receiverThread(receiverThread): def __init__(self, MQparameters, receiver_id): super(device_state_receiverThread,self).__init__( MQparameters, None, None, "device_state", "fanout", receiver_id ) def callback(self, ch, method, properties, body): response_tag = f"{time.asctime()} received: {body}." print(response_tag) ch.basic_ack(delivery_tag = method.delivery_tag) # 根据rabbitmq重新设置消息参数 class alarm_state_senderThread(senderThread): def __init__(self, MQparameters, input_msg_queue, sender_id, ): super(alarm_state_senderThread,self).__init__( MQparameters, \ input_msg_queue, sender_id, "alarm_state", "fanout" ) # 根据rabbitmq重新设置消息参数 class alarm_state_receiverThread(receiverThread): def __init__(self, MQparameters, receiver_id): super(alarm_state_receiverThread,self).__init__( MQparameters, None, None, "alarm_state", "fanout", receiver_id ) def callback(self, ch, method, properties, body): response_tag = f"{time.asctime()} received: {body}." print(response_tag) ch.basic_ack(delivery_tag = method.delivery_tag) # 根据rabbitmq重新设置消息参数 class command_request_senderThread(senderThread): def __init__(self, MQparameters, input_msg_queue, sender_id, ): super(command_request_senderThread,self).__init__( MQparameters, \ input_msg_queue, sender_id, "command_request", "topic" ) # 根据rabbitmq重新设置消息参数 class command_request_receiverThread(receiverThread): def __init__(self, MQparameters, received_msg_queue, response_msg_queue, receiver_id): super(command_request_receiverThread,self).__init__( MQparameters, received_msg_queue, response_msg_queue, "command_request", "topic", receiver_id ) def callback(self, ch, method, properties, body): received_msg = Message(None,None,body,None) self.queue.put(received_msg) response_tag = f"{time.asctime()},{self.receiver_id} received {self.exchangeName} message." print(response_tag) self.response_queue.put(Message(self.receiver_id,self.receiver_id,response_tag,'')) ch.basic_ack(delivery_tag = method.delivery_tag) # 根据rabbitmq重新设置消息参数 class command_response_senderThread(senderThread): def __init__(self, MQparameters, input_msg_queue, sender_id, ): super(command_response_senderThread,self).__init__( MQparameters, \ input_msg_queue, sender_id, "command_response", "fanout" ) # 根据rabbitmq重新设置消息参数 class command_response_receiverThread(receiverThread): def __init__(self, MQparameters, receiver_id): super(command_response_receiverThread,self).__init__( MQparameters, None, None, "command_response", "fanout", receiver_id ) def callback(self, ch, method, properties, body): response_tag = f"{time.asctime()} received: {body}." print(response_tag) ch.basic_ack(delivery_tag = method.delivery_tag) # TODO class image_data_senderThread(senderThread): def __init__(self, MQparameters, input_msg_queue, sender_id, ): super(command_response_senderThread,self).__init__( MQparameters, \ input_msg_queue, sender_id, "image_data", "fanout" ) class image_data_receiverThread(receiverThread): pass def message_sender(MQparams: pika.ConnectionParameters, exchange_name: str, exchange_type:str, routing_key:str, message:str, ): # connect to RabbitMQ connection = pika.BlockingConnection(MQparams) # declare a exchange channel = connection.channel() channel.exchange_declare(exchange=exchange_name, exchange_type=exchange_type) # publish a PERSISTENT message channel.basic_publish( exchange=exchange_name, routing_key=routing_key, body=message, properties=pika.BasicProperties( delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE ) ) # close the connection connection.close() def message_receiver(MQparams: pika.ConnectionParameters, exchange_name: str, exchange_type:str, queue_name:str, binding_keys:List[str], callback_fn:callable ): # connect to RabbitMQ connection = pika.BlockingConnection(MQparams) # declare a exchange channel = connection.channel() channel.exchange_declare(exchange=exchange_name, exchange_type=exchange_type) # declare a durable queue channel.queue_declare(queue=queue_name,durable=True) # literate the binding keys and bind the queue for binding_key in binding_keys: channel.queue_bind(queue=queue_name,exchange=exchange_name, routing_key=binding_key) # waiting for consuming the messages channel.basic_qos(prefetch_count=1) channel.basic_consume(queue=queue_name, auto_ack=False, on_message_callback=callback_fn ) channel.start_consuming() # class device_state_up_msg_sender_thread(threading.Thread): # def __init__(self, MQparameters,device_state_msg_queue): # super().__init__(self) # self.parameters = MQparameters # self.queue = device_state_msg_queue # self.connection = pika.BlockingConnection(self.parameters) # def stop(self): # """stop the thread""" # self.is_stop = True # self.connection.close() # def run(self): # while not self.is_stop: # print("device_state_up_msg_sender_thread started") # channel = self.connection.channel() # channel.exchange_declare(exchange='device_state', exchange_type='fanout') # while True: # try: # if not self.queue.empty(): # msg = self.msg_queue.get() # channel.basic_publish(exchange='device_state', routing_key='', body=msg) # print(time.asctime(), "send device state message") # elif self.queue.empty(): # time.sleep(1) # except Exception as e: # break # class device_state_up_msg_receiver_thread(threading.Thread): # def __init__(self, MQparameters): # super().__init__(self) # self.parameters = MQparameters # self.connection = pika.BlockingConnection(self.parameters) # def stop(self): # """stop the thread""" # self.is_stop = True # self.connection.close() # def run(self): # while not self.is_stop: # print("device_state_up_msg_receiver_thread started") # channel = self.connection.channel() # channel.exchange_declare(exchange='device_state', exchange_type='fanout') # result = channel.queue_declare(queue='', exclusive=True, durable=True) # queue_name = result.method.queue # channel.queue_bind(exchange='device_state', queue=queue_name) # def callback(ch, method, properties, body): # print(f"Received: {body}") # ch.basic_ack(delivery_tag = method.delivery_tag) # channel.basic_consume(queue=queue_name, \ # on_message_callback=callback) # channel.basic_qos(prefetch_count=1) # channel.start_consuming() # class alarm_state_up_msg_sender_thread(threading.Thread): # def __init__(self, MQparameters,alarm_state_msg_queue): # super().__init__(self) # self.parameters = MQparameters # self.queue = alarm_state_msg_queue # self.connection = pika.BlockingConnection(self.parameters) # def stop(self): # """stop the thread""" # self.is_stop = True # self.connection.close() # def run(self): # while not self.is_stop: # print("alarm_state_up_msg_sender_thread started") # channel = self.connection.channel() # channel.exchange_declare(exchange='alarm_state', exchange_type='fanout') # while True: # try: # if not self.queue.empty(): # msg = self.msg_queue.get() # channel.basic_publish(exchange='alarm_state', routing_key='', body=msg) # print(time.asctime(), "send alarm state message") # elif self.queue.empty(): # time.sleep(1) # except Exception as e: # break # class alarm_state_up_msg_receiver_thread(threading.Thread): # def __init__(self, MQparameters): # super().__init__(self) # self.parameters = MQparameters # self.connection = pika.BlockingConnection(self.parameters) # def stop(self): # """stop the thread""" # self.is_stop = True # self.connection.close() # def run(self): # while not self.is_stop: # print("alarm_state_up_msg_receiver_thread started") # channel = self.connection.channel() # channel.exchange_declare(exchange='alarm_state', exchange_type='fanout') # result = channel.queue_declare(queue='', exclusive=True, durable=True) # queue_name = result.method.queue # channel.queue_bind(exchange='alarm_state', queue=queue_name) # def callback(ch, method, properties, body): # print(f"Received: {body}") # ch.basic_ack(delivery_tag = method.delivery_tag) # channel.basic_consume(queue=queue_name, \ # on_message_callback=callback) # channel.basic_qos(prefetch_count=1) # channel.start_consuming() # class command_request_msg_sender_thread(threading.Thread): # def __init__(self, MQparameters,command_request_msg_queue,routing_key_queue): # super().__init__(self) # self.parameters = MQparameters # self.command_queue = command_request_msg_queue # self.key_queue = routing_key_queue # self.connection = pika.BlockingConnection(self.parameters) # def stop(self): # """stop the thread""" # self.is_stop = True # self.connection.close() # def run(self): # while not self.is_stop: # print("command_request_msg_sender_thread started") # channel = self.connection.channel() # channel.exchange_declare(exchange='command_msg', exchange_type='topic') # while True: # try: # if not self.command_queue.empty() and \ # not self.key_queue.empty(): # msg = self.command_queue.get() # routing_key = self.key_queue.get() # corr_id = str(uuid.uuid4()) # channel.basic_publish(exchange='command_msg', # routing_key = routing_key, # body=msg, # properties=pika.BasicProperties( # correlation_id=corr_id # ) # ) # print(time.asctime(), f"send command request with ID:[{corr_id}]") # elif self.queue.empty(): # time.sleep(1) # except Exception as e: # break # class command_request_msg_receiver_thread(threading.Thread): # def __init__(self, MQparameters,command_request_msg_queue, \ # command_response_msg_queue, routing_key): # super().__init__(self) # self.parameters = MQparameters # self.queue = command_request_msg_queue # self.response_queue = command_response_msg_queue # self.routing_key = routing_key # self.connection = pika.BlockingConnection(self.parameters) # def stop(self): # """stop the thread""" # self.is_stop = True # self.connection.close() # def run(self): # while not self.is_stop: # print("command_request_msg_receiver_thread started") # channel = self.connection.channel() # channel.exchange_declare(exchange='command_msg', exchange_type='topic') # result = channel.queue_declare(queue='', exclusive=True, durable=True) # queue_name = result.method.queue # channel.queue_bind(exchange='command_msg', queue=queue_name, \ # routing_key=self.routing_key) # def callback(ch, method, prop, body): # print(time.asctime(), f"reveive command request with ID:[{prop.corr_id}]") # self.queue.put(body) # self.response_queue.put("") # ch.basic_ack(delivery_tag = method.delivery_tag) # channel.basic_consume(queue=queue_name, \ # on_message_callback=callback) # channel.basic_qos(prefetch_count=1) # channel.start_consuming() # class command_response_msg_sender_thread(threading.Thread): # def __init__(self, MQparameters,command_request_msg_queue): # super().__init__(self) # self.parameters = MQparameters # self.queue = command_request_msg_queue # self.connection = pika.BlockingConnection(self.parameters) # def stop(self): # """stop the thread""" # self.is_stop = True # self.connection.close() # def run(self): # while not self.is_stop: # print("command_response_msg_sender_thread started") # channel = self.connection.channel() # channel.exchange_declare(exchange='command_response', exchange_type='fanout') # while True: # try: # if not self.queue.empty(): # msg = self.msg_queue.get() # channel.basic_publish(exchange='command_response', routing_key='', body=msg) # print(time.asctime(), "send device state message") # elif self.queue.empty(): # time.sleep(1) # except Exception as e: # break # class command_down_rpc_sender_thread(threading.Thread): # def __init__(self, MQparameters, command_queue, routing_key_queue): # super().__init__(self) # self.parameters = MQparameters # self.command_queue = command_queue # self.key_queue = routing_key_queue # self.connection = pika.BlockingConnection(self.parameters) # def stop(self): # """stop the thread""" # self.is_stop = True # self.connection.close() # def run(self): # while not self.is_stop: # # print("command_down_rpc_sender_thread started") # channel = self.connection.channel() # # declare a exchange # channel.exchange_declare(exchange="command_rpc", \ # exchange_type='topic') # # declare a callback queue # result = channel.queue_declare(queue='', exclusive=True) # callback_queue = result.method.queue # # bingding the queue # channel.queue_bind( # exchange="command_rpc", # queue=callback_queue, # routing_key=routing_key # ) # # init the command uuid # corr_id = str(uuid.uuid4()) # command_response = None # # def a function to print the command respond message # def respond_callback(ch, method, properties, body): # if properties.correlation_id == corr_id: # command_response = body # print(body) # # start the cosumer # channel.basic_consume( # queue=callback_queue, # on_message_callback=respond_callback, # auto_ack=True # ) # # publish with tropic exchange # channel.basic_publish( # exchange="command_rpc", # routing_key=routing_key, # body = request_command, # properties=pika.BasicProperties( # reply_to=callback_queue, # correlation_id=corr_id # ), # ) # with command_response is None: # self.connection.process_data_events() # class command_down_rpc_receiver_thread(threading.Thread): # def __init__(self, MQparameters): # super().__init__(self) # self.parameters = MQparameters # self.connection = pika.BlockingConnection(self.parameters) # def stop(self): # """stop the thread""" # self.is_stop = True # self.connection.close() # def run(self,commands_queue): # while not self.is_stop: # print("command_down_rpc_receiver_thread started") # channel = self.connection.channel() # # declare a exchange # channel.exchange_declare(exchange="command_rpc", \ # exchange_type='topic') # # declare a callback queue # result = channel.queue_declare(queue='', exclusive=True) # callback_queue = result.method.queue # # bingding the queue # channel.queue_bind( # exchange="command_rpc", # queue=callback_queue, # routing_key='pod*' # ) # # def a function to reply the request from the sender # def on_request(ch, method, prop, body): # commands_queue.put(body) # response = f"{time.asctime()} {self.device_id} receive the command uuid:[{prop.correlation_id}]" # print(response) # ch.basic_publish(exchange="command_rpc", # routing_key = prop.reply_to, # properties = pika.BasicProperties( # correlation_id=prop.corr_id # ), # body = response) # ch.basic_ack(delivery_tag = method.delivery_tag) # # start the cosumer # channel.basic_qos(prefetch_count=1) # channel.basic_consume( # queue=callback_queue, # on_message_callback=on_request, # auto_ack=True # ) # channel.start_consuming() # class image_data_up_msg_sender_thread(threading.Thread): # pass # class image_data_up_msg_receiver_thread(threading.Thread): # pass # class ply_data_up_msg_sender_thread(threading.Thread): # pass # class ply_data_up_msg_receiver_thread(threading.Thread): # pass