From 74c96c72625aff6613d69c7168e1a7d76da22870 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期五, 24 六月 2022 17:18:31 +0800
Subject: [PATCH] add taskuuid
---
env/requirements/ipywidgets-7.7.1-py2.py3-none-any.whl | 0
src/microwave/servers.py | 3
src/engine/MessageThreads.py | 147 +++++++-----
env/requirements/widgetsnbextension-3.6.1-py2.py3-none-any.whl | 0
src/utils/tools.py | 23 +
src/engine/AcquisitionMain.py | 78 +++++
README.md | 28 ++
/dev/null | 0
src/protos/aerial.proto | 230 +++++++++++--------
env/requirements/open3d-0.14.1-cp39-cp39-win_amd64.whl | 0
requirements.txt | 20
src/Server.py | 71 +++++
env/requirements/jupyterlab_widgets-1.1.1-py3-none-any.whl | 0
env/requirements/numpy-1.23.0-cp39-cp39-win_amd64.whl | 0
src/engine/devices.py | 63 ++++-
15 files changed, 466 insertions(+), 197 deletions(-)
diff --git a/README.md b/README.md
index b12bded..78d850f 100644
--- a/README.md
+++ b/README.md
@@ -5,4 +5,30 @@
1. 鏍规嵁proto鏁版嵁缁撴瀯锛岄噸鏂颁紭鍖杝rc\engine\MessageThreads.py鐨勭浉鍏冲唴瀹癸紱
2. 閲囬泦绔細淇敼src\engine\devices.py鍐呯殑鐩告満鐨勬暟鎹噰闆嗙嚎绋嬶紱
3. 鏈嶅姟绔細澧炲姞鍥惧儚鏁版嵁鐨勬秷鎭В鏋愶紱
-4. src\engine\MessageThreads.py鍐呮坊鍔犳暟鎹彂閫佸拰鎺ユ敹绾跨▼銆�
\ No newline at end of file
+4. src\engine\MessageThreads.py鍐呮坊鍔犳暟鎹彂閫佸拰鎺ユ敹绾跨▼銆�
+
+
+## MQ keys
+
+馃巿Given that the name parameters of one couple of the server and client is set to as below shown:
+馃毄server name : `server0`
+馃毄client name : `bunker0.block0.device0`
+
+
+1. Server:
+ > Binding keys for receive messages from clients are shown below:
+ >> `server0.logs.test0`
+ >> `server0.images.test0`
+ >> `server0.plys.test0`
+
+ > Routing keys for send commands to the client are shown below:
+ >> `bunker0.block0.device0`
+
+2. Client:
+ > Binding keys for receive commands from the server are shown below:
+ >> `bunker0.block0.device0`
+
+ > Routing keys for send messages to the server are shown below:
+ >> `server0.logs.test0`
+ >> `server0.images.test0`
+ >> `server0.plys.test0`
\ No newline at end of file
diff --git a/env/requirements/ipywidgets-7.7.0-py2.py3-none-any.whl b/env/requirements/ipywidgets-7.7.1-py2.py3-none-any.whl
similarity index 84%
rename from env/requirements/ipywidgets-7.7.0-py2.py3-none-any.whl
rename to env/requirements/ipywidgets-7.7.1-py2.py3-none-any.whl
index 171191d..7c305c0 100644
--- a/env/requirements/ipywidgets-7.7.0-py2.py3-none-any.whl
+++ b/env/requirements/ipywidgets-7.7.1-py2.py3-none-any.whl
Binary files differ
diff --git a/env/requirements/jupyterlab_widgets-1.1.0-py3-none-any.whl b/env/requirements/jupyterlab_widgets-1.1.0-py3-none-any.whl
deleted file mode 100644
index c4c53f7..0000000
--- a/env/requirements/jupyterlab_widgets-1.1.0-py3-none-any.whl
+++ /dev/null
Binary files differ
diff --git a/env/requirements/jupyterlab_widgets-1.1.1-py3-none-any.whl b/env/requirements/jupyterlab_widgets-1.1.1-py3-none-any.whl
new file mode 100644
index 0000000..991a544
--- /dev/null
+++ b/env/requirements/jupyterlab_widgets-1.1.1-py3-none-any.whl
Binary files differ
diff --git a/env/requirements/numpy-1.22.4-cp39-cp39-win_amd64.whl b/env/requirements/numpy-1.22.4-cp39-cp39-win_amd64.whl
deleted file mode 100644
index a5e59ff..0000000
--- a/env/requirements/numpy-1.22.4-cp39-cp39-win_amd64.whl
+++ /dev/null
Binary files differ
diff --git a/env/requirements/numpy-1.23.0-cp39-cp39-win_amd64.whl b/env/requirements/numpy-1.23.0-cp39-cp39-win_amd64.whl
new file mode 100644
index 0000000..6126135
--- /dev/null
+++ b/env/requirements/numpy-1.23.0-cp39-cp39-win_amd64.whl
Binary files differ
diff --git a/env/requirements/open3d-0.14.1-cp39-cp39-win_amd64.whl b/env/requirements/open3d-0.14.1-cp39-cp39-win_amd64.whl
new file mode 100644
index 0000000..ba26c00
--- /dev/null
+++ b/env/requirements/open3d-0.14.1-cp39-cp39-win_amd64.whl
Binary files differ
diff --git a/env/requirements/widgetsnbextension-3.6.0-py2.py3-none-any.whl b/env/requirements/widgetsnbextension-3.6.0-py2.py3-none-any.whl
deleted file mode 100644
index 4ad756f..0000000
--- a/env/requirements/widgetsnbextension-3.6.0-py2.py3-none-any.whl
+++ /dev/null
Binary files differ
diff --git a/env/requirements/widgetsnbextension-3.6.1-py2.py3-none-any.whl b/env/requirements/widgetsnbextension-3.6.1-py2.py3-none-any.whl
new file mode 100644
index 0000000..9ea5712
--- /dev/null
+++ b/env/requirements/widgetsnbextension-3.6.1-py2.py3-none-any.whl
Binary files differ
diff --git a/requirements.txt b/requirements.txt
index 5f59e49..cdae0d6 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,10 +1,10 @@
-numpy
-Flask
-pika
-protobuf
-tqdm
-opencv-python
-pyserial
-bitarray
-open3d
-protoc-wheel-0
\ No newline at end of file
+numpy==1.23.0
+Flask==2.1.2
+pika==1.2.1
+protobuf==4.21.1
+tqdm==4.64.0
+opencv-python==4.6.0.66
+pyserial==3.5
+bitarray==2.5.1
+protoc-wheel-0==21.1
+open3d==0.14.1
\ No newline at end of file
diff --git a/src/Server.py b/src/Server.py
index 9194321..5818ae8 100644
--- a/src/Server.py
+++ b/src/Server.py
@@ -1,10 +1,9 @@
import sys # NOQA: E402
import os
-from flask import Flask
+from flask import Flask, request
import pika
import uuid
import queue
-from werkzeug.wrappers import request
from threading import Thread
from datetime import datetime
@@ -13,6 +12,7 @@
from .engine.DeviceManager import DeviceManager
from .engine import MessageThreads as ths
from .protos import aerial_pb2 as pb
+from .utils.tools import update_folder_for_today
@@ -35,9 +35,22 @@
self.WEBport = WEBport
self.save_folder = save_folder
self.server_name = server_name
-
- # self.route('/api', methods=['POST'])(self.api)
+ today = datetime.now().strftime('%Y-%m-%d')
+ today_folder = os.path.join(self.save_folder,today)
+
+ self.today = {"today":today,
+ "today_folder":today_folder,
+ "sup_folder":self.save_folder,
+ "sub_folder": ""}
+
+ self.task_uuid_folder = {"init":os.path.join(self.today["today_folder"],"init")}
+
+ if not os.path.exists(self.task_uuid_folder["init"]):
+ os.makedirs(self.task_uuid_folder["init"])
+
+
+ self.route('/api', methods=['POST'])(self.api)
self.route('/', methods=['GET'])(self.hello)
# self._def_routes()
@@ -51,7 +64,7 @@
def _connect_to_MQ(self):
credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
self.parameters = pika.ConnectionParameters(
- credentials=credentials, host=self.MQhost)
+ credentials=credentials, host=self.MQhost,heartbeat=0)
# 杩炴帴闃熷垪鏈嶅姟鍣�
try:
self.connection = pika.BlockingConnection(self.parameters)
@@ -110,7 +123,7 @@
self.MsgDecoder_Thread = ths.decodeThread(
queue_list= queue_list,
queue_tags= queue_tags,
- save_path= self.save_folder
+ task_folder= self.task_uuid_folder
)
self.MsgDecoder_Thread.setDaemon(True)
@@ -135,6 +148,52 @@
def hello(self):
return 'Hello, This is acquisition demo API!'
+ def api(self):
+
+ try:
+ success, response_bytes = self.verify_commands(request.data)
+ except:
+ response_bytes = b'command error!'
+
+ response = Flask.response_class(
+ response_bytes,
+ mimetype='application/x-bytes'
+ )
+ return response,200
+
+
+ def verify_commands(self,input_command:bytes):
+
+ # parse the command from bytes to the pb object
+ req_command = pb.CommandMessage()
+ req_command.ParseFromString(input_command)
+
+ # extracting some info for use
+ source = req_command.source
+ # generatetime = req_command.generatetime
+ # syscommand = req_command.syscommand
+ # subcommand = req_command.subcommand
+ targetclient = req_command.targetclient
+ # returnserver = req_command.returnserver
+ taskuuid = req_command.taskuuid
+
+ update_folder_for_today(self.today)
+ task_path = os.path.join(self.today["today_folder"],taskuuid)
+
+ self.task_uuid_folder[taskuuid] = task_path
+
+ if not os.path.exists(task_path):
+ os.makedirs(task_path)
+
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ print(f"[x] {curr_time}|(From {source}): task [{taskuuid}] received and sending it to [{targetclient}]." )
+
+ self.CommandsSender_queue.put([targetclient,input_command])
+ response_bytes = bytes(f"Receive command success! The data will be save in [{os.path.abspath(task_path)}]",\
+ encoding="utf8")
+
+ return True, response_bytes
+
def run_rabbitmq(self):
self._connect_to_MQ()
self._receiver_Thread_Launcher()
diff --git a/src/engine/AcquisitionMain.py b/src/engine/AcquisitionMain.py
index e2a580c..af152a0 100644
--- a/src/engine/AcquisitionMain.py
+++ b/src/engine/AcquisitionMain.py
@@ -1,7 +1,9 @@
import sys # NOQA: E402
import os
+from time import sleep
import pika
import queue
+from datetime import datetime
# from .CaptureThread import CaptureThread
# from .ControllThread import ControllThread
@@ -14,7 +16,7 @@
# src_path = os.path.abspath(src_path)
# sys.path.insert(0, src_path)
-# from ..protos import aerial_pb2 as pb
+from ..protos import aerial_pb2 as pb
IMAGE_WIDTH = 3840
@@ -116,7 +118,7 @@
def _connect_to_MQ(self):
credentials = pika.PlainCredentials(self.MQuser, self.MQpassword)
self.parameters = pika.ConnectionParameters(
- credentials=credentials, host=self.MQhost)
+ credentials=credentials, host=self.MQhost,heartbeat=0)
# 杩炴帴闃熷垪鏈嶅姟鍣�
try:
self.connection = pika.BlockingConnection(self.parameters)
@@ -133,7 +135,7 @@
MQparams=self.parameters,
exchange_name='commands_topic',
exchange_type='topic',
- MQqueue_name="server0_commands_queue",
+ MQqueue_name=f"{self.device_id}_commands_queue",
binding_keys=[f"{self.device_id}"],
message_queue=self.CommandsReceiver_queue
)
@@ -206,19 +208,80 @@
timeout=self.timeout,
logs_queue=self.LogsSender_queue,
device_id=self.device_id,
- routing_key="server0"
+ init_binding_key=self.device_id,
+ init_routing_key="server0"
)
self.mc = microwaveController(
MicrowaveHost=self.MICROWAVEhost,
MicrowavePort=self.MICROWAVEport,
init_packet_threshold=100,
- init_timeusage_threshold=3,
+ init_timeusage_threshold=0.2,
device_id=self.device_id,
logs_queue=self.LogsSender_queue,
plys_queue=self.PlysSender_queue,
- routing_key="server0"
+ init_binding_key=self.device_id,
+ init_routing_key="server0"
)
+
+ def command_control(self):
+ while True:
+
+ try:
+ while not self.CommandsReceiver_queue.empty():
+ command_bytes= self.CommandsReceiver_queue.get()
+
+ # parse the command from bytes to the pb object
+ req_command = pb.CommandMessage()
+ req_command.ParseFromString(command_bytes)
+
+ # extracting some info for use
+ source = req_command.source
+ generatetime = req_command.generatetime
+ syscommand = req_command.syscommand
+ subcommand = req_command.subcommand
+ targetclient = req_command.targetclient
+ returnserver = req_command.returnserver
+ taskuuid = req_command.taskuuid
+
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ print(f"[x] {curr_time}|(From {source}): task [{taskuuid}] received!" )
+
+ # command for camera
+ if syscommand == pb.SysCommand.CAMERA_COMMAND:
+ # capture image
+ if subcommand == pb.SubCommand.ONE_TIME_CAPTURE:
+ self.cc.capture(routing_key=returnserver,taskuuid=taskuuid)
+ # unknown command
+ else:
+ pass
+
+ # command for microwave device
+ elif syscommand == pb.SysCommand.MICROWAVE_COMMAND:
+ if subcommand == pb.SubCommand.ONE_CIRCLE_SCAN:
+ self.mc.scan(routing_key=returnserver,taskuuid=taskuuid)
+ # unknown command
+ else:
+ pass
+
+ # command for dust cleaner
+ elif syscommand == pb.Syscommand.DUSTCLEANER_COMMAND:
+ if subcommand == pb.SubCommand.ONE_CLEAN_SCAN:
+ self.dc.clean(routing_key=returnserver,taskuuid=taskuuid)
+ # unknown command
+ else:
+ pass
+ # unknown command
+ else:
+ pass
+
+ else:
+ sleep(1)
+
+ except:
+
+ break
+
def run_rabbitmq(self):
@@ -236,6 +299,8 @@
self.run_rabbitmq()
# connecting to devices and wrap data to the corressponding queue
self.run_device()
+ sleep(10)
+ self.command_control()
def stop_all(self):
@@ -254,7 +319,6 @@
# self.stop_all()
# print("main thread end.")
self.run_all()
- self.CommandsReceiver_Thread.join()
return 0
diff --git a/src/engine/MessageThreads.py b/src/engine/MessageThreads.py
index 998a3e2..8cc2344 100644
--- a/src/engine/MessageThreads.py
+++ b/src/engine/MessageThreads.py
@@ -67,20 +67,20 @@
message = self.message_queue.get()
msg_routing_key = message[0]
msg_body = message[1]
+ if msg_body is not False:
+ # publish a PERSISTENT message
+ channel.basic_publish(
+ exchange = self.exchange_name,
+ routing_key = msg_routing_key,
+ body = msg_body,
+ properties=pika.BasicProperties(
+ delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
+ )
+ )
- # publish a PERSISTENT message
- channel.basic_publish(
- exchange = self.exchange_name,
- routing_key = msg_routing_key,
- body = msg_body,
- properties=pika.BasicProperties(
- delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
- )
- )
-
- timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
- caption = f"[x] {timestamp} | Sending message to [{msg_routing_key}]"
- print(caption)
+ timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ caption = f"[x] {timestamp} | Sending message to [{msg_routing_key}]"
+ print(caption)
else:
time.sleep(1)
@@ -143,20 +143,21 @@
if self.is_stop:
ch.close()
- self.message_queue.put(body)
+ if body is not False:
+ self.message_queue.put(body)
ch.basic_ack(delivery_tag = method.delivery_tag)
class decodeThread(threading.Thread):
def __init__(self,
queue_list:List[queue.Queue],
queue_tags:List[str],
- save_path:str):
+ task_folder:dict[str,str]):
super().__init__()
self.queue_list = queue_list
self.queue_tags = queue_tags
self.queue_num = 0
- self.save_path = save_path
+ self.task_folder = task_folder
self.is_stop = False
assert len(self.queue_list) == len(self.queue_tags)
@@ -167,22 +168,22 @@
def run(self) -> None:
- # creating a path for saving data
- time_tag = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
- self.save_root_folder = os.path.join(self.save_path,time_tag)
- os.makedirs(self.save_root_folder)
+ # # creating a path for saving data
+ # time_tag = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ # self.save_root_folder = os.path.join(self.task_folder,time_tag)
+ # os.makedirs(self.save_root_folder)
- self.logsmsg_folder = os.path.join(self.save_root_folder,"LogsMsg")
- self.imagesmsg_folder = os.path.join(self.save_root_folder,"ImagesMsg")
- self.plysmsg_folder = os.path.join(self.save_root_folder,"PlysMsg")
- self.commandsmsg_folder = os.path.join(self.save_root_folder,"CommandsMsg")
+ # self.logsmsg_folder = os.path.join(self.save_root_folder,"LogsMsg")
+ # self.imagesmsg_folder = os.path.join(self.save_root_folder,"ImagesMsg")
+ # self.plysmsg_folder = os.path.join(self.save_root_folder,"PlysMsg")
+ # self.commandsmsg_folder = os.path.join(self.save_root_folder,"CommandsMsg")
- sub_folder_list = [self.logsmsg_folder,self.imagesmsg_folder,
- self.plysmsg_folder,self.commandsmsg_folder]
+ # sub_folder_list = [self.logsmsg_folder,self.imagesmsg_folder,
+ # self.plysmsg_folder,self.commandsmsg_folder]
- for folder in sub_folder_list:
- os.mkdir(folder)
- print(f"[NOTICE] Creating folder: {folder}")
+ # for folder in sub_folder_list:
+ # os.mkdir(folder)
+ # print(f"[NOTICE] Creating folder: {folder}")
# process the queue list
self.queue_num = len(self.queue_list)
@@ -221,16 +222,19 @@
msg_pb = pb.LogsMessage()
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)
-
# process the msg
msg_source = msg_pb.source
msg_timestamp = msg_pb.timestamp
msg_context = msg_pb.context
+ taskuuid = msg_pb.taskuuid
+
+ save_folder = self.task_folder[taskuuid]
+
+ # save the raw pb msg
+ # curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ raw_file_name = os.path.join(save_folder,f"{msg_timestamp}-{msg_source}.LogsMsg")
+ with open(raw_file_name, 'wb') as file:
+ file.write(msg)
# display the info to the console
curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
@@ -248,15 +252,18 @@
msg_pb = pb.ImagesMessage()
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.imagesmsg_folder,f"{curr_time}.ImagesMsg")
- with open(raw_file_name, 'wb') as file:
- file.write(msg)
-
# process the msg
msg_source = msg_pb.source
msg_timestamp = msg_pb.timestamp
+ taskuuid = msg_pb.taskuuid
+ save_folder = self.task_folder[taskuuid]
+
+ # save the raw pb msg
+ # curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ raw_file_name = os.path.join(save_folder,f"{msg_timestamp}-{msg_source}.ImagesMsg")
+ with open(raw_file_name, 'wb') as file:
+ file.write(msg)
+
# process the images
id = 0
@@ -268,8 +275,8 @@
if (image_byte is None) | (len(image_byte) != imagelegth ):
pass
else:
- filename = f"{msg_source}_{msg_timestamp}_image{id}.jpg"
- filepath = os.path.join(self.imagesmsg_folder,filename)
+ filename = f"{msg_timestamp}-{msg_source}-image{id}.jpg"
+ filepath = os.path.join(save_folder,filename)
with open(filepath, 'wb') as f:
f.write(image_byte)
@@ -289,15 +296,18 @@
msg_pb = pb.PlysMessage()
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.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
+ taskuuid = msg_pb.taskuuid
+
+ save_folder = self.task_folder[taskuuid]
+
+ # save the raw pb msg
+ # curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ raw_file_name = os.path.join(save_folder,f"{msg_timestamp}-{msg_source}.PlysMsg")
+ with open(raw_file_name, 'wb') as file:
+ file.write(msg)
# process the muti ply data
id = 0
@@ -317,13 +327,14 @@
data_array = np.frombuffer(data_byte,dtype=dt)
data_array = data_array.reshape(rows,cols,order=order)
# save the data with .ply format
- filename = f"{msg_source}_{msg_timestamp}_pointcloud{id}"
- write_ply(data_array,self.plysmsg_folder,filename)
+ filename = f"{msg_timestamp}-{msg_source}-pointcloud{id}"
+ write_ply(data_array,save_folder,filename)
# next ply item
id += 1
def decode_CommandsMsg(self,queue):
+
raise NotImplementedError("NOT IMPLEMENT!")
@@ -332,15 +343,18 @@
def send_logs_tool(logs:List[str],
logs_queue:queue.Queue,
source:str,
- routing_key:str):
+ routing_key:str,
+ taskuuid:str = "init"):
for i, log in enumerate(logs):
if log is not None:
# constract a logs message
logsMsg = pb.LogsMessage()
logsMsg.source = source
- logsMsg.timestamp = f"{time.asctime()}"
+ curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ logsMsg.timestamp = curr_time
logsMsg.context = log
+ logsMsg.taskuuid = taskuuid
# convert the proto object to string
logsMsg_bytes = logsMsg.SerializeToString()
# add logsmsg to queue
@@ -350,13 +364,15 @@
def send_images_tool(images:List[np.array],
images_queue:queue.Queue,
source:str,
- routing_key:str):
+ routing_key:str,
+ taskuuid:str = "init"):
# constract a images message
imagesMsg = pb.ImagesMessage()
imagesMsg.source = source
curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
imagesMsg.timestamp = curr_time
+ imagesMsg.taskuuid = taskuuid
for img in images:
if img is not None:
@@ -377,13 +393,17 @@
def send_pointcloud_tool(pointclouds:queue.Queue,
plys_queue:queue.Queue,
source:str,
- routing_key:str):
+ routing_key:str,
+ taskuuid:str = "init"):
# constract a plys message
plysMsg = pb.PlysMessage()
plysMsg.source = source
curr_time = datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
plysMsg.timestamp = curr_time
+ plysMsg.taskuuid = taskuuid
+
+ empty = True
run =True
@@ -393,6 +413,7 @@
value = pointclouds.get()
if value is not False:
+ empty = False
pointcloud = np.array(value)
# get the attributes of the PointCloudItem
rows = pointcloud.shape[0]
@@ -412,13 +433,19 @@
item.data = data
else:
- # converting the pb ply msg to bytes
- plysMsg_bytes = plysMsg.SerializeToString()
- # add the plysMsg to the queue
- plys_queue.put([routing_key,plysMsg_bytes])
- # end the outer while loop and triminate this function
+ if not empty:
+ # converting the pb ply msg to bytes
+ plysMsg_bytes = plysMsg.SerializeToString()
+ # add the plysMsg to the queue
+ plys_queue.put([routing_key,plysMsg_bytes])
+ # end the outer while loop and triminate this function
+
+ else:
+ print("[x] NO poindcloud has been captured!")
run = False
+ else:
+ time.sleep(3)
diff --git a/src/engine/devices.py b/src/engine/devices.py
index 837cc3c..e4c72a9 100644
--- a/src/engine/devices.py
+++ b/src/engine/devices.py
@@ -24,7 +24,8 @@
logs_queue:queue.Queue,
images_queue:queue.Queue,
init_routing_key:str,
- init_binding_key:str
+ init_binding_key:str,
+ init_taskuuid:str = "init"
):
self.IMAGE_WIDTH = IMAGE_WIDTH
@@ -36,6 +37,8 @@
self.routing_key = init_routing_key
self.init_routing_key = init_routing_key
self.binding_key = init_binding_key
+ self.init_taskuuid = init_taskuuid
+ self.taskuuid = init_taskuuid
self._check_cameras()
@@ -68,12 +71,14 @@
send_logs_tool(logs=logs,
logs_queue=self.logs_queue,
source=self.device_id,
- routing_key=self.routing_key+".logs.test0")
+ routing_key=self.routing_key+".logs.test0",
+ taskuuid=self.taskuuid)
send_images_tool(images=images,
images_queue=self.images_queue,
source=self.device_id,
- routing_key=self.routing_key+".images.test0")
+ routing_key=self.routing_key+".images.test0",
+ taskuuid=self.taskuuid)
def _capture_image(self,mode:str = "capture"):
@@ -118,7 +123,7 @@
if TEST:
self.working_ports.append(dev_port)
if is_reading:
- self.camera_work[dev_port] == True
+ self.camera_work[dev_port] = True
log_list.append(log)
@@ -127,14 +132,18 @@
return log_list, image_list
- def capture(self,routing_key:str):
+ def capture(self,routing_key:str, taskuuid:str):
self.routing_key = routing_key
+ self.taskuuid = taskuuid
(logs,images) = self._capture_image()
- self._generate_msg = (logs,images)
+
+ # generate the msg and put it to queues
+ self._generate_msg(logs,images)
self.routing_key = self.init_routing_key
+ self.taskuuid = self.init_taskuuid
@@ -144,16 +153,23 @@
MicrowaveHost: str,
MicrowavePort: int,
init_packet_threshold: Union[int, None],
- init_timeusage_threshold: Union[int, None],
+ init_timeusage_threshold: Union[int, float, None],
device_id:str,
logs_queue:queue.Queue,
plys_queue:queue.Queue,
- routing_key:str,
+ init_routing_key:str,
+ init_binding_key:str,
+ init_taskuuid:str = "init"
):
super().__init__(MicrowaveHost, MicrowavePort)
self.device_id = device_id
self.plys_queue = plys_queue
- self.routing_key = routing_key
+
+ self.routing_key = init_routing_key
+ self.init_routing_key = init_routing_key
+ self.binding_key = init_binding_key
+ self.init_taskuuid = init_taskuuid
+ self.taskuuid = init_taskuuid
# testing the device when instance the class
self._get_pointcloud(init_packet_threshold,init_timeusage_threshold)
@@ -168,16 +184,34 @@
send_pointcloud_tool(pointclouds=self.pointcloud_queue,
plys_queue=self.plys_queue,
source=self.device_id,
- routing_key=self.routing_key+".plys.test0")
+ routing_key=self.routing_key+".plys.test0",
+ taskuuid=self.taskuuid)
+
+
+ def scan(self,routing_key:str,taskuuid:str):
+
+ # self.routing_key = routing_key
+ # self.taskuuid = self.init_taskuuid
+
+ # self._generate_msg()
+
+ raise NotImplementedError
####################################### DUST CLEANER #########################################
class dustCleanerController(DustCleaner):
- def __init__(self, com, bps, timeout, logs_queue, device_id,routing_key):
+ def __init__(self, com, bps, timeout, logs_queue, device_id,
+ init_routing_key:str,
+ init_binding_key:str,
+ init_taskuuid:str = "init"):
super().__init__(com, bps, timeout)
self.logs_queue = logs_queue
self.device_id = device_id
- self.routing_key = routing_key
+ self.routing_key = init_routing_key
+ self.init_routing_key = init_routing_key
+ self.binding_key = init_binding_key
+ self.init_taskuuid = init_taskuuid
+ self.taskuuid = init_taskuuid
self._check_cleaner()
@@ -198,3 +232,8 @@
logs_queue=self.logs_queue,
source = self.device_id,
routing_key=self.routing_key+".logs.test0")
+
+
+ def clean(self,routing_key:str,taskuuid:str):
+
+ raise NotImplementedError
diff --git a/src/microwave/servers.py b/src/microwave/servers.py
index c678379..47b15a4 100644
--- a/src/microwave/servers.py
+++ b/src/microwave/servers.py
@@ -258,7 +258,8 @@
if self.udp_get_stop.isSet():
print(f"{time.asctime()} | UDP-Get server is stopped by thresholds or keyborad interrupt.")
self.udp_queue.put(False)
- super().shutdown()
+ self._BaseServer__shutdown_request = True
+ # self.__is_shut_down.set()
diff --git a/src/protos/aerial.proto b/src/protos/aerial.proto
index c03a45b..d3ff8e6 100644
--- a/src/protos/aerial.proto
+++ b/src/protos/aerial.proto
@@ -1,125 +1,159 @@
syntax = "proto3";
-enum SysCommand {
- UNKOWN_COMMAND = 0; // 鏈煡鍛戒护
- SYS_COMMAND = 1; // 绯荤粺鍛戒护
- SENSOR_COMMAND = 2; // 浼犳劅鍣ㄥ懡浠�
- FIVE_G_COMMAND = 3; // 閫氳5G妯″潡鍛戒护
- CAMERA_COMMAND = 4; // 鎽勫儚澶村弬鏁板懡浠�
- IO_COMMAND = 5; // 澶栬IO鍙ｆ搷浣滃懡浠�
- SYS_PARAMS_COMMAND = 6; // 绯荤粺鍙傛暟鑾峰彇
- WIFI_COMMAND = 7; // 璁惧WIFI鍙傛暟璁剧疆
- KERNEL_COMMAND = 8; // 鏍稿績璁＄畻鍙傛暟璁�
- DEVICE_COMMAND = 9; // 宸ヤ綔璁惧鍙傛暟璁剧疆
- MICRO_WAVE_COMMAND = 10; // 姣背娉�
-}
+// enum SysCommand {
+// UNKOWN_COMMAND = 0; // 鏈煡鍛戒护
+// SYS_COMMAND = 1; // 绯荤粺鍛戒护
+// SENSOR_COMMAND = 2; // 浼犳劅鍣ㄥ懡浠�
+// FIVE_G_COMMAND = 3; // 閫氳5G妯″潡鍛戒护
+// CAMERA_COMMAND = 4; // 鎽勫儚澶村弬鏁板懡浠�
+// IO_COMMAND = 5; // 澶栬IO鍙ｆ搷浣滃懡浠�
+// SYS_PARAMS_COMMAND = 6; // 绯荤粺鍙傛暟鑾峰彇
+// WIFI_COMMAND = 7; // 璁惧WIFI鍙傛暟璁剧疆
+// KERNEL_COMMAND = 8; // 鏍稿績璁＄畻鍙傛暟璁�
+// DEVICE_COMMAND = 9; // 宸ヤ綔璁惧鍙傛暟璁剧疆
+// MICRO_WAVE_COMMAND = 10; // 姣背娉�
+// }
-enum SubCommand {
- option allow_alias = true;
+// enum SubCommand {
+// option allow_alias = true;
- UNKOWN_SUB_COMMAND = 0; // 鏈煡瀛愬懡浠�
+// UNKOWN_SUB_COMMAND = 0; // 鏈煡瀛愬懡浠�
- // 绯荤粺鍛戒护
- POWER_ON_SUB_COMMAND = 1; // 寮�鏈�
- POWER_OFF_SUB_COMMAND = 2; // 鍏虫満
+// // 绯荤粺鍛戒护
+// POWER_ON_SUB_COMMAND = 1; // 寮�鏈�
+// POWER_OFF_SUB_COMMAND = 2; // 鍏虫満
- // 浼犳劅鍣ㄥ懡浠�
- SENSOR_INIT_SUB_COMMAND = 1; // 浼犳劅鍣ㄥ垵濮嬪寲
- SENSOR_REQURE_SUB_COMMAND = 2; // 浼犳劅鍣ㄥ弬鏁拌幏鍙�
- RTK_SETTING_SUB_COMMAND = 3; // RTK澶栬鍙傛暟璁剧疆
- RTK_REQURE_SUB_COMMAND = 4; // RTK澶栬鏁版嵁鑾峰彇
+// // 浼犳劅鍣ㄥ懡浠�
+// SENSOR_INIT_SUB_COMMAND = 1; // 浼犳劅鍣ㄥ垵濮嬪寲
+// SENSOR_REQURE_SUB_COMMAND = 2; // 浼犳劅鍣ㄥ弬鏁拌幏鍙�
+// RTK_SETTING_SUB_COMMAND = 3; // RTK澶栬鍙傛暟璁剧疆
+// RTK_REQURE_SUB_COMMAND = 4; // RTK澶栬鏁版嵁鑾峰彇
- // 閫氳5G妯″潡鍛戒护
- FIVE_G_INIT_SUB_COMMAND = 1; // 5G妯″潡鍙傛暟鍒濆鍖�
- FIVE_G_SEND_IMAGE_SUB_COMMAND = 2; // 5G閫氳鍥惧儚鏁版嵁鍙戦��
- FIVE_G_REQURE_SUB_COMMAND = 3; // 5G閫氳鎺ユ敹
+// // 閫氳5G妯″潡鍛戒护
+// FIVE_G_INIT_SUB_COMMAND = 1; // 5G妯″潡鍙傛暟鍒濆鍖�
+// FIVE_G_SEND_IMAGE_SUB_COMMAND = 2; // 5G閫氳鍥惧儚鏁版嵁鍙戦��
+// FIVE_G_REQURE_SUB_COMMAND = 3; // 5G閫氳鎺ユ敹
- // 鎽勫儚澶村弬鏁板懡浠�
- CAMERA_NUMBER_SUB_COMMAND = 1; // 璁惧鎽勫儚澶翠釜鏁�
- CAMERA_RESOLUTION_SUB_COMMAND = 2; // 璁惧鎽勫儚澶村伐浣滃垎杈ㄧ巼
- CAMERA_FRAME_RATE_SUB_COMMAND = 3; // 鎽勫儚澶村伐浣滃抚鐜�
- CAMERA_OTHER_PARAMS_SUB_COMMAND = 4; // 鎽勫儚澶村伐浣滃叾浠栧弬鏁�
+// // 鎽勫儚澶村弬鏁板懡浠�
+// CAMERA_NUMBER_SUB_COMMAND = 1; // 璁惧鎽勫儚澶翠釜鏁�
+// CAMERA_RESOLUTION_SUB_COMMAND = 2; // 璁惧鎽勫儚澶村伐浣滃垎杈ㄧ巼
+// CAMERA_FRAME_RATE_SUB_COMMAND = 3; // 鎽勫儚澶村伐浣滃抚鐜�
+// CAMERA_OTHER_PARAMS_SUB_COMMAND = 4; // 鎽勫儚澶村伐浣滃叾浠栧弬鏁�
- // 澶栬IO鍙ｆ搷浣滃懡浠�
- CAMERA_DUST_REMOVAL_START_SUB_COMMAND = 1; // 鎽勫儚澶撮櫎灏樺惎鍔�
- CAMERA_CONTROL_START_SUB_COMMAND = 2; // 澶栬鎺у埗鏁版嵁鍚姩锛�18璺帶鍒讹級
+// // 澶栬IO鍙ｆ搷浣滃懡浠�
+// CAMERA_DUST_REMOVAL_START_SUB_COMMAND = 1; // 鎽勫儚澶撮櫎灏樺惎鍔�
+// CAMERA_CONTROL_START_SUB_COMMAND = 2; // 澶栬鎺у埗鏁版嵁鍚姩锛�18璺帶鍒讹級
- // 绯荤粺鍙傛暟鑾峰彇
- CPU_FREQUENCY_SUB_COMMAND = 1; // CPU涓婚宸ヤ綔鏁版嵁
- CPU_KERNAL_TEMP_SUB_COMMAND = 2; // CPU鏍稿績娓╁害鏁版嵁
- CPU_BETTERY_DATA_SUB_COMMAND = 3; // 鐢垫睜渚涚數鐢甸噺鏁版嵁
+// // 绯荤粺鍙傛暟鑾峰彇
+// CPU_FREQUENCY_SUB_COMMAND = 1; // CPU涓婚宸ヤ綔鏁版嵁
+// CPU_KERNAL_TEMP_SUB_COMMAND = 2; // CPU鏍稿績娓╁害鏁版嵁
+// CPU_BETTERY_DATA_SUB_COMMAND = 3; // 鐢垫睜渚涚數鐢甸噺鏁版嵁
- // 璁惧WIFI鍙傛暟璁剧疆
- WIFI_HOTSPOT_SETTING_SUB_COMMAND = 1; // WIFI鐑偣璁剧疆
- WIFI_HOTSPOT_START_STOP_SUB_COMMAND = 2; // WIFI鐑偣寮�鍚拰鍏抽棴
+// // 璁惧WIFI鍙傛暟璁剧疆
+// WIFI_HOTSPOT_SETTING_SUB_COMMAND = 1; // WIFI鐑偣璁剧疆
+// WIFI_HOTSPOT_START_STOP_SUB_COMMAND = 2; // WIFI鐑偣寮�鍚拰鍏抽棴
- // 鏍稿績璁＄畻鍙傛暟璁剧疆
- POINT_CLOUD_ACCURACY_SUB_COMMAND = 1; // 鐐逛簯璁＄畻绮惧害
- SPLICING_PARAMS_SUB_COMMAND = 2; // 鎷兼帴鍙傛暟
- MAP_INTERFACE_PLATFORM_SUB_COMMAND = 3; // 鍦板浘鎺ュ彛骞冲彴
- DATA_STORE_BACKUP_SUB_COMMAND = 4; // 鏁版嵁瀛樺偍澶囦唤
+// // 鏍稿績璁＄畻鍙傛暟璁剧疆
+// POINT_CLOUD_ACCURACY_SUB_COMMAND = 1; // 鐐逛簯璁＄畻绮惧害
+// SPLICING_PARAMS_SUB_COMMAND = 2; // 鎷兼帴鍙傛暟
+// MAP_INTERFACE_PLATFORM_SUB_COMMAND = 3; // 鍦板浘鎺ュ彛骞冲彴
+// DATA_STORE_BACKUP_SUB_COMMAND = 4; // 鏁版嵁瀛樺偍澶囦唤
- // 宸ヤ綔璁惧鍙傛暟璁剧疆
- SYSTEM_TIMER_SUB_COMMAND = 1; // 绯荤粺宸ヤ綔瀹氭椂
- SYSTEM_COORD_HEIGHT_SUB_COMMAND = 2; // 宸ヤ綔鍖哄煙鍧愭爣涓庨珮搴�
- TRIGGER_CONDITION_SUB_COMMAND = 3; // 宸ヤ綔瑙﹀彂鏉′欢
- GRAVITY_FAILURE_SUB_COMMAND = 4; // 閲嶅姏鍔犻�熷害澶辨晥鍙傛暟
- NINE_AXIS_FAILURE_SUB_COMMAND = 5; // 涔濊酱鍔犻�熷害澶辨晥鍙傛暟
+// // 宸ヤ綔璁惧鍙傛暟璁剧疆
+// SYSTEM_TIMER_SUB_COMMAND = 1; // 绯荤粺宸ヤ綔瀹氭椂
+// SYSTEM_COORD_HEIGHT_SUB_COMMAND = 2; // 宸ヤ綔鍖哄煙鍧愭爣涓庨珮搴�
+// TRIGGER_CONDITION_SUB_COMMAND = 3; // 宸ヤ綔瑙﹀彂鏉′欢
+// GRAVITY_FAILURE_SUB_COMMAND = 4; // 閲嶅姏鍔犻�熷害澶辨晥鍙傛暟
+// NINE_AXIS_FAILURE_SUB_COMMAND = 5; // 涔濊酱鍔犻�熷害澶辨晥鍙傛暟
- // 姣背娉�
- MICRO_WAVE_SUB_COMMAND = 1; // 姣背娉�
+// // 姣背娉�
+// MICRO_WAVE_SUB_COMMAND = 1; // 姣背娉�
+// }
+
+// message DataField {
+// int32 length = 1; // 鏁版嵁闀垮害
+// bytes data = 2; // 鏁版嵁鍐呭
+// int32 checksum = 3; // 鏍￠獙鐮�
+// }
+
+// message CapturedData {
+// int64 capture_time = 1; // 閲囬泦鏃堕棿
+// string capture_device =2; // 閲囬泦璁惧
+// int32 capture_port = 3; // 閲囬泦璁惧搴忓彿
+// DataField DataField = 4;
+// string data_type = 5; // 鏁版嵁绫诲瀷 image udp
+// }
+
+// message DataMessage{
+// int64 sending_time = 1;
+// string sending_device = 2;
+// string target_device = 3;
+// CapturedData CapturedData = 5;
+// }
+
+// message StateData {
+// int64 state_time = 1;
+// string state_device = 2;
+// DataField DataField = 4;
+// }
+
+// message StateMessage {
+// int64 sending_time = 1;
+// string sending_device = 2;
+// string target_device = 3;
+// CapturedData StatedData = 5;
+// }
+
+// message CommandData {
+// SysCommand SysCommand = 1;
+// SubCommand SubCommand = 2;
+// }
+
+// message CommandMessage {
+// int64 sending_time = 1;
+// string sending_device = 2;
+// string target_device = 3;
+// CommandData CommandData =4;
+// }
+
+enum SysCommand{
+ UNKOWN_COMMAND = 0;
+ CAMERA_COMMAND = 1;
+ MICROWAVE_COMMAND = 2;
+ DUSTCLEANER_COMMAND = 3;
+
}
-message DataField {
- int32 length = 1; // 鏁版嵁闀垮害
- bytes data = 2; // 鏁版嵁鍐呭
- int32 checksum = 3; // 鏍￠獙鐮�
+enum SubCommand{
+ UNKOWN_SUBCOMMAND = 0;
+
+ // camera params
+ ONE_TIME_CAPTURE = 1;
+
+ // microwave params
+ ONE_CIRCLE_SCAN = 2;
+
+ // dust cleaner params
+ ONE_CLEAN_SCAN = 4;
+
}
-message CapturedData {
- int64 capture_time = 1; // 閲囬泦鏃堕棿
- string capture_device =2; // 閲囬泦璁惧
- int32 capture_port = 3; // 閲囬泦璁惧搴忓彿
- DataField DataField = 4;
- string data_type = 5; // 鏁版嵁绫诲瀷 image udp
-}
-message DataMessage{
- int64 sending_time = 1;
- string sending_device = 2;
- string target_device = 3;
- CapturedData CapturedData = 5;
-}
-
-message StateData {
- int64 state_time = 1;
- string state_device = 2;
- DataField DataField = 4;
-}
-
-message StateMessage {
- int64 sending_time = 1;
- string sending_device = 2;
- string target_device = 3;
- CapturedData StatedData = 5;
-}
-
-message CommandData {
- SysCommand SysCommand = 1;
- SubCommand SubCommand = 2;
-}
-
-message CommandMessage {
- int64 sending_time = 1;
- string sending_device = 2;
- string target_device = 3;
- CommandData CommandData =4;
+message CommandMessage{
+ string source = 1;
+ string generatetime = 2;
+ SysCommand syscommand = 3;
+ SubCommand subcommand = 4;
+ string targetclient = 5;
+ string returnserver = 6;
+ string taskuuid = 7;
}
message LogsMessage{
string source = 1;
string timestamp = 2;
string context = 3;
+ optional string taskuuid = 4;
}
@@ -135,6 +169,7 @@
}
repeated ImageItems Image = 3;
+ optional string taskuuid = 4;
}
@@ -156,6 +191,7 @@
}
repeated PointCloudItem Plydata = 3;
+ optional string taskuuid = 4;
// enum ref_system
// {
diff --git a/src/utils/tools.py b/src/utils/tools.py
index 8f9fa00..1d1cfc5 100644
--- a/src/utils/tools.py
+++ b/src/utils/tools.py
@@ -1,8 +1,8 @@
-from email import header
+
import os, sys
import numpy as np
import time
-from tqdm import tqdm
+from datetime import datetime
def file_fliter(folder:str, file_ext:str, sort_format = None):
@@ -76,10 +76,27 @@
header="".join(header),
comments="",
fmt=fmt)
- print(f"{time.asctime()} {name}.ply have been saved with {n_p} points!")
+ print(f"{time.asctime()} | {name}.ply have been saved with {n_p} points!")
+def update_folder_for_today(folder_tree:dict[str,str]):
+ """
+ "sup_folder" + "today" +"sub_folder" >> "today_folder"
+ """
+ try:
+ sup = folder_tree["sup_folder"]
+ sub = folder_tree["sub_folder"]
+
+ today = datetime.now().strftime('%Y-%m-%d')
+ today_folder = os.path.join(sup,today,sub)
+
+ folder_tree["today"] = today
+ folder_tree["today_folder"] = today_folder
+ except Exception as e:
+ raise ValueError(f"{e}")
+
+
if __name__ == "__main__":
ply_array = read_ply('src\\microwave\\demo\\demoData\\2522.ply')
--
Gitblit v1.9.1