import sys # NOQA: E402
|
import os
|
import cv2
|
import logging
|
import json
|
import pika
|
import queue
|
import requests
|
import time
|
import threading
|
|
from .CaptureThread import CaptureThread
|
from .ControllThread import ControllThread
|
from .MicroWaveUDPThread import MicroWaveUDPThread
|
from . import MessageThreads as ths
|
from .CameraDevice import cameraController
|
|
# 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):
|
"""
|
acquisition main class
|
"""
|
# def __init__(self,
|
# rabbitmq_host: str,
|
# rabbitmq_user: str,
|
# rabbitmq_password: str,
|
# rabbitmq_queue: str,
|
# device_id: str,
|
# api_url: str,
|
# udp_server_host: str,
|
# config: Config) -> None:
|
# """
|
# init AcquisitionMain class
|
# Args:
|
# rabbitmq_host: rabbitmq server host
|
# rabbitmq_user: rabbitmq user
|
# rabbitmq_password: rabbitmq password
|
# rabbitmq_queue: rabbitmq upload queue
|
# device_id: device id
|
# api_url: server api url
|
# udp_server_host: udp server host for micro wave, localhost for testing or 192.168.2.65 for production
|
# config: config object
|
# """
|
# self.rabbitmq_host: str = rabbitmq_host
|
# self.udp_server_host: str = udp_server_host
|
# self.rabbitmq_user: str = rabbitmq_user
|
# self.rabbitmq_password: str = rabbitmq_password
|
# self.rabbitmq_queue: str = rabbitmq_queue
|
# self.device_id: str = device_id
|
# self.message_send: int = 0
|
# self.api_url: str = api_url
|
# self.config = config
|
# self.image_quality: float = 95
|
# self.cameras: list = []
|
# self.msg_queue = queue.Queue()
|
# # thread members
|
# self._controll_thread = None
|
# self._capture_thread = None
|
# self._udp_thread = None
|
|
|
def __init__(self, MQhost, MQuser, MQpassword, deviceID, MICROWAVEhost):
|
|
# device ID
|
self.device_id = deviceID
|
# rabbit MQ
|
self.MQhost = MQhost
|
self.MQuser = MQuser
|
self.MQpassword = MQpassword
|
self.parameters = None
|
# camera
|
self.IMAGE_WIDTH = 3840
|
self.IMAGE_HEIGHT = 2880
|
self.CAMERA_COUNT = 4
|
# microwave
|
self.MICROWAVEhost = MICROWAVEhost
|
# threads
|
self._device_state_sender_thread = None
|
self._alarm_state_sender_thread = None
|
self._command_request_receiver_thread = None
|
self._command_response_sender_thread = None
|
# queue
|
self.device_state_queue = queue.Queue()
|
self.alarm_state_queue = queue.Queue()
|
self.command_request_queue = queue.Queue()
|
self.command_response_queue = queue.Queue()
|
self.image_data_queue = queue.Queue()
|
self.udp_data_queue = queue.Queue()
|
self.ply_data_queue = queue.Queue()
|
|
self.queue_dict = {
|
"device_state":self.device_state_queue,
|
"alarm_state":self.alarm_state_queue,
|
"command_request":self.command_request_queue,
|
"command_response":self.command_response_queue,
|
"image_data":self.image_data_queue,
|
"udp_data":self.udp_data_queue,
|
"ply_data":self.ply_data_queue
|
}
|
|
|
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 _start_device_state_up_msg_sender_thread(self):
|
"""start device state thread"""
|
if self._device_state_sender_thread is None:
|
self._device_state_sender_thread = \
|
ths.device_state_senderThread(self.parameters,\
|
self.device_state_queue,self.device_id)
|
self._device_state_sender_thread.setDaemon(True)
|
self._device_state_sender_thread.start()
|
|
def _stop_device_state_up_msg_sender_thread(self):
|
"""stop device state thread"""
|
if self._device_state_sender_thread is not None:
|
self._device_state_sender_thread.stop()
|
self._device_state_sender_thread = None
|
|
def _start_alarm_state_up_msg_sender_thread(self):
|
"""start alarm state thread"""
|
if self._alarm_state_sender_thread is None:
|
self._alarm_state_sender_thread = \
|
ths.alarm_state_senderThread(self.parameters,\
|
self.alarm_state_queue,self.device_id)
|
self._alarm_state_sender_thread.setDaemon(True)
|
self._alarm_state_sender_thread.start()
|
|
def _stop_alarm_state_up_msg_sender_thread(self):
|
"""stop alarm state thread"""
|
if self._alarm_state_sender_thread is not None:
|
self._alarm_state_sender_thread.stop()
|
self._alarm_state_sender_thread = None
|
|
|
def _start_command_response_sender_thread(self):
|
"""start command response thread"""
|
if self._command_response_sender_thread is None:
|
self._command_response_sender_thread = \
|
ths.command_response_senderThread(self.parameters, \
|
self.command_response_queue,self.device_id)
|
self._command_response_sender_thread.setDaemon(True)
|
self._command_response_sender_thread.start()
|
|
def _stop_command_response_sender_thread(self):
|
"""stop command response thread"""
|
if self._command_response_sender_thread is not None:
|
self._command_response_sender_thread.stop()
|
self._command_response_sender_thread = None
|
|
|
def _start_command_request_receiver_thread(self):
|
"""start command request thread"""
|
if self._command_request_receiver_thread is None:
|
self._command_request_receiver_thread = \
|
ths.command_request_receiverThread(self.parameters,\
|
self.command_request_queue,self.command_response_queue,self.device_id)
|
self._command_request_receiver_thread.setDaemon(True)
|
self._command_request_receiver_thread.start()
|
|
def _stop_command_request_receiver_thread(self):
|
"""stop command request thread"""
|
if self._command_request_receiver_thread is not None:
|
self._command_request_receiver_thread.stop()
|
self._command_request_receiver_thread = None
|
|
|
def run_rabbitmq(self):
|
self._connect_to_MQ()
|
self._start_device_state_up_msg_sender_thread()
|
self._start_alarm_state_up_msg_sender_thread()
|
self._start_command_response_sender_thread()
|
self._start_command_request_receiver_thread()
|
|
def run_device(self):
|
self.connecting_camera()
|
self.connecting_microwave()
|
self.connecting_dustCleaner()
|
|
def connecting_camera(self):
|
# init the camera controller
|
cc = cameraController(self.IMAGE_WIDTH,self.IMAGE_HEIGHT,\
|
self.CAMERA_COUNT,self.device_id,self.queue_dict)
|
# connecting to camera via opencv
|
cc.open_cameras()
|
# capture the data, and wrap it to data queue
|
# cc._start_captureThread()
|
|
|
def connecting_microwave(self):
|
pass
|
|
def connecting_dustCleaner(self):
|
pass
|
|
|
|
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):
|
self._stop_alarm_state_up_msg_sender_thread()
|
self._stop_device_state_up_msg_sender_thread()
|
self._stop_command_request_receiver_thread()
|
self._start_command_response_sender_thread()
|
|
|
def start(self):
|
try:
|
self.run_all()
|
self._device_state_sender_thread.join()
|
except Exception as e:
|
print("running error, stop all threads")
|
|
self._stop_alarm_state_up_msg_sender_thread()
|
self.stop
|
print("main thread end.")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 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()
|
