import pika
|
import threading
|
from ctypes import sizeof
|
from .DeviceManager import DeviceManager
|
|
# # 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 ..microwave.components import HeaderBlock
|
from ..protos import aerial_pb2 as pb
|
|
class MessageQueueThread(threading.Thread):
|
def __init__(self,
|
host: str,
|
user: str,
|
password: str,
|
queue: str,
|
device_manager: DeviceManager,
|
save_folder: str):
|
threading.Thread.__init__(self)
|
self._device_manager = device_manager
|
self.save_folder = save_folder
|
self._def_routes()
|
self.message_received: int = 0
|
|
credentials = pika.PlainCredentials(user, password)
|
parameters = pika.ConnectionParameters(
|
credentials=credentials, host=host)
|
connection = pika.BlockingConnection(parameters)
|
self._channel = connection.channel()
|
self._channel.queue_declare(queue=queue, durable=True)
|
self._channel.basic_qos(prefetch_count=1)
|
self._channel.basic_consume(
|
on_message_callback=self._on_request, queue=queue)
|
|
def _def_routes(self):
|
self._routes = {
|
pb.SysCommand.FIVE_G_COMMAND: self._on_five_g_command,
|
pb.SysCommand.MICRO_WAVE_COMMAND: self._on_micro_wave_command,
|
}
|
|
def _default(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
|
print('onknown command')
|
|
def _write_image(self, device: dict, idx: int, image: bytes):
|
if (image is None) | (len(image) == 0):
|
return
|
filename = f'{device["path"]}/{idx}-{device["images"]}.jpg'
|
with open(filename, 'wb') as f:
|
f.write(image)
|
f.close()
|
|
def _on_five_g_command(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
|
# print('five g command')
|
if pb.SubCommand.FIVE_G_SEND_IMAGE_SUB_COMMAND == req.sub_command:
|
print(
|
f'send image sub command received, length={req.data1.length}')
|
if req.data1.length > 0:
|
device = self._device_manager.get_device(req.device_id, self.save_folder)
|
img_msg = pb.ImagesMessage()
|
img_msg.ParseFromString(req.data1.data)
|
# print(img_msg.image1)
|
count = device["images"]
|
self._write_image(device, 1, img_msg.image1)
|
self._write_image(device, 2, img_msg.image2)
|
self._write_image(device, 3, img_msg.image3)
|
self._write_image(device, 4, img_msg.image4)
|
self._write_image(device, 5, img_msg.image5)
|
device['images'] += 1
|
|
def _on_micro_wave_command(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
|
print('micro wave command')
|
if pb.SubCommand.MICRO_WAVE_SUB_COMMAND == req.sub_command:
|
print('micro wave sub command received')
|
device = self._device_manager.get_device(req.device_id,self.save_folder)
|
micro_wave_msg = pb.MicroWaveMessage()
|
micro_wave_msg.ParseFromString(req.data1.data)
|
if (micro_wave_msg.length == len(micro_wave_msg.data)):
|
header = HeaderBlock.from_buffer_copy(micro_wave_msg[:sizeof(HeaderBlock)])
|
filename = f'{device["path"]}/{device["micro_wave"]}-{header.FrameNumber}.micro_wave'
|
with open(filename, 'wb') as f:
|
f.write(micro_wave_msg.data)
|
f.close()
|
device['micro_wave'] += 1
|
|
def _on_request(self, ch, method, properties, body):
|
req = pb.RequestCommand()
|
req.ParseFromString(body)
|
dispatcher = self._routes.get(req.sys_command, self._default)
|
resp = pb.ResponseCommand()
|
dispatcher(req, resp)
|
resp.checksum = 12345
|
resp.state = pb.ResponseState.SUCCESS
|
if properties.reply_to != None:
|
ch.basic_publish(exchange='',
|
routing_key=properties.reply_to, # 收消息的队列
|
properties=pika.BasicProperties(
|
correlation_id=properties.correlation_id), # 返回消息的队列
|
body=resp.SerializeToString()) # 返回结果数据
|
ch.basic_ack(delivery_tag=method.delivery_tag) # 确保消息被 客户端接收
|
self.message_received += 1
|
return
|
|
def run(self):
|
print(' [*] RabbitMQ Server started. Waiting for messages.')
|
self._channel.start_consuming()
|
