From da6ff0858f4997f79f5d8cdc88b7367a5cb710b1 Mon Sep 17 00:00:00 2001
From: wu_xinjun <bent20140204@outlook.com>
Date: 星期二, 07 六月 2022 10:55:26 +0800
Subject: [PATCH] Merge commit 'mqdev' into mqaddFeature
---
src/microwave/servers.py | 122 ++++++++++++++++++++++++----------------
1 files changed, 74 insertions(+), 48 deletions(-)
diff --git a/src/microwave/servers.py b/src/microwave/servers.py
index e9c315b..8686b4a 100644
--- a/src/microwave/servers.py
+++ b/src/microwave/servers.py
@@ -1,15 +1,10 @@
import os, sys
from ctypes import sizeof, string_at
from socketserver import BaseRequestHandler, UDPServer
+import warnings
from .components import *
from .constants import *
-
-# 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(0, src_path)
-
-from protos import aerial_pb2 as pb
+from ..protos import aerial_pb2 as pb
@@ -32,22 +27,61 @@
self.DataFrame = None
self.Header = None
self.Footer = None
+
+ def _is_handshake(self,data):
+ handshake = Handshake.from_buffer_copy(data)
+ if HAND_SHAKE_MAGIC == string_at(handshake.Magic, 8):
+ self.DataFrameReady = False
+ self.state = UDP_STATE_HAND_SHAKE
+ self.DataFrame = None
+ self.Header = None
+ self.Footer = None
+ return True
+ else:
+ return False
+
+ def _get_header(self, buf):
+ self.Header = HeaderBlock.from_buffer_copy(buf[:sizeof(HeaderBlock)])
+ if HEADER_MAGIC == string_at(self.Header.Magic, 8):
+ self.DataFrame = bytearray(buf)
+ N = self.Header.NumberOfDetection
+ Nt = self.Header.NumberOfTrack
+ self.PointCount = 0
+ self.TrackCount = 0
+ self.DataLength = N * 8 + Nt * sizeof(TrackerBlock)
+ self.RemainLength = self.DataLength + sizeof(FooterBlock)
+ self.data_buf = bytearray()
+ self.footer_buf = bytearray()
+ self.Points = []
+ self.ReceivedDataLength = 0
+ self.ReceivedFooterLength = 0
+ # the next packet should be data block
+ self.state = UDP_STATE_DATA
+ return True
+ else:
+ False
def _receive_data(self, data) -> None:
"""
receive data packet from udp
"""
- self.data_buf.extend(data)
+ self.data_buf.extend(bytes(data))
self.ReceivedDataLength += len(data)
if self.ReceivedDataLength == self.DataLength:
# if received data length equal expected data length
# then we can get footer block
self.state = UDP_STATE_FOOTER
- return
+ return True
+ elif self.ReceivedDataLength > self.DataLength:
+ self.state = UDP_STATE_IDLE
+ warnings.warn(f"encounter an error when getting the detection data in udp packets ...")
+ return False
+ else:
+ return True
def _receive_footer(self, data) -> None:
# receive footer block
- self.footer_buf.extend(data)
+ self.footer_buf.extend(bytes(data))
self.ReceivedFooterLength += len(data)
if self.ReceivedFooterLength == sizeof(FooterBlock):
self.Footer = FooterBlock.from_buffer_copy(self.footer_buf)
@@ -56,57 +90,49 @@
# reset state to idle, ready for next data frame
self.state = UDP_STATE_IDLE
return
+
def AddBuf(self, buf: bytes) -> None:
# add a udp packet to receiver
buf_size = len(buf)
- if self.state == UDP_STATE_IDLE:
- if buf_size == sizeof(Handshake):
- handshake = Handshake.from_buffer_copy(buf)
- if HAND_SHAKE_MAGIC == string_at(handshake.Magic, 8):
- self.DataFrameReady = False
- self.state = UDP_STATE_HAND_SHAKE
- self.DataFrame = None
- self.Header = None
- self.Footer = None
- elif self.state == UDP_STATE_HAND_SHAKE:
- # after received handshake packet, we can get header block
- self.Header = HeaderBlock.from_buffer_copy(buf[:sizeof(HeaderBlock)])
- if HEADER_MAGIC == string_at(self.Header.Magic, 8):
- self.DataFrame = bytearray(buf)
- N = self.Header.NumberOfDetection
- Nt = self.Header.NumberOfTrack
- self.PointCount = 0
- self.TrackCount = 0
- self.DataLength = N * 8 + Nt * sizeof(TrackerBlock)
- self.RemainLength = self.DataLength + sizeof(FooterBlock)
- self.data_buf = bytearray()
- self.footer_buf = bytearray()
- self.Points = []
- self.ReceivedDataLength = 0
- self.ReceivedFooterLength = 0
- # the next packet should be data block
- self.state = UDP_STATE_DATA
+ if buf_size == sizeof(Handshake):
+ self._is_handshake(buf)
+ if self.state == UDP_STATE_HAND_SHAKE:
+ # after received handshake packet, we can get header block
+ try:
+ # make sure to get a complete header
+ result = self._get_header(buf)
+ except Exception as e:
+ warnings.warn(f"encounter an unknow error {e} when getting the header data in udp packets ...")
+ result = False
+
+ if not result:
+ # waiting for the next handshake udp packet
+ self.state = UDP_STATE_IDLE
+ else:
+ # process the detection data in the first data packets
header_size = sizeof(HeaderBlock)
- # 澶勭悊绗竴涓暟鎹寘涓殑鏁版嵁
data_size = min(buf_size - header_size, self.DataLength)
- self._receive_data(buf[header_size: header_size + data_size])
-
- # 鍒ゆ柇绗竴涓暟鎹寘涓槸鍚﹀寘鍚獸ooter Block
- if (header_size + data_size) < buf_size:
- self._receive_footer(buf[header_size + data_size])
+
+ if self._receive_data(buf[header_size: header_size + data_size]):
+
+ # 鍒ゆ柇绗竴涓暟鎹寘涓槸鍚﹀寘鍚獸ooter Block
+ if (header_size + data_size) < buf_size:
+ self._receive_footer(buf[header_size + data_size:])
+
elif self.state == UDP_STATE_DATA:
# receive data block
- self.DataFrame.extend(buf)
+ self.DataFrame.extend(bytes(buf))
remain_data_size = self.DataLength - self.ReceivedDataLength
data_size = min(buf_size, remain_data_size)
- self._receive_data(buf[: data_size])
- if data_size < buf_size:
- self._receive_footer(buf[data_size: ])
+ if self._receive_data(buf[: data_size]):
+ if data_size < buf_size:
+ self._receive_footer(buf[data_size: ])
+
elif self.state == UDP_STATE_FOOTER:
# receive footer block
- self.DataFrame.extend(buf)
+ self.DataFrame.extend(bytes(buf))
self._receive_footer(buf)
--
Gitblit v1.9.1