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.backup | 257 ++++++++
TEST_startServer.bat | 9
.vscode/launch.json | 4
src/Client.py | 6
src/Server.py | 164 +++-
src/Server.py.backup | 115 +++
src/engine/CameraDevice.py | 121 +++
src/engine/__init__.py | 4
src/engine/MessageThreads.py | 650 ++++++++++++++++++++
src/utils/tools.py | 2
src/engine/AcquisitionMain.py | 539 +++++++++++-----
TEST_startClient.bat | 2
12 files changed, 1,632 insertions(+), 241 deletions(-)
diff --git a/.vscode/launch.json b/.vscode/launch.json
index c0e1721..df02b89 100644
--- a/.vscode/launch.json
+++ b/.vscode/launch.json
@@ -48,9 +48,9 @@
"localhost",
"chai",
"password123",
- "aerial_rpc",
+ "0.0.0.0",
"5000",
- "./Images"
+ "./Received_data_test"
],
"console": "integratedTerminal"
},
diff --git a/TEST_startClient.bat b/TEST_startClient.bat
index 69df83a..b584100 100644
--- a/TEST_startClient.bat
+++ b/TEST_startClient.bat
@@ -12,7 +12,7 @@
call %envsetting%
)
-set MQhost=192.168.1.224
+set MQhost=192.168.1.14
set MQuser=chai
set MQpassword=password123
set QUEUEname=aerial_rpc
diff --git a/TEST_startServer.bat b/TEST_startServer.bat
index 30abc3a..b782d6d 100644
--- a/TEST_startServer.bat
+++ b/TEST_startServer.bat
@@ -17,17 +17,16 @@
set MQhost=localhost
set MQuser=chai
set MQpassword=password123
-set QUEUEname=aerial_rpc
+set WEBhost=0.0.0.0
set WEBport=5000
-set SAVEfolder=./ImagesTEST
+set SAVEfolder=./Received_data_testing
set LOGfile=ServerTEST.log
@REM Launch the Server
-@REM -------- Usage: Server.py [MQhost] [MQuser] [MQpassword] [QUEUEname] [WEBport] [SAVEfolder]")
echo the log file will be saved in the [%LOGfile%] file
echo Server.py is running ...
-@REM -------- Usage: Server.py [MQhost] [MQuser] [MQpassword] [QUEUEname] [WEBport] [SAVEfolder]")
-%aerialpython% ./src/Server.py %MQhost% %MQuser% %MQpassword% %QUEUEname% %WEBport% %SAVEfolder% >> %LOGfile%
+@REM -------- Usage: Server.py [MQhost] [MQuser] [MQpassword] [WEBhost] [WEBport] [SAVEfolder]")
+%aerialpython% ./src/Server.py %MQhost% %MQuser% %MQpassword% %WEBhost% %WEBport% %SAVEfolder%
echo Server.py has been terminited.
pause
diff --git a/src/Client.py b/src/Client.py
index 3758f0f..2703a4f 100644
--- a/src/Client.py
+++ b/src/Client.py
@@ -44,9 +44,7 @@
client = AcquisitionMain(MQhost,
MQuser,
MQpassword,
- QUEUEname,
- DEVICEname,
- SERVERhost,
+ DEVICEname,
MICROWAVEhost, # udp server host, 192.168.2.65 for production
- config)
+ )
client.start()
diff --git a/src/Server.py b/src/Server.py
index b2e41a4..d4683e5 100644
--- a/src/Server.py
+++ b/src/Server.py
@@ -1,56 +1,135 @@
-from posixpath import abspath
import sys # NOQA: E402
import os
-from flask import Flask, request
+from flask import Flask
+from numpy import true_divide
+import pika
+import uuid
+import queue
+from werkzeug.wrappers import request
+
+from zmq import device
+
batfile_dir = os.path.join(__file__, "..", "protos/compile.bat" )
batfile_path = os.path.abspath(batfile_dir)
os.system(batfile_path)
-from engine.DeviceManager import DeviceManager
-from engine.MessageQueueThread import MessageQueueThread
from protos import aerial_pb2 as pb
+from engine import MessageThreads as ths
class MyServer(Flask):
- def __init__(self, host, user, password, queue, port):
+ def __init__(self, MQhost, MQuser, MQpassword, WEBhost, WEBport, SAVEfolder):
Flask.__init__(self, __name__)
- self.host = host
- self.user = user
- self.password = password
- self.queue = queue
- self.port = port
-
+ # rabbit MQ
+ self.MQhost = MQhost
+ self.MQuser = MQuser
+ self.MQpassword = MQpassword
+ self.parameters = None
+ # Web server
+ self.WEBhost = WEBhost
+ self.WEBport = WEBport
+ self.SAVEfolder = SAVEfolder
self.route('/api', methods=['POST'])(self.api)
self.route('/', methods=['GET'])(self.hello)
self._def_routes()
- self._device_manager = DeviceManager(host, user, password)
+ # threads
+ self._device_state_reveiver_thread = None
+ self._alarm_state_receiver_thread = None
+ self._command_request_sender_thread = None
+ self._command_response_receiver_thread = None
+ # queue
+ self.command_request_queue = queue.Queue()
+
+ 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(self.parameters)
+ print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
+
+ def _start_device_state_up_msg_receiver_thread(self):
+ """start device state thread"""
+ if self._device_state_reveiver_thread is None:
+ self._device_state_reveiver_thread = \
+ ths.device_state_receiverThread(self.parameters,"server")
+ self._device_state_reveiver_thread.setDaemon(True)
+ self._device_state_reveiver_thread.start()
+
+ def _stop_device_state_up_msg_receiver_thread(self):
+ """stop device state thread"""
+ if self._device_state_reveiver_thread is not None:
+ self._device_state_reveiver_thread.stop()
+ self._device_state_reveiver_thread = None
+
+ def _start_alarm_state_up_msg_receiver_thread(self):
+ """start alarm state thread"""
+ if self._alarm_state_receiver_thread is None:
+ self._alarm_state_receiver_thread = \
+ ths.alarm_state_receiverThread(self.parameters,"server")
+ self._alarm_state_receiver_thread.setDaemon(True)
+ self._alarm_state_receiver_thread.start()
+
+ def _stop_alarm_state_up_msg_receiver_thread(self):
+ """stop alarm state thread"""
+ if self._alarm_state_receiver_thread is not None:
+ self._alarm_state_receiver_thread.stop()
+ self._alarm_state_receiver_thread = None
+
+ def _start_command_response_receiver_thread(self):
+ """start command response thread"""
+ if self._command_response_receiver_thread is None:
+ self._command_response_receiver_thread = \
+ ths.command_response_receiverThread(self.parameters,"server")
+ self._command_response_receiver_thread.setDaemon(True)
+ self._command_response_receiver_thread.start()
+
+ def _stop_command_response_receiver_thread(self):
+ """stop command response thread"""
+ if self._command_response_receiver_thread is not None:
+ self._command_response_receiver_thread.stop()
+ self._command_response_receiver_thread = None
+
+ def _start_command_request_sender_thread(self):
+ """start command request thread"""
+ if self._command_request_sender_thread is None:
+ self._command_request_sender_thread = \
+ ths.command_request_senderThread(self.parameters,self.command_request_queue,"server")
+ self._command_request_sender_thread.setDaemon(True)
+ self._command_request_sender_thread.start()
+
+ def _stop_command_request_sender_thread(self):
+ """stop command request thread"""
+ if self._command_request_sender_thread is not None:
+ self._command_request_sender_thread.stop()
+ self._command_request_sender_thread = None
def _def_routes(self):
self.routes = {
pb.SysCommand.SYS_COMMAND: self._on_sys_command,
}
-
+
def _on_sys_command(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
- if pb.SubCommand.POWER_ON_SUB_COMMAND == req.sub_command:
- print('power on message received')
- self._device_manager.get_device(req.device_id,save_folder)
- elif pb.SubCommand.POWER_OFF_SUB_COMMAND == req.sub_command:
- device = self._device_manager.get_device(req.device_id,save_folder)
- device['power_on'] = False
- print('power off message received')
+
+ self.command_request_queue.put(ths.Message(None,None,req.command,req.target))
return
def _default(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
print('onknown command')
def api(self):
+
req = pb.RequestCommand()
req.ParseFromString(request.data)
+
+ # a function: _on_sys_command, refer to self._on_sys_command
dispatcher = self.routes.get(req.sys_command, self._default)
+
resp = pb.ResponseCommand()
- dispatcher(req, resp)
+ dispatcher(req, resp) # call the function
+
resp.checksum = 12345
response = Flask.response_class(
resp.SerializeToString(),
@@ -61,39 +140,22 @@
def hello(self):
return 'Hello, This is acquisition demo API!'
- def run_rabbitmq(self, host: str, user: str, password: str, queue: str) -> None:
- self._message_queue_thread = MessageQueueThread(
- host, user, password, queue, self._device_manager, save_folder)
- self._message_queue_thread.start()
+ def run_rabbitmq(self):
+ self._connect_to_MQ()
+ self._start_device_state_up_msg_receiver_thread()
+ self._start_alarm_state_up_msg_receiver_thread()
+ self._start_command_response_receiver_thread()
+ self._start_command_request_sender_thread()
def run_all(self):
- self.run_rabbitmq(self.host, self.user, self.password, self.queue)
- self.run(debug=False, host="0.0.0.0", port=self.port)
-
- def send_controll_message(self, device_id, msg):
- device = self._device_manager.get_device(device_id,save_folder)
- if device is None:
- channel = device.controll_channel
- channel.basic_publish(
- exchange='',
- routing_key='',
- body=msg.SerializeToString())
-
- def send_dust_cleaner_cmd(self, device_id, sub_cmd, period):
- msg = pb.ControllMessage()
- msg.sys_command = pb.ControlSysCommand.DUST_CLEANER_COMMAND
- msg.sub_command = sub_cmd
- data_msg = pb.DustCleanerMessage()
- data_msg.period = period
- data1_str = data_msg.SerializeToString()
- msg.data1.length = len(data1_str)
- msg.data1.data = data1_str
+ self.run_rabbitmq()
+ self.run(debug=False, host=self.WEBhost, port=self.WEBport)
if __name__ == "__main__":
def help():
- print("Usage: Server.py [MQhost] [MQuser] [MQpassword] [QUEUEname] [WEBport] [SAVEfolder]")
- print("EXAMPLE: Server.py localhost chai password123 aerial_rpc 5000 ./Images")
+ print("Usage: Server.py [MQhost] [MQuser] [MQpassword] [WEBhost] [WEBport] [SAVEfolder]")
+ print("EXAMPLE: Server.py localhost chai password123 0.0.0.0 5000 ./Received_data/")
exit(1)
if len(sys.argv) != 7:
@@ -102,14 +164,12 @@
MQhost = str(sys.argv[1])
MQuser = str(sys.argv[2])
MQpassword = str(sys.argv[3])
- QUEUEname = str(sys.argv[4])
+ WEBhost = str(sys.argv[4])
WEBport = int(sys.argv[5])
SAVEfolder = str(sys.argv[6])
- global save_folder
- save_folder = SAVEfolder
except Exception as e:
print(e)
help()
- server = MyServer(host="localhost", user='chai', password='password123', queue='aerial_rpc', port=5000)
+ server = MyServer(MQhost, MQuser, MQpassword, WEBhost, WEBport, SAVEfolder)
server.run_all()
diff --git a/src/Server.py.backup b/src/Server.py.backup
new file mode 100644
index 0000000..b2e41a4
--- /dev/null
+++ b/src/Server.py.backup
@@ -0,0 +1,115 @@
+from posixpath import abspath
+import sys # NOQA: E402
+import os
+from flask import Flask, request
+
+batfile_dir = os.path.join(__file__, "..", "protos/compile.bat" )
+batfile_path = os.path.abspath(batfile_dir)
+os.system(batfile_path)
+
+from engine.DeviceManager import DeviceManager
+from engine.MessageQueueThread import MessageQueueThread
+from protos import aerial_pb2 as pb
+
+
+
+class MyServer(Flask):
+ def __init__(self, host, user, password, queue, port):
+ Flask.__init__(self, __name__)
+ self.host = host
+ self.user = user
+ self.password = password
+ self.queue = queue
+ self.port = port
+
+ self.route('/api', methods=['POST'])(self.api)
+ self.route('/', methods=['GET'])(self.hello)
+ self._def_routes()
+ self._device_manager = DeviceManager(host, user, password)
+
+ def _def_routes(self):
+ self.routes = {
+ pb.SysCommand.SYS_COMMAND: self._on_sys_command,
+ }
+
+ def _on_sys_command(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
+ if pb.SubCommand.POWER_ON_SUB_COMMAND == req.sub_command:
+ print('power on message received')
+ self._device_manager.get_device(req.device_id,save_folder)
+ elif pb.SubCommand.POWER_OFF_SUB_COMMAND == req.sub_command:
+ device = self._device_manager.get_device(req.device_id,save_folder)
+ device['power_on'] = False
+ print('power off message received')
+ return
+
+ def _default(self, req: pb.RequestCommand, resp: pb.ResponseCommand):
+ print('onknown command')
+
+ def api(self):
+ req = pb.RequestCommand()
+ req.ParseFromString(request.data)
+ dispatcher = self.routes.get(req.sys_command, self._default)
+ resp = pb.ResponseCommand()
+ dispatcher(req, resp)
+ resp.checksum = 12345
+ response = Flask.response_class(
+ resp.SerializeToString(),
+ mimetype='application/x-protobuf'
+ )
+ return response, 200
+
+ def hello(self):
+ return 'Hello, This is acquisition demo API!'
+
+ def run_rabbitmq(self, host: str, user: str, password: str, queue: str) -> None:
+ self._message_queue_thread = MessageQueueThread(
+ host, user, password, queue, self._device_manager, save_folder)
+ self._message_queue_thread.start()
+
+ def run_all(self):
+ self.run_rabbitmq(self.host, self.user, self.password, self.queue)
+ self.run(debug=False, host="0.0.0.0", port=self.port)
+
+ def send_controll_message(self, device_id, msg):
+ device = self._device_manager.get_device(device_id,save_folder)
+ if device is None:
+ channel = device.controll_channel
+ channel.basic_publish(
+ exchange='',
+ routing_key='',
+ body=msg.SerializeToString())
+
+ def send_dust_cleaner_cmd(self, device_id, sub_cmd, period):
+ msg = pb.ControllMessage()
+ msg.sys_command = pb.ControlSysCommand.DUST_CLEANER_COMMAND
+ msg.sub_command = sub_cmd
+ data_msg = pb.DustCleanerMessage()
+ data_msg.period = period
+ data1_str = data_msg.SerializeToString()
+ msg.data1.length = len(data1_str)
+ msg.data1.data = data1_str
+
+
+if __name__ == "__main__":
+ def help():
+ print("Usage: Server.py [MQhost] [MQuser] [MQpassword] [QUEUEname] [WEBport] [SAVEfolder]")
+ print("EXAMPLE: Server.py localhost chai password123 aerial_rpc 5000 ./Images")
+ exit(1)
+
+ if len(sys.argv) != 7:
+ help()
+ try:
+ MQhost = str(sys.argv[1])
+ MQuser = str(sys.argv[2])
+ MQpassword = str(sys.argv[3])
+ QUEUEname = str(sys.argv[4])
+ WEBport = int(sys.argv[5])
+ SAVEfolder = str(sys.argv[6])
+ global save_folder
+ save_folder = SAVEfolder
+
+ except Exception as e:
+ print(e)
+ help()
+ server = MyServer(host="localhost", user='chai', password='password123', queue='aerial_rpc', port=5000)
+ server.run_all()
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()
diff --git a/src/engine/AcquisitionMain.py.backup b/src/engine/AcquisitionMain.py.backup
new file mode 100644
index 0000000..4373fa6
--- /dev/null
+++ b/src/engine/AcquisitionMain.py.backup
@@ -0,0 +1,257 @@
+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()
diff --git a/src/engine/CameraDevice.py b/src/engine/CameraDevice.py
new file mode 100644
index 0000000..627ebf4
--- /dev/null
+++ b/src/engine/CameraDevice.py
@@ -0,0 +1,121 @@
+import cv2
+import time
+import threading
+import numpy as np
+import os, sys
+
+# 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
+
+class cameraController(object):
+
+ def __init__(self,IMAGE_WIDTH,IMAGE_HEIGHT,CAMERA_COUNT,device_id,queue_dict):
+ self.image_width = IMAGE_WIDTH
+ self.image_height = IMAGE_HEIGHT
+ self.camera_count = CAMERA_COUNT
+ self.device_id = device_id
+ self.queue_dict = queue_dict
+ self.device_state_queue = self.queue_dict["device_state"]
+ self.image_data_queue = self.queue_dict["image_data"]
+ self._capture_thread = None
+
+ 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(self.CAMERA_COUNT):
+ camera = cv2.VideoCapture(dev_port, cv2.CAP_DSHOW)
+ camera.set(cv2.CAP_PROP_FRAME_WIDTH, self.IMAGE_WIDTH)
+ camera.set(cv2.CAP_PROP_FRAME_HEIGHT, self.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:
+ self.working_ports.append(dev_port)
+ self.cameras.append(camera)
+ msg = f"{time.asctime()} {self.device_id} camera port {dev_port} is working and reads images ({h} x {w})"
+ else:
+ msg = f"{time.asctime()} {self.device_id} camera port {dev_port} is not working"
+
+ # add msg to queue
+ self.device_state_queue.put(msg)
+ print(msg)
+
+
+ def _start_captureThread(self):
+ """start capture thread"""
+ if self._capture_thread is None:
+ self._capture_thread = \
+ CaptureThread(self.cameras,self.image_data_queue,interval=10)
+ self._capture_thread.setDaemon(True)
+ self._capture_thread.start()
+
+ def _stop_captureThread(self):
+ """stop capture thread"""
+ if self._capture_thread is not None:
+ self._capture_thread.stop()
+ self._capture_thread = None
+
+
+class CaptureThread(threading.Thread):
+ """
+ camera capture thread
+ """
+ def __init__(self, cameras, msg_queue, interval, image_quality = 95):
+ """init the thread"""
+ threading.Thread.__init__(self)
+ self.cameras = cameras
+ self.msg_queue = msg_queue
+ self.interval = interval
+ self.image_quality = image_quality
+ self.is_stop = False
+
+ def stop(self):
+ """stop the thread"""
+ self.is_stop = True
+
+ def run(self):
+ print("CaptureThread started")
+ self.next_time = time.time() + self.interval
+ while not self.is_stop:
+ this_time = time.time()
+ if this_time > self.next_time:
+ data1 = pb.ImagesMessage()
+ for i in range(len(self.cameras)):
+ _, frame = self.cameras[i].read()
+ encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), self.image_quality]
+ _, encimg = cv2.imencode('.jpg', frame, encode_param)
+ if frame is not None:
+ np_array = np.asarray(encimg) # frame 杞崲涓簄umpy鏁扮粍
+ byte_array = np_array.tobytes() # numpy鏁扮粍杞崲涓篵yte鏁扮粍
+ if i == 0:
+ data1.image1 = byte_array
+ elif i == 1:
+ data1.image2 = byte_array
+ elif i == 2:
+ data1.image3 = byte_array
+ elif i == 3:
+ data1.image4 = byte_array
+ data1_str = data1.SerializeToString()
+
+ # create protobuf message
+ req = pb.RequestCommand()
+ req.sys_command = pb.SysCommand.FIVE_G_COMMAND
+ req.sub_command = pb.SubCommand.FIVE_G_SEND_IMAGE_SUB_COMMAND
+ req.data1.length = len(data1_str)
+ req.data1.data = data1_str
+ # add message to queue
+ self.msg_queue.put(req)
+ self.next_time += self.interval
+ time.sleep(0.1)
+ for camera in self.cameras:
+ camera.release()
+ print("CaptureThread stopped")
\ No newline at end of file
diff --git a/src/engine/MessageThreads.py b/src/engine/MessageThreads.py
new file mode 100644
index 0000000..0fd459d
--- /dev/null
+++ b/src/engine/MessageThreads.py
@@ -0,0 +1,650 @@
+import threading
+import pika
+import time
+
+
+
+class Message(object):
+ def __init__(self,sender_id,sender_key,body,routing_key):
+ self.sender_id = sender_id
+ self.sender_key = sender_key
+ self.body = body
+ self.routing_key = routing_key
+
+class senderThread(threading.Thread):
+ def __init__(self, MQparameters,input_msg_queue,sender_id,exchangeName,exchangeType):
+ super(senderThread,self).__init__()
+ self.parameters = MQparameters
+ self.queue = input_msg_queue
+ self.exchangeName = exchangeName
+ self.exchangeType = exchangeType
+ self.sender_id = sender_id
+ self.is_stop = False
+ self.connection = pika.BlockingConnection(self.parameters)
+
+ def stop(self):
+ """stop the thread"""
+ self.is_stop = True
+ self.connection.close()
+
+ def run(self):
+
+ while not self.is_stop:
+ channel = self.connection.channel()
+ channel.exchange_declare(exchange=self.exchangeName, exchange_type=self.exchangeType)
+
+ while True:
+ try:
+ if not self.queue.empty():
+ msg = self.msg_queue.get()
+ body = msg.body
+ routing_key = msg.routing_key
+ channel.basic_publish(exchange=self.exchangeName, routing_key=routing_key, body=body)
+
+ response_tag = f"{time.asctime()},{self.sender_id} send {self.exchangeName} message to {routing_key}"
+
+ print(response_tag)
+
+ elif self.queue.empty():
+ time.sleep(1)
+ except Exception as e:
+ break
+
+class receiverThread(threading.Thread):
+ def __init__(self, MQparameters,received_msg_queue,output_response_queue,exchangeName,exchangeType,receiver_id):
+ super(receiverThread,self).__init__()
+ self.parameters = MQparameters
+ self.queue = received_msg_queue
+ self.response_queue = output_response_queue
+ self.exchangeName = exchangeName
+ self.exchangeType = exchangeType
+ self.receiver_id = receiver_id
+ self.is_stop = False
+ self.connection = pika.BlockingConnection(self.parameters)
+
+ def stop(self):
+ """stop the thread"""
+ self.is_stop = True
+ self.connection.close()
+
+
+ def run(self):
+ while not self.is_stop:
+
+ channel = self.connection.channel()
+ channel.exchange_declare(exchange=self.exchangeName, exchange_type=self.exchangeType)
+
+ result = channel.queue_declare(queue='', exclusive=True, durable=True)
+ queue_name = result.method.queue
+
+ channel.queue_bind(exchange=self.exchangeName, queue=queue_name, routing_key=self.receiver_id)
+
+ channel.basic_consume(queue=queue_name, \
+ on_message_callback=self.callback)
+
+ channel.basic_qos(prefetch_count=1)
+ channel.start_consuming()
+
+ def callback(ch, method, properties, body):
+ raise NotImplementedError(
+ "Need to implement the callback function")
+
+class device_state_senderThread(senderThread):
+ def __init__(self,
+ MQparameters,
+ input_msg_queue,
+ sender_id,
+ ):
+ super(device_state_senderThread,self).__init__(
+ MQparameters, \
+ input_msg_queue,
+ sender_id,
+ "device_state",
+ "fanout"
+ )
+
+class device_state_receiverThread(receiverThread):
+ def __init__(self,
+ MQparameters,
+ receiver_id):
+ super(device_state_receiverThread,self).__init__(
+ MQparameters,
+ None,
+ None,
+ "device_state",
+ "fanout",
+ receiver_id
+ )
+
+ def callback(self, ch, method, properties, body):
+
+ response_tag = f"{time.asctime()} received: {body}."
+ print(response_tag)
+ ch.basic_ack(delivery_tag = method.delivery_tag)
+
+class alarm_state_senderThread(senderThread):
+ def __init__(self,
+ MQparameters,
+ input_msg_queue,
+ sender_id,
+ ):
+ super(alarm_state_senderThread,self).__init__(
+ MQparameters, \
+ input_msg_queue,
+ sender_id,
+ "alarm_state",
+ "fanout"
+ )
+
+class alarm_state_receiverThread(receiverThread):
+ def __init__(self,
+ MQparameters,
+ receiver_id):
+ super(alarm_state_receiverThread,self).__init__(
+ MQparameters,
+ None,
+ None,
+ "alarm_state",
+ "fanout",
+ receiver_id
+ )
+
+ def callback(self, ch, method, properties, body):
+
+ response_tag = f"{time.asctime()} received: {body}."
+ print(response_tag)
+ ch.basic_ack(delivery_tag = method.delivery_tag)
+
+class command_request_senderThread(senderThread):
+ def __init__(self,
+ MQparameters,
+ input_msg_queue,
+ sender_id,
+ ):
+ super(command_request_senderThread,self).__init__(
+ MQparameters, \
+ input_msg_queue,
+ sender_id,
+ "command_request",
+ "topic"
+ )
+
+class command_request_receiverThread(receiverThread):
+ def __init__(self,
+ MQparameters,
+ received_msg_queue,
+ response_msg_queue,
+ receiver_id):
+ super(command_request_receiverThread,self).__init__(
+ MQparameters,
+ received_msg_queue,
+ response_msg_queue,
+ "command_request",
+ "topic",
+ receiver_id
+ )
+
+ def callback(self, ch, method, properties, body):
+ received_msg = Message(None,None,body,None)
+ self.queue.put(received_msg)
+
+ response_tag = f"{time.asctime()},{self.receiver_id} received {self.exchangeName} message."
+ print(response_tag)
+ self.response_queue.put(Message(self.receiver_id,self.receiver_id,response_tag,''))
+ ch.basic_ack(delivery_tag = method.delivery_tag)
+
+class command_response_senderThread(senderThread):
+ def __init__(self,
+ MQparameters,
+ input_msg_queue,
+ sender_id,
+ ):
+ super(command_response_senderThread,self).__init__(
+ MQparameters, \
+ input_msg_queue,
+ sender_id,
+ "command_response",
+ "fanout"
+ )
+
+class command_response_receiverThread(receiverThread):
+ def __init__(self,
+ MQparameters,
+ receiver_id):
+ super(command_response_receiverThread,self).__init__(
+ MQparameters,
+ None,
+ None,
+ "command_response",
+ "fanout",
+ receiver_id
+ )
+
+ def callback(self, ch, method, properties, body):
+ response_tag = f"{time.asctime()} received: {body}."
+ print(response_tag)
+ ch.basic_ack(delivery_tag = method.delivery_tag)
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+# class device_state_up_msg_sender_thread(threading.Thread):
+# def __init__(self, MQparameters,device_state_msg_queue):
+# super().__init__(self)
+# self.parameters = MQparameters
+# self.queue = device_state_msg_queue
+# self.connection = pika.BlockingConnection(self.parameters)
+
+# def stop(self):
+# """stop the thread"""
+# self.is_stop = True
+# self.connection.close()
+
+# def run(self):
+
+# while not self.is_stop:
+# print("device_state_up_msg_sender_thread started")
+# channel = self.connection.channel()
+# channel.exchange_declare(exchange='device_state', exchange_type='fanout')
+
+# while True:
+# try:
+# if not self.queue.empty():
+# msg = self.msg_queue.get()
+# channel.basic_publish(exchange='device_state', routing_key='', body=msg)
+# print(time.asctime(), "send device state message")
+# elif self.queue.empty():
+# time.sleep(1)
+# except Exception as e:
+# break
+
+# class device_state_up_msg_receiver_thread(threading.Thread):
+# def __init__(self, MQparameters):
+# super().__init__(self)
+# self.parameters = MQparameters
+# self.connection = pika.BlockingConnection(self.parameters)
+
+# def stop(self):
+# """stop the thread"""
+# self.is_stop = True
+# self.connection.close()
+
+# def run(self):
+# while not self.is_stop:
+# print("device_state_up_msg_receiver_thread started")
+
+# channel = self.connection.channel()
+# channel.exchange_declare(exchange='device_state', exchange_type='fanout')
+
+# result = channel.queue_declare(queue='', exclusive=True, durable=True)
+# queue_name = result.method.queue
+
+# channel.queue_bind(exchange='device_state', queue=queue_name)
+
+# def callback(ch, method, properties, body):
+# print(f"Received: {body}")
+# ch.basic_ack(delivery_tag = method.delivery_tag)
+
+# channel.basic_consume(queue=queue_name, \
+# on_message_callback=callback)
+
+# channel.basic_qos(prefetch_count=1)
+# channel.start_consuming()
+
+# class alarm_state_up_msg_sender_thread(threading.Thread):
+
+# def __init__(self, MQparameters,alarm_state_msg_queue):
+# super().__init__(self)
+# self.parameters = MQparameters
+# self.queue = alarm_state_msg_queue
+# self.connection = pika.BlockingConnection(self.parameters)
+
+# def stop(self):
+# """stop the thread"""
+# self.is_stop = True
+# self.connection.close()
+
+# def run(self):
+
+# while not self.is_stop:
+# print("alarm_state_up_msg_sender_thread started")
+# channel = self.connection.channel()
+# channel.exchange_declare(exchange='alarm_state', exchange_type='fanout')
+
+# while True:
+# try:
+# if not self.queue.empty():
+# msg = self.msg_queue.get()
+# channel.basic_publish(exchange='alarm_state', routing_key='', body=msg)
+# print(time.asctime(), "send alarm state message")
+# elif self.queue.empty():
+# time.sleep(1)
+# except Exception as e:
+# break
+
+# class alarm_state_up_msg_receiver_thread(threading.Thread):
+# def __init__(self, MQparameters):
+# super().__init__(self)
+# self.parameters = MQparameters
+# self.connection = pika.BlockingConnection(self.parameters)
+
+# def stop(self):
+# """stop the thread"""
+# self.is_stop = True
+# self.connection.close()
+
+# def run(self):
+
+# while not self.is_stop:
+# print("alarm_state_up_msg_receiver_thread started")
+# channel = self.connection.channel()
+# channel.exchange_declare(exchange='alarm_state', exchange_type='fanout')
+
+# result = channel.queue_declare(queue='', exclusive=True, durable=True)
+# queue_name = result.method.queue
+
+# channel.queue_bind(exchange='alarm_state', queue=queue_name)
+
+# def callback(ch, method, properties, body):
+# print(f"Received: {body}")
+# ch.basic_ack(delivery_tag = method.delivery_tag)
+
+# channel.basic_consume(queue=queue_name, \
+# on_message_callback=callback)
+
+# channel.basic_qos(prefetch_count=1)
+# channel.start_consuming()
+
+# class command_request_msg_sender_thread(threading.Thread):
+
+# def __init__(self, MQparameters,command_request_msg_queue,routing_key_queue):
+# super().__init__(self)
+# self.parameters = MQparameters
+# self.command_queue = command_request_msg_queue
+# self.key_queue = routing_key_queue
+# self.connection = pika.BlockingConnection(self.parameters)
+
+# def stop(self):
+# """stop the thread"""
+# self.is_stop = True
+# self.connection.close()
+
+# def run(self):
+
+# while not self.is_stop:
+# print("command_request_msg_sender_thread started")
+# channel = self.connection.channel()
+# channel.exchange_declare(exchange='command_msg', exchange_type='topic')
+
+# while True:
+# try:
+# if not self.command_queue.empty() and \
+# not self.key_queue.empty():
+# msg = self.command_queue.get()
+# routing_key = self.key_queue.get()
+# corr_id = str(uuid.uuid4())
+# channel.basic_publish(exchange='command_msg',
+# routing_key = routing_key,
+# body=msg,
+# properties=pika.BasicProperties(
+# correlation_id=corr_id
+# )
+# )
+# print(time.asctime(), f"send command request with ID:[{corr_id}]")
+# elif self.queue.empty():
+# time.sleep(1)
+# except Exception as e:
+# break
+
+# class command_request_msg_receiver_thread(threading.Thread):
+# def __init__(self, MQparameters,command_request_msg_queue, \
+# command_response_msg_queue, routing_key):
+# super().__init__(self)
+# self.parameters = MQparameters
+# self.queue = command_request_msg_queue
+# self.response_queue = command_response_msg_queue
+# self.routing_key = routing_key
+# self.connection = pika.BlockingConnection(self.parameters)
+
+# def stop(self):
+# """stop the thread"""
+# self.is_stop = True
+# self.connection.close()
+
+# def run(self):
+
+# while not self.is_stop:
+# print("command_request_msg_receiver_thread started")
+# channel = self.connection.channel()
+# channel.exchange_declare(exchange='command_msg', exchange_type='topic')
+
+# result = channel.queue_declare(queue='', exclusive=True, durable=True)
+# queue_name = result.method.queue
+
+# channel.queue_bind(exchange='command_msg', queue=queue_name, \
+# routing_key=self.routing_key)
+
+# def callback(ch, method, prop, body):
+# print(time.asctime(), f"reveive command request with ID:[{prop.corr_id}]")
+# self.queue.put(body)
+# self.response_queue.put("")
+# ch.basic_ack(delivery_tag = method.delivery_tag)
+
+# channel.basic_consume(queue=queue_name, \
+# on_message_callback=callback)
+
+# channel.basic_qos(prefetch_count=1)
+# channel.start_consuming()
+
+# class command_response_msg_sender_thread(threading.Thread):
+# def __init__(self, MQparameters,command_request_msg_queue):
+# super().__init__(self)
+# self.parameters = MQparameters
+# self.queue = command_request_msg_queue
+# self.connection = pika.BlockingConnection(self.parameters)
+
+# def stop(self):
+# """stop the thread"""
+# self.is_stop = True
+# self.connection.close()
+
+# def run(self):
+
+# while not self.is_stop:
+# print("command_response_msg_sender_thread started")
+# channel = self.connection.channel()
+# channel.exchange_declare(exchange='command_response', exchange_type='fanout')
+
+# while True:
+# try:
+# if not self.queue.empty():
+# msg = self.msg_queue.get()
+# channel.basic_publish(exchange='command_response', routing_key='', body=msg)
+# print(time.asctime(), "send device state message")
+# elif self.queue.empty():
+# time.sleep(1)
+# except Exception as e:
+# break
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+# class command_down_rpc_sender_thread(threading.Thread):
+# def __init__(self, MQparameters, command_queue, routing_key_queue):
+# super().__init__(self)
+# self.parameters = MQparameters
+# self.command_queue = command_queue
+# self.key_queue = routing_key_queue
+# self.connection = pika.BlockingConnection(self.parameters)
+
+# def stop(self):
+# """stop the thread"""
+# self.is_stop = True
+# self.connection.close()
+
+# def run(self):
+
+# while not self.is_stop:
+# # print("command_down_rpc_sender_thread started")
+
+# channel = self.connection.channel()
+
+# # declare a exchange
+# channel.exchange_declare(exchange="command_rpc", \
+# exchange_type='topic')
+
+# # declare a callback queue
+# result = channel.queue_declare(queue='', exclusive=True)
+# callback_queue = result.method.queue
+
+# # bingding the queue
+# channel.queue_bind(
+# exchange="command_rpc",
+# queue=callback_queue,
+# routing_key=routing_key
+# )
+
+# # init the command uuid
+# corr_id = str(uuid.uuid4())
+# command_response = None
+
+# # def a function to print the command respond message
+# def respond_callback(ch, method, properties, body):
+# if properties.correlation_id == corr_id:
+# command_response = body
+# print(body)
+
+# # start the cosumer
+# channel.basic_consume(
+# queue=callback_queue,
+# on_message_callback=respond_callback,
+# auto_ack=True
+# )
+
+# # publish with tropic exchange
+# channel.basic_publish(
+# exchange="command_rpc",
+# routing_key=routing_key,
+# body = request_command,
+# properties=pika.BasicProperties(
+# reply_to=callback_queue,
+# correlation_id=corr_id
+# ),
+# )
+# with command_response is None:
+# self.connection.process_data_events()
+
+# class command_down_rpc_receiver_thread(threading.Thread):
+# def __init__(self, MQparameters):
+# super().__init__(self)
+# self.parameters = MQparameters
+# self.connection = pika.BlockingConnection(self.parameters)
+
+# def stop(self):
+# """stop the thread"""
+# self.is_stop = True
+# self.connection.close()
+
+# def run(self,commands_queue):
+# while not self.is_stop:
+# print("command_down_rpc_receiver_thread started")
+# channel = self.connection.channel()
+
+# # declare a exchange
+# channel.exchange_declare(exchange="command_rpc", \
+# exchange_type='topic')
+
+# # declare a callback queue
+# result = channel.queue_declare(queue='', exclusive=True)
+# callback_queue = result.method.queue
+
+# # bingding the queue
+# channel.queue_bind(
+# exchange="command_rpc",
+# queue=callback_queue,
+# routing_key='pod*'
+# )
+
+# # def a function to reply the request from the sender
+# def on_request(ch, method, prop, body):
+# commands_queue.put(body)
+
+# response = f"{time.asctime()} {self.device_id} receive the command uuid:[{prop.correlation_id}]"
+# print(response)
+
+# ch.basic_publish(exchange="command_rpc",
+# routing_key = prop.reply_to,
+# properties = pika.BasicProperties(
+# correlation_id=prop.corr_id
+# ),
+# body = response)
+# ch.basic_ack(delivery_tag = method.delivery_tag)
+
+# # start the cosumer
+# channel.basic_qos(prefetch_count=1)
+# channel.basic_consume(
+# queue=callback_queue,
+# on_message_callback=on_request,
+# auto_ack=True
+# )
+# channel.start_consuming()
+
+# class image_data_up_msg_sender_thread(threading.Thread):
+# pass
+
+# class image_data_up_msg_receiver_thread(threading.Thread):
+# pass
+
+# class ply_data_up_msg_sender_thread(threading.Thread):
+# pass
+
+# class ply_data_up_msg_receiver_thread(threading.Thread):
+# pass
\ No newline at end of file
diff --git a/src/engine/__init__.py b/src/engine/__init__.py
index 2a904ab..777c4fd 100644
--- a/src/engine/__init__.py
+++ b/src/engine/__init__.py
@@ -5,4 +5,6 @@
from .DustCleaner import *
from .DustCleanerThread import *
from .MessageQueueThread import *
-from .MicroWaveUDPThread import *
\ No newline at end of file
+from .MicroWaveUDPThread import *
+from .MessageThreads import *
+from .CameraDevice import *
\ No newline at end of file
diff --git a/src/utils/tools.py b/src/utils/tools.py
index 463b80a..3531cb3 100644
--- a/src/utils/tools.py
+++ b/src/utils/tools.py
@@ -38,6 +38,8 @@
return ply_array
+
+
if __name__ == "__main__":
ply_array = read_ply('src\\microwave\\demo\\demoData\\2522.ply')
--
Gitblit v1.9.1