From 14cc8f11a79fabec83f3c5754af45583fb4edcd0 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期四, 16 六月 2022 17:11:21 +0800
Subject: [PATCH] udp_get rebuild
---
src/microwave/servers.py | 127 ++++++++++++++++++++----
src/microwave/deviceTools.py | 67 ++++++-------
src/engine/Microwave.py | 55 +++++++++++
3 files changed, 190 insertions(+), 59 deletions(-)
diff --git a/src/engine/Microwave.py b/src/engine/Microwave.py
new file mode 100644
index 0000000..a59cfed
--- /dev/null
+++ b/src/engine/Microwave.py
@@ -0,0 +1,55 @@
+import os,sys
+import queue
+import time
+from datetime import datetime
+from typing import Union
+
+from .MessageThreads import send_logs_tool
+from ..microwave.servers import MicroWaveUDPThread
+
+
+class micorwaveMannager(object):
+
+ def __init__(self,
+ MicrowaveHost:str,
+ MicrowavePort:int,
+ packet_threshold:Union[int, None],
+ timeusage_threshold:Union[int, None]
+
+
+ ):
+
+ # microwave
+ self.MicrowaveHost = MicrowaveHost
+ self.MicrowavePort = MicrowavePort
+ # udp get
+ self.packet_threshold = packet_threshold
+ self.timeusage_threshold = timeusage_threshold
+ self.udp_queue = queue.Queue()
+ self.udp_get_stop = {"value":False}
+ self._udp_get_thread_name = "Thread-UDPGet"
+ # udp to pointcloud
+
+
+ def _udp_get_thread_launcher(self):
+ self._udp_get_thread = MicroWaveUDPThread(MicrowaveHost=self.MicrowaveHost,
+ MicrowavePort=self.MicrowavePort,
+ udp_queue=self.udp_queue,
+ udp_get_stop=self.udp_get_stop,
+ packet_threshold=self.packetThreshold,
+ timeusage_threshold=self.timeusage_threshold)
+ self._udp_get_thread.setDaemon(True)
+ self._udp_get_thread.setName(self._udp_get_thread_name)
+
+
+ def _udp2pointcloud_thread_launcher(self):
+
+ pass
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/src/microwave/deviceTools.py b/src/microwave/deviceTools.py
index 7031f4f..e57379a 100644
--- a/src/microwave/deviceTools.py
+++ b/src/microwave/deviceTools.py
@@ -1,4 +1,5 @@
import datetime
+import queue
from socketserver import BaseRequestHandler, UDPServer
import time
import os, sys
@@ -7,42 +8,17 @@
import numpy as np
from ..utils import file_fliter, write_ply
-from .servers import MicroWaveReceiver
+from .servers import MicroWaveReceiver,MicroWaveUDPThread
from .parsers import MicroWaveParser
-# # 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(1, src_path)
-
-# udp_get
-class PointCloudUDPRequestHandler(BaseRequestHandler):
-
- def handle(self):
- packet_threshold = self.server.packet_threshold
- packed_count = self.server.packed_count
- path = self.server.path
-
- print(time.asctime(),' Got connection from {}, {}'.format(self.client_address, packed_count))
- # Get message and client socket
- msg, sock = self.request
- time_tag = datetime.datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
- f = open('{}/{}_packet{}.udp'.format(path,time_tag,packed_count), 'wb')
- f.write(msg)
- f.close()
- self.server.packed_count += 1
- self.server.udp_list.append(msg)
-
- if packed_count >= packet_threshold:
- UDPServer.server_close(self.server)
def udp_get_tool(microwaveHost,packetThreshold,deviceName,saveFolder,subFolder):
# host = ''
# host = '192.168.2.65' # use this host to get the udp data from microwave device
port = 9911
- server =UDPServer((microwaveHost, port), PointCloudUDPRequestHandler)
-
+ udp_queue = queue.Queue()
+ udp_get_stop = {"value": False}
if subFolder is not None:
path = os.path.join(saveFolder,subFolder)
@@ -52,23 +28,42 @@
if not os.path.exists(path):
os.makedirs(path)
- server.path = os.path.abspath(path)
- server.packet_threshold = packetThreshold
- server.packed_count = 0
- server.udp_list = []
+ udp_thread = MicroWaveUDPThread(MicrowaveHost=microwaveHost,
+ MicrowavePort=port,
+ udp_queue=udp_queue,
+ udp_get_stop=udp_get_stop,
+ packet_threshold=packetThreshold,
+ timeusage_threshold=None)
+
+ udp_thread.setDaemon(True)
+ udp_thread.setName("UDPGet_Thread")
print(time.asctime(),'Server started on port', port)
print('....')
print('ctrl-c to quit server.')
try:
- server.serve_forever()
+ udp_thread.start()
+
+ run = True
+
+ while run:
+ packed_count = 0
+ while run or not udp_queue.empty():
+ value = udp_queue.get()
+ while value is not False:
+ time_tag = datetime.datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
+ with ('{}/{}_packet{}.udp'.format(path,time_tag,packed_count), 'wb') as f:
+ f.write(udp_queue.get())
+ packed_count += 1
+ else:
+ run = False
+
except KeyboardInterrupt:
- server.server_close()
+ udp_get_stop = {"value": True}
print(time.asctime(),"Server Stopped")
except:
- server.server_close()
+ udp_get_stop = {"value": True}
print(time.asctime(),"Server Stopped")
- return server.udp_list
# udp2bin
def udp2bin_tool(udp_folder_Or_list):
diff --git a/src/microwave/servers.py b/src/microwave/servers.py
index 8686b4a..fec7edd 100644
--- a/src/microwave/servers.py
+++ b/src/microwave/servers.py
@@ -1,10 +1,13 @@
import os, sys
from ctypes import sizeof, string_at
-from socketserver import BaseRequestHandler, UDPServer
+from socketserver import _AddressType, _RequestType, BaseRequestHandler, BaseServer, UDPServer
+import threading
+from typing import Union
import warnings
+import time
+import queue
from .components import *
from .constants import *
-from ..protos import aerial_pb2 as pb
@@ -141,32 +144,110 @@
"""
micro wave udp request handler
"""
+ def __init__(self,
+ request: _RequestType,
+ client_address: _AddressType,
+ server: BaseServer) -> None:
+ super().__init__(request, client_address, server)
+
def handle(self):
# Get message and client socket
- buf, _ = self.request
- receiver = self.server.Receiver
- # add buffer to receiver
- receiver.AddBuf(buf)
- if receiver.DataFrameReady:
- # dataframe ready, add to udp_queue
- msg = pb.RequestCommand()
- msg.sys_command = pb.SysCommand.MICRO_WAVE_COMMAND
- msg.sub_command = pb.SubCommand.MICRO_WAVE_SUB_COMMAND
- data1 = pb.MicroWaveMessage()
- data1.length = len(receiver.DataFrame)
- data1.data = bytes(receiver.DataFrame)
- data1_str = data1.SerializeToString()
- msg.data1.length = len(data1_str)
- msg.data1.data = data1_str
- self.server.msg_queue.put(msg)
+ while not self.server.udp_get_stop["value"]:
+ buf, _ = self.request
+ packed_count = self.server.packed_count
+ print(f"{time.asctime()} | Got connection from {self.client_address}, {packed_count}")
+ self.server.udp_queue.put(buf)
+ self.server.packed_count += 1
+ else:
+ UDPServer.server_close(self.server)
+ self.server.udp_queue.put(False)
+ print(f"{time.asctime()} | Server is stopped by thresholds or keyborad interrupt.")
+
class MicroWaveUDPServer(UDPServer):
"""
micro wave udp server
"""
- def __init__(self, server_address, msg_queue):
- super().__init__(server_address, MicroWaveRequestHandler)
- # create receiver instance
- self.Receiver = MicroWaveReceiver()
- self.msg_queue = msg_queue
\ No newline at end of file
+ def __init__(self,
+ server_address: tuple[str,int],
+ RequestHandlerClass: MicroWaveRequestHandler,
+ udp_queue:queue.Queue,
+ udp_get_stop:dict,
+ packet_threshold:Union[int, None],
+ timeusage_threshold:Union[int, None]
+ ) -> None:
+
+ self.server_address = server_address
+ self.RequestHandlerClass = MicroWaveRequestHandler
+ self.udp_queue = udp_queue
+ self.udp_get_stop = udp_get_stop
+ self.packet_threshold = packet_threshold
+ self.timeusage_threshold = timeusage_threshold # mins
+ self.packed_count = 0
+
+ super().__init__(server_address, RequestHandlerClass)
+
+ # record start time of the server
+ while self.timeusage_threshold is not None:
+ self.start_time = time.time()
+
+ def close_request(self, request: _RequestType) -> None:
+
+ # check to stop the server via muti-thresdhold
+ while self.packet_threshold is not None and \
+ self.packed_count >= self.packet_threshold:
+
+ self.udp_get_stop["value"] = True
+
+ while self.timeusage_threshold is not None:
+ cur_time = time.time() # seconds
+ time_usage = int(cur_time - self.start_time)
+
+ if time_usage > 60*self.timeusage_threshold:
+ self.udp_get_stop["value"] = True
+
+ return super().close_request(request)
+
+
+
+class MicroWaveUDPThread(threading.Thread):
+ def __init__(self,
+ MicrowaveHost:str,
+ MicrowavePort:int,
+ udp_queue:queue.Queue,
+ udp_get_stop:dict,
+ packet_threshold:Union[int, None],
+ timeusage_threshold:Union[int, None]
+ ):
+ super().__init__()
+
+ self.MicrowaveHost = MicrowaveHost
+ self.MicrowavePort = MicrowavePort
+ self.udp_queue = udp_queue
+ self.udp_get_stop = udp_get_stop
+ self.packet_threshold = packet_threshold
+ self.timeusage_threshold = timeusage_threshold
+
+ def stop(self):
+ """stop the server to stop the """
+ self.udp_get_stop["value"] = True
+
+ def run(self):
+ server = MicroWaveUDPServer(server_address=(self.MicrowaveHost,self.MicrowavePort),
+ RequestHandlerClass=MicroWaveRequestHandler,
+ udp_queue=self.udp_queue,
+ udp_get_stop=self.udp_get_stop,
+ packet_threshold=self.packet_threshold,
+ timeusage_threshold=self.timeusage_threshold
+ )
+ try:
+ server.serve_forever()
+ except:
+ server.server_close()
+ print(time.asctime(),"UDP Getting Server Stopped with Unknown Reason!")
+
+
+
+
+
--
Gitblit v1.9.1