import queue
|
from typing import Union
|
import threading
|
|
from .MessageThreads import send_logs_tool
|
from ..microwave.servers import MicroWaveUDPThread,UDP2PointcloudThread
|
|
|
class micorwaveMannager(object):
|
|
def __init__(self,
|
MicrowaveHost:str,
|
MicrowavePort:int,
|
):
|
|
# microwave
|
self.MicrowaveHost = MicrowaveHost
|
self.MicrowavePort = MicrowavePort
|
# udp get
|
self._udp_get_thread = None
|
# self.packet_threshold = packet_threshold
|
# self.timeusage_threshold = timeusage_threshold
|
self.udp_queue = queue.Queue()
|
self.udp_get_stop = threading.Event()
|
self._udp_get_thread_name = "Thread-UDPGet"
|
# udp to pointcloud
|
self._udp2pointcloud_thread = None
|
self.pointcloud_queue = queue.Queue()
|
self._udp2pointcloud_thread_name = "Thread-UDP2PointCloud"
|
|
|
def _udp_get_thread_launcher(self,
|
packet_threshold:Union[int, None],
|
timeusage_threshold:Union[int, None]):
|
self.udp_get_stop.clear()
|
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=packet_threshold,
|
timeusage_threshold=timeusage_threshold)
|
self._udp_get_thread.setDaemon(True)
|
self._udp_get_thread.setName(self._udp_get_thread_name)
|
self._udp_get_thread.start()
|
|
|
def _udp2pointcloud_thread_launcher(self):
|
self._udp2pointcloud_thread = UDP2PointcloudThread(
|
udp_queue = self.udp_queue,
|
pointcloud_queue=self.pointcloud_queue
|
)
|
self._udp2pointcloud_thread.setDaemon(True)
|
self._udp2pointcloud_thread.setName(self._udp2pointcloud_thread_name)
|
self._udp2pointcloud_thread.start()
|
|
|
|
