From e839f2e83fed7eaceb677242f5f9650a79fb8c43 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期五, 04 三月 2022 14:16:57 +0800
Subject: [PATCH] temp commit
---
src/engine/AcquisitionMain.py | 539 ++++++++++++++++++++++++++++++++++++++++-------------------
1 files changed, 363 insertions(+), 176 deletions(-)
diff --git a/src/engine/AcquisitionMain.py b/src/engine/AcquisitionMain.py
index 4373fa6..dfdbe31 100644
--- a/src/engine/AcquisitionMain.py
+++ b/src/engine/AcquisitionMain.py
@@ -11,6 +11,8 @@
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))
@@ -60,198 +62,383 @@
f.close()
return
-class AcquisitionMain:
+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,
+ # 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 __init__(self, MQhost, MQuser, MQpassword, deviceID, MICROWAVEhost):
- 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()
+ # 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()
- def _stop_capture_thread(self):
- """stop camera capture thread"""
- if not self._capture_thread is None:
- self._capture_thread.stop()
- self._capture_thread = None
+ 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 _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)
+ def _connect_to_MQ(self):
+ credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
+ self.parameters = pika.ConnectionParameters(
+ credentials=credentials, host=self.MQhost)
# 杩炴帴闃熷垪鏈嶅姟鍣�
- self._connection = pika.BlockingConnection(parameters)
- self.channel = self._connection.channel()
- self.channel.queue_declare(queue='', exclusive=True)
+ 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 _close_message_queue(self) -> None:
- """close message queue of rabbitmq"""
- self._connection.close()
+ 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 _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 _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 _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 _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 _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 _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 _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 _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 _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 _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 _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_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()
+ pass
+
+ 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 start(self):
+ self.run_all()
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ # 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()
+ # def start(self) -> None:
+ # """start all threads"""
+ # self._start_controll_thread()
+ # self._start_capture_thread()
+ # self._start_udp_thread()
+ # self._main_loop()
--
Gitblit v1.9.1