udp_get rebuild

wu_xinjun

2022-06-16 14cc8f11a79fabec83f3c5754af45583fb4edcd0

udp_get rebuild

 src/engine/Microwave.py

New file
@@ -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



    



    

 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):

 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
    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!")