udp_get rebuild

wu_xinjun

2022-06-16 14cc8f11a79fabec83f3c5754af45583fb4edcd0

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