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()
|
