From d8e5c7c93ed28afa2af1d68ea20cff962599626f Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期二, 14 六月 2022 15:57:55 +0800
Subject: [PATCH] server enhanced!
---
/dev/null | 134 ------
src/protos/aerial.proto | 35 +
scripts/mqServer.py | 17
src/slam/__init__.py | 6
src/Server.py | 217 ++++------
src/engine/__init__.py | 20
scripts_launch/TEST_startServer.bat | 9
src/engine/MessageThreads.py | 709 ++--------------------------------
src/__init__.py | 10
src/microwave/__init__.py | 10
src/utils/__init__.py | 2
11 files changed, 219 insertions(+), 950 deletions(-)
diff --git a/scripts/mqServer.py b/scripts/mqServer.py
index d1db978..931e7b0 100644
--- a/scripts/mqServer.py
+++ b/scripts/mqServer.py
@@ -8,26 +8,31 @@
from src.Server import MyServer
-
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:
+ if len(sys.argv) != 8:
help()
try:
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])
- save_folder = SAVEfolder
+ save_folder = str(sys.argv[6])
+ server_name = str(sys.argv[7])
except Exception as e:
print(e)
help()
- server = MyServer(host=MQhost, user=MQuser, password=MQpassword, queue=QUEUEname, port=WEBport, SAVEfolder=save_folder)
+ server = MyServer(MQhost=MQhost,
+ MQuser=MQuser,
+ MQpassword=MQpassword,
+ WEBhost=WEBhost,
+ WEBport=WEBport,
+ save_folder=save_folder,
+ server_name=server_name)
server.run_all()
\ No newline at end of file
diff --git a/scripts_launch/TEST_startServer.bat b/scripts_launch/TEST_startServer.bat
index f956433..06a18e0 100644
--- a/scripts_launch/TEST_startServer.bat
+++ b/scripts_launch/TEST_startServer.bat
@@ -16,20 +16,21 @@
echo change to path: %cd%
-@REM set MQhost=localhost
-set MQhost=47.92.33.19
+set MQhost=localhost
+@REM set MQhost=47.92.33.19
set MQuser=chai
set MQpassword=password123
set WEBhost=0.0.0.0
set WEBport=5000
-set SAVEfolder=./Received_data_testing
+set SAVEfolder=./data/test/
+set SERVERname=testserver
set LOGfile=ServerTEST.log
@REM Launch the Server
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% ./scripts/mqServer.py %MQhost% %MQuser% %MQpassword% %QUEUEname% %WEBport% %SAVEfolder%
+%aerialpython% ./scripts/mqServer.py %MQhost% %MQuser% %MQpassword% %WEBhost% %WEBport% %SAVEfolder% %SERVERname%
echo Server.py has been terminited.
pause
diff --git a/src/Server.py b/src/Server.py
index 90dacf3..3b50887 100644
--- a/src/Server.py
+++ b/src/Server.py
@@ -1,175 +1,142 @@
import sys # NOQA: E402
import os
from flask import Flask
-from numpy import true_divide
import pika
import uuid
import queue
from werkzeug.wrappers import request
+from threading import Thread
+from datetime import datetime
-from zmq import device
from .engine.DeviceManager import DeviceManager
-from .engine.MessageQueueThread import MessageQueueThread
+from .engine import MessageThreads as ths
from .protos import aerial_pb2 as pb
+
class MyServer(Flask):
- def __init__(self, host, user, password, queue, port, SAVEfolder):
+ def __init__(self,
+ MQhost:str,
+ MQuser:str,
+ MQpassword:str,
+ WEBhost:str,
+ WEBport:int,
+ save_folder:str,
+ server_name:str
+ ):
Flask.__init__(self, __name__)
- self.host = host
- self.user = user
- self.password = password
- self.queue = queue
- self.port = port
- global save_folder
- save_folder = SAVEfolder
+ self.MQhost = MQhost
+ self.MQuser = MQuser
+ self.MQpassword = MQpassword
+ self.WEBhost = WEBhost
+ self.WEBport = WEBport
+ self.save_folder = save_folder
+ self.server_name = "server0"
- self.route('/api', methods=['POST'])(self.api)
+ # self.route('/api', methods=['POST'])(self.api)
self.route('/', methods=['GET'])(self.hello)
- self._def_routes()
- # 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
+ # self._def_routes()
+
# queue
- self.command_request_queue = queue.Queue()
+ self.LogsReceiver_queue = queue.Queue()
+ self.ImagesReceiver_queue = queue.Queue()
+ self.PlysReceiver_queue = queue.Queue()
+
+ self.CommandsSender_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)
+ connection = pika.BlockingConnection(self.parameters)
print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
+ connection.close()
- 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 _receiver_Thread_Launcher(self):
- 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()
+ # launch logs receiver thread
+ self.LogsReceiver_Thread = ths.receiverThread(
+ MQparams=self.parameters,
+ exchange_name='logs_topic',
+ exchange_type='topic',
+ MQqueue_name=f'logs_queue',
+ binding_keys='server.logs.test0',
+ message_queue=self.LogsReceiver_queue
+ )
+ self.LogsReceiver_Thread.setDaemon(True)
+ self.LogsReceiver_Thread.setName("logs_thread")
+ self.LogsReceiver_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
+ # launch images receiver thread
+ self.ImagesReceiver_Thread = ths.receiverThread(
+ MQparams=self.parameters,
+ exchange_name='images_topic',
+ exchange_type='topic',
+ MQqueue_name=f'images_queue',
+ binding_keys="server.images.test0",
+ message_queue=self.ImagesReceiver_queue
+ )
+ self.ImagesReceiver_Thread.setDaemon(True)
+ self.ImagesReceiver_Thread.setName("images_thread")
+ self.ImagesReceiver_Thread.start()
- 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()
+ # launch plys receiver thread
+ self.PlysReceiver_Thread = ths.receiverThread(
+ MQparams=self.parameters,
+ exchange_name='plys_topic',
+ exchange_type='topic',
+ MQqueue_name=f'plys_queue',
+ binding_keys='server.plys.test0',
+ message_queue=self.PlysReceiver_queue
+ )
+ self.PlysReceiver_Thread.setDaemon(True)
+ self.PlysReceiver_Thread.setName("plys_thread")
+ self.PlysReceiver_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
+ # launch decode thread
+ queue_list = [self.LogsReceiver_queue,self.ImagesReceiver_queue,self.PlysReceiver_queue]
+ queue_tags = ["Logs", "Images", "Plys"]
+ self.MsgDecoder_Thread = ths.decodeThread(
+ queue_list= queue_list,
+ queue_tags= queue_tags,
+ save_path= self.save_folder
+ )
- 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()
+ self.MsgDecoder_Thread.setDaemon(True)
+ self.MsgDecoder_Thread.setName("decode_thread")
+ self.MsgDecoder_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):
+ def _sender_Thread_Launcher(self):
- self.command_request_queue.put(ths.Message(None,None,req.command,req.target))
- return
+ # launcher command sender thread
+ self.CommandsSender_Thread = ths.senderThread(
+ MQparams=self.parameters,
+ exchange_name='commands_topic',
+ exchange_type='topic',
+ message_queue=self.CommandsSender_queue
+ )
+ self.CommandsSender_Thread.setDaemon(True)
+ self.CommandsSender_Thread.setName("commands_thread")
+ self.CommandsSender_Thread.start()
- 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) # call the function
-
- 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):
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()
+ self._receiver_Thread_Launcher()
+ self._sender_Thread_Launcher()
+
def run_all(self):
self.run_rabbitmq()
self.run(debug=False, host=self.WEBhost, port=self.WEBport)
-<<<<<<< HEAD
-
-if __name__ == "__main__":
- def help():
- 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:
- help()
- try:
- MQhost = str(sys.argv[1])
- MQuser = str(sys.argv[2])
- MQpassword = str(sys.argv[3])
- WEBhost = str(sys.argv[4])
- WEBport = int(sys.argv[5])
- SAVEfolder = str(sys.argv[6])
-
- except Exception as e:
- print(e)
- help()
- server = MyServer(MQhost, MQuser, MQpassword, WEBhost, WEBport, SAVEfolder)
- server.run_all()
-=======
->>>>>>> aa8a0ba3a86ff5dbbdfcd68c72a5e70e1c52e877
diff --git a/src/Server1.py b/src/Server1.py
deleted file mode 100644
index 1ba60f2..0000000
--- a/src/Server1.py
+++ /dev/null
@@ -1,134 +0,0 @@
-import sys # NOQA: E402
-import os
-from flask import Flask
-import pika
-import uuid
-import queue
-from werkzeug.wrappers import request
-from threading import Thread
-from datetime import datetime
-
-
-
-from .engine.DeviceManager import DeviceManager
-from .engine import MessageThreads as ths
-from .protos import aerial_pb2 as pb
-
-
-
-class MyServer(Flask):
- def __init__(self, host, user, password, queue, port, SAVEfolder):
- Flask.__init__(self, __name__)
- self.host = host
- self.user = user
- self.password = password
- self.queue = queue
- self.port = port
- global save_folder
- save_folder = SAVEfolder
-
-
- self.route('/api', methods=['POST'])(self.api)
- self.route('/', methods=['GET'])(self.hello)
- self._def_routes()
- # threads
-
- # 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)
- # 杩炴帴闃熷垪鏈嶅姟鍣�
- connection = pika.BlockingConnection(self.parameters)
- print(f"connected to MQ server: {self.MQhost} with user {self.MQuser}")
- connection.close()
-
-
-
- def _message_receiver_Thread_launcher(self):
-
- def callback(ch,method,properties,body):
- timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
- print(f"[x] {timestamp} | [{method.routing_key}]:{body}")
-
- target_args = (self.parameters,
- "message_topic",
- "topic",
- "message_queue",
- ["server.server.message", "server.server0.message"],
- callback)
-
- self.message_receiver_thread = Thread(target=ths.message_receiver,
- args=target_args)
- self.message_receiver_thread.setDaemon(True)
- self.message_receiver_thread.start()
-
-
- def _image_receiver_Thread_launcher(self):
-
- def callback(ch,method,properties,body):
-
- timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
- print(f"[x] {timestamp} | [{method.routing_key}]:{body}")
-
- target_args = (self.parameters,
- "message_topic",
- "topic",
- "message_queue",
- ["server.server.message", "server.server0.message"],
- callback)
-
- self.message_receiver_thread = Thread(target=ths.message_receiver,
- args=target_args)
- self.message_receiver_thread.setDaemon(True)
- self.message_receiver_thread.start()
-
-
-
- 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):
-
- 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) # call the function
-
- 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):
- 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.run(debug=False, host=self.WEBhost, port=self.WEBport)
-
diff --git a/src/__init__.py b/src/__init__.py
index 29a8a06..4eb1365 100644
--- a/src/__init__.py
+++ b/src/__init__.py
@@ -1,5 +1,5 @@
-from .engine import *
-from .microwave import *
-from .protos import *
-from .slam import *
-from .utils import *
\ No newline at end of file
+# from .engine import *
+# from .microwave import *
+# from .protos import *
+# from .slam import *
+# from .utils import *
\ No newline at end of file
diff --git a/src/engine/MessageThreads.py b/src/engine/MessageThreads.py
index 2b33936..04d86e2 100644
--- a/src/engine/MessageThreads.py
+++ b/src/engine/MessageThreads.py
@@ -7,6 +7,7 @@
import sys
from datetime import datetime
from typing import List
+import numpy as np
# # add python path of src to sys.path
# src_path = os.path.join(__file__, *(['..'] * 2))
@@ -14,6 +15,7 @@
# sys.path.insert(0, src_path)
from ..protos import aerial_pb2 as pb
+from ..utils.tools import write_ply
@@ -33,7 +35,7 @@
# routing_key:str,
message_queue:queue.Queue,
):
- super(senderThread,self).__init__()
+ super().__init__()
self.MQparams = MQparams
self.exchange_name = exchange_name
@@ -78,7 +80,7 @@
)
timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
- caption = f"[x] {timestamp} | Sending message to[{msg_routing_key}]")
+ caption = f"[x] {timestamp} | Sending message to[{msg_routing_key}]"
print(caption)
else:
@@ -94,10 +96,10 @@
exchange_name: str,
exchange_type:str,
MQqueue_name:str,
- binding_keys:List(str),
+ binding_keys:List[str],
message_queue:queue.Queue,
):
- super(senderThread,self).__init__()
+ super().__init__()
self.MQparams = MQparams
self.exchange_name = exchange_name
@@ -127,7 +129,7 @@
# declare the queue and bind it with keys
channel.queue_declare(queue=self.MQqueue_name, durable=True)
for binding_key in self.binding_keys:
- channel.queue_bind(exchange=self.exchangeName, queue=self.MQqueue_name, routing_key=binding_key)
+ channel.queue_bind(exchange=self.exchange_name, queue=self.MQqueue_name, routing_key=binding_key)
# consume the message
channel.basic_qos(prefetch_count=1)
@@ -143,8 +145,8 @@
class decodeThread(threading.Thread):
def __init__(self,
- queue_list:List(queue.Queue),
- queue_tags:List(str),
+ queue_list:List[queue.Queue],
+ queue_tags:List[str],
save_path:str):
super().__init__()
@@ -155,7 +157,6 @@
self.is_stop = False
assert len(self.queue_list) == len(self.queue_tags)
-
def stop(self):
"""stop the thread"""
@@ -197,8 +198,7 @@
decode_method(queue)
# next queue
i += 1
-
-
+
def decode_select(self,queue_tags:str):
route_dic = {
"Logs":self.decode_LogsMsg,
@@ -219,6 +219,7 @@
msg_pb.ParseFromString(msg)
# save the raw pb msg
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
raw_file_name = os.path.join(self.logsmsg_folder,f"{curr_time}.LogsMsg")
with open(raw_file_name, 'wb') as file:
file.write(msg)
@@ -235,7 +236,6 @@
# next msg
count += 1
-
def decode_ImagesMsg(self,queue):
while not queue.empty():
@@ -245,7 +245,8 @@
msg_pb = pb.ImagesMessage()
msg_pb.ParseFromString(msg)
- # save the raw pb msg
+ # save the raw pb msg
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
raw_file_name = os.path.join(self.imagesmsg_folder,f"{curr_time}.ImagesMsg")
with open(raw_file_name, 'wb') as file:
file.write(msg)
@@ -261,7 +262,7 @@
image_byte = image.data
# save the images
- if (image_byte is None) | (len(image_byte) == 0):
+ if (image_byte is None) | (len(image_byte) != imagelegth ):
pass
else:
filename = f"{msg_source}_{msg_timestamp}_image{id}.jpg"
@@ -276,656 +277,52 @@
# next image item
id += 1
-
def decode_PlysMsg(self,queue):
+ while not queue.empty():
+ # extracting the msg from queue
+ msg = queue.pop()
+ msg_pb = pb.PlyMessage()
+ msg_pb.ParseFromString(msg)
-
- pass
+ # save the raw pb msg
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ raw_file_name = os.path.join(self.plysmsg_folder,f"{curr_time}.PlysMsg")
+ with open(raw_file_name, 'wb') as file:
+ file.write(msg)
+
+ # process the msg
+ msg_source = msg_pb.source
+ msg_timestamp = msg_pb.timestamp
+
+ # process the muti ply data
+ id = 0
+ for ply in msg_pb.Plydata:
+ data_legth = ply.length
+ data_byte = ply.data
+ rows = ply.rows
+ cols = ply.cols
+ order = ply.order
+ dt = ply.d_types
+
+ # save the pointcloud array data
+ if (data_byte is None) | (len(data_byte) != data_legth):
+ pass
+ else:
+ # frame the bytes to array
+ data_array = np.frombuffer(data_byte,dtype=dt)
+ data_array = data_array.reshape(rows,cols)
+ # save the data with .ply format
+ filename = f"{msg_source}_{msg_timestamp}_pointcloud{id}"
+ write_ply(data_array,self.plysmsg_folder,filename)
+
+ # next ply item
+ id += 1
def decode_CommandsMsg(self,queue):
- pass
+ raise NotImplementedError("NOT IMPLEMENT!")
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-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"
- )
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-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)
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-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"
- )
- # 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-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)
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-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"
- )
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-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)
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-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"
- )
-# 鏍规嵁rabbitmq閲嶆柊璁剧疆娑堟伅鍙傛暟
-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)
-
-
-# TODO
-class image_data_senderThread(senderThread):
- def __init__(self,
- MQparameters,
- input_msg_queue,
- sender_id,
- ):
- super(command_response_senderThread,self).__init__(
- MQparameters, \
- input_msg_queue,
- sender_id,
- "image_data",
- "fanout"
- )
-
-class image_data_receiverThread(receiverThread):
- pass
-
-
-def message_sender(MQparams: pika.ConnectionParameters,
- exchange_name: str,
- exchange_type:str,
- routing_key:str,
- message:str,
- ):
- # connect to RabbitMQ
- connection = pika.BlockingConnection(MQparams)
-
- # declare a exchange
- channel = connection.channel()
- channel.exchange_declare(exchange=exchange_name,
- exchange_type=exchange_type)
- # publish a PERSISTENT message
- channel.basic_publish(
- exchange=exchange_name,
- routing_key=routing_key,
- body=message,
- properties=pika.BasicProperties(
- delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
- )
- )
- # close the connection
- connection.close()
-
-def message_receiver(MQparams: pika.ConnectionParameters,
- exchange_name: str,
- exchange_type:str,
- queue_name:str,
- binding_keys:List[str],
- callback_fn:callable
- ):
- # connect to RabbitMQ
- connection = pika.BlockingConnection(MQparams)
-
- # declare a exchange
- channel = connection.channel()
- channel.exchange_declare(exchange=exchange_name,
- exchange_type=exchange_type)
-
- # declare a durable queue
- channel.queue_declare(queue=queue_name,durable=True)
-
- # literate the binding keys and bind the queue
- for binding_key in binding_keys:
- channel.queue_bind(queue=queue_name,exchange=exchange_name,
- routing_key=binding_key)
-
- # waiting for consuming the messages
- channel.basic_qos(prefetch_count=1)
- channel.basic_consume(queue=queue_name,
- auto_ack=False,
- on_message_callback=callback_fn
- )
- channel.start_consuming()
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-# 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 965b63f..48ca3a4 100644
--- a/src/engine/__init__.py
+++ b/src/engine/__init__.py
@@ -1,10 +1,10 @@
-from .AcquisitionMain import *
-from .CaptureThread import *
-from .ControllThread import *
-from .DeviceManager import *
-from .DustCleaner import *
-from .DustCleanerThread import *
-from .MessageQueueThread import *
-from .MicroWaveUDPThread import *
-from .MessageThreads import *
-from .devices import *
\ No newline at end of file
+# from .AcquisitionMain import *
+# from .CaptureThread import *
+# from .ControllThread import *
+# from .DeviceManager import *
+# from .DustCleaner import *
+# from .DustCleanerThread import *
+# from .MessageQueueThread import *
+# from .MicroWaveUDPThread import *
+# from .MessageThreads import *
+# from .devices import *
\ No newline at end of file
diff --git a/src/microwave/__init__.py b/src/microwave/__init__.py
index ba4d4a0..11f4734 100644
--- a/src/microwave/__init__.py
+++ b/src/microwave/__init__.py
@@ -1,5 +1,5 @@
-from .components import *
-from .constants import *
-from .parsers import *
-from .servers import *
-from .deviceTools import *
\ No newline at end of file
+# from .components import *
+# from .constants import *
+# from .parsers import *
+# from .servers import *
+# from .deviceTools import *
\ No newline at end of file
diff --git a/src/protos/aerial.proto b/src/protos/aerial.proto
index 1a81118..04e7b2f 100644
--- a/src/protos/aerial.proto
+++ b/src/protos/aerial.proto
@@ -138,17 +138,50 @@
}
+enum DataType
+{
+ c_int16 = 1;
+ c_int32 = 2;
+ c_int64 = 3;
+ c_float16 = 4;
+ c_float32 = 5;
+ c_float = 6;
+}
+
+
message PlyMessage
{
string source = 1;
string timestamp = 2;
- bytes plydata = 3;
+
+ message PointCloudItem
+ {
+ int32 length = 1;
+ int32 rows = 2;
+ int32 cols = 3;
+ string order = 4;
+ DataType d_types = 5;
+ bytes data = 6
+
+ }
+
+ repeated PointCloudItem Plydata = 3;
+
+ // enum ref_system
+ // {
+ // MICROWAVE_SYSTEM = 0;
+ // DEVICE_SYSTEM = 1;
+ // }
+
+ // ref_system reference = 4
}
+
+
// message MicroWaveMessage
// {
// int32 length = 1; // 鏁版嵁闀垮害
diff --git a/src/slam/__init__.py b/src/slam/__init__.py
index dd48035..05bcc71 100644
--- a/src/slam/__init__.py
+++ b/src/slam/__init__.py
@@ -1,3 +1,3 @@
-from .core import *
-from .manualMode import *
-from .autoMode import *
\ No newline at end of file
+# from .core import *
+# from .manualMode import *
+# from .autoMode import *
\ No newline at end of file
diff --git a/src/utils/__init__.py b/src/utils/__init__.py
index 6ac1ccc..647bef1 100644
--- a/src/utils/__init__.py
+++ b/src/utils/__init__.py
@@ -1 +1 @@
-from .tools import *
\ No newline at end of file
+# from .tools import *
\ No newline at end of file
--
Gitblit v1.9.1