| | |
|---|
| | | |
|---|
| | | from tkinter import Frame |
|---|
| | | from .components import * |
|---|
| | | from .constants import * |
|---|
| | | from ctypes import sizeof, string_at |
|---|
| | | |
|---|
| | | |
|---|
| | | perDetectionBytes = 8 |
|---|
| | | perTrackerBytes = 32 |
|---|
| | | |
|---|
| | | class MicroWaveParser: |
|---|
| | | @classmethod |
|---|
| | | def ParserFrame(cls, buffer: bytes) -> list: |
|---|
| | | """ |
|---|
| | | 解析完整的一帧数据，数据包括数据头，数据体，数据尾，返回点云数组 |
|---|
| | | 解析完整的帧数据，数据包括数据头，数据体，数据尾，返回点云数组 |
|---|
| | | """ |
|---|
| | | header = HeaderBlock(buffer[:sizeof(HeaderBlock)]) |
|---|
| | | if HEADER_MAGIC == string_at(header.Magic, 8): |
|---|
| | | footer = FooterBlock(buffer[-sizeof(FooterBlock):]) |
|---|
| | | point_cloud = [] |
|---|
| | | for i in range(header.NumberOfDetection): |
|---|
| | | detection = DetectionBlock(header, footer, buffer[i * 8 + sizeof(HeaderBlock): (i + 1) * 8 + sizeof(HeaderBlock)]) |
|---|
| | | point = (detection.X, detection.Y, detection.Z) |
|---|
| | | point_cloud.append(point) |
|---|
| | | return point_cloud |
|---|
| | | else: |
|---|
| | | return [] |
|---|
| | | buffer_size = len(buffer) |
|---|
| | | frame_start = 0 |
|---|
| | | frame_end = 0 |
|---|
| | | frame_id = 0 |
|---|
| | | |
|---|
| | | ply_data_list = [] |
|---|
| | | |
|---|
| | | while frame_start < buffer_size: |
|---|
| | | # process the current frame |
|---|
| | | header = HeaderBlock(buffer[frame_start:frame_start+sizeof(HeaderBlock)]) |
|---|
| | | if HEADER_MAGIC == string_at(header.Magic, 8): |
|---|
| | | numsDetection = header.NumberOfDetection |
|---|
| | | numsTracker = header.NumberOfTrack |
|---|
| | | |
|---|
| | | # get the end point of the current frame |
|---|
| | | frame_end = frame_start + \ |
|---|
| | | sizeof(HeaderBlock) + \ |
|---|
| | | numsDetection * perDetectionBytes + \ |
|---|
| | | numsTracker * perTrackerBytes + \ |
|---|
| | | sizeof(FooterBlock) |
|---|
| | | |
|---|
| | | if frame_end <= buffer_size: |
|---|
| | | # get the frame data and footer data |
|---|
| | | frame = buffer[frame_start:frame_end] |
|---|
| | | footer = FooterBlock(frame[-sizeof(FooterBlock):]) |
|---|
| | | # get the cloud point data |
|---|
| | | point_cloud = [] |
|---|
| | | |
|---|
| | | for i in range(numsDetection): |
|---|
| | | |
|---|
| | | point_data = frame[i * 8 + sizeof(HeaderBlock): (i + 1) * 8 + sizeof(HeaderBlock)] |
|---|
| | | detection = DetectionBlock(header, footer, point_data) |
|---|
| | | point = (detection.X, detection.Y, detection.Z) |
|---|
| | | point_cloud.append(point) |
|---|
| | | |
|---|
| | | # append the point cloud data to the list |
|---|
| | | ply_data_list.append([frame_id, header.FrameNumber, point_cloud]) |
|---|
| | | |
|---|
| | | # update the parameters for next frame in the while loop |
|---|
| | | frame_id += 1 |
|---|
| | | frame_start = frame_end |
|---|
| | | else: |
|---|
| | | raise ValueError("wrong header magic code for microwave data") |
|---|
| | | |
|---|
| | | return ply_data_list |
|---|
