From 0dc337197311d0fec373bf479f8ee0ac26b4cad6 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期三, 15 六月 2022 16:33:47 +0800
Subject: [PATCH] client enhanced&camera contorller
---
src/engine/AcquisitionMain.py | 317 +++++++++++++++++++++-------------------------------
1 files changed, 127 insertions(+), 190 deletions(-)
diff --git a/src/engine/AcquisitionMain.py b/src/engine/AcquisitionMain.py
index bb2c8be..17d80ed 100644
--- a/src/engine/AcquisitionMain.py
+++ b/src/engine/AcquisitionMain.py
@@ -1,112 +1,71 @@
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 .CaptureThread import CaptureThread
+# from .ControllThread import ControllThread
+# from .MicroWaveUDPThread import MicroWaveUDPThread
from . import MessageThreads as ths
-from .devices import cameraController
+from .devices import cameraController, dustCleanerController
# # 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
+# 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'
+# 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
+# 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']
+# # 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
+# 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):
+ def __init__(self,
+ MQhost:str,
+ MQuser:str,
+ MQpassword:str,
+ deviceID:str,
+ MICROWAVEhost:str):
# device ID
self.device_id = deviceID
@@ -122,28 +81,14 @@
# 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.threads = []
- 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
- }
+ # 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
@@ -159,6 +104,7 @@
}
+
def _connect_to_MQ(self):
credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
self.parameters = pika.ConnectionParameters(
@@ -169,100 +115,92 @@
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")
+ 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 _receiver_Thread_Launcher(self):
- 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
+ # launch a commands receiver thread
+ self.CommandsReceiver_Thread = ths.receiverThread(
+ MQparams=self.parameters,
+ exchange_name='commands_topic',
+ exchange_type='topic',
+ MQqueue_name="commands_queue",
+ binding_keys=[f"{self.device_id}"],
+ message_queue=self.CommandsReceiver_queue
+ )
+ self.CommandsReceiver_Thread.setDaemon(True)
+ self.CommandsReceiver_Thread.setName("commands_thread")
+ self.CommandsReceiver_Thread.start()
- 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()
+ self.threads.append(self.CommandsReceiver_Thread)
+
+ def _sender_Thread_Launcher(self):
- 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
+ # 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("logs_thread")
+ self.LogsSender_Thread.start()
+
+ self.threads.append(self.LogsSender_Thread)
- 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()
+ # 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("image_thread")
+ self.ImagesSender_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
+ 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("plys_thread")
+ self.PlysSender_Thread.start()
+
+ self.threads.append(self.PlysSender_Thread)
- 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 _connect_device(self):
+ # connect to cameras
+ self.cc = cameraController(
+ IMAGE_WIDTH=IMAGE_WIDTH,
+ IMAGE_HEIGHT=IMAGE_HEIGHT,
+ CAMERA_COUNT=CAMERA_COUNT,
+ device_id=self.device_id,
+ logs_queue=self.LogsSender_queue,
+ images_queue=self.ImagesSender_queue,
+ init_binding_key="client.client.client",
+ init_routing_key="server"
+ )
+ # self.dc = dustCleanerController()
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()
+ self._receiver_Thread_Launcher()
+ self._sender_Thread_Launcher()
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
-
+ self._connect_device()
def run_all(self):
@@ -273,21 +211,20 @@
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()
+ for t in self.threads:
+ t.stop()
+ print(f"stop {t.name}")
def start(self):
try:
self.run_all()
- self._device_state_sender_thread.join()
+ print("[*] Client is runing ...")
+ self.CommandsReceiver_Thread.join()
except Exception as e:
+ print(e)
print("running error, stop all threads")
-
- self._stop_alarm_state_up_msg_sender_thread()
- self.stop
+ self.stop_all()
print("main thread end.")
return 0
--
Gitblit v1.9.1