import sys # NOQA: E402
|
import os
|
import cv2
|
import logging
|
import json
|
import pika
|
import queue
|
import requests
|
import time
|
|
from .CaptureThread import CaptureThread
|
from .ControllThread import ControllThread
|
from .MicroWaveUDPThread import MicroWaveUDPThread
|
|
# 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:
|
"""
|
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 _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()
|
