"""I provide a specialized Ammos Reader for IF data.""" from ammosreader.AmmosGlobalFrameBody import AmmosGlobalFrameBody from ammosreader.AmmosIFDataHeader import AmmosIFDataHeader from ammosreader.AmmosExtendedIFDataHeader import AmmosExtendedIFDataHeader from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader from ammosreader.AmmosSingleFrame import AmmosSingleFrame from ammosreader.AmmosIFDataBlock import AmmosIFDataBlock from ammosreader.AmmosContainer import AmmosContainer class AmmosIFReader(): STANDARD_IF_DATA_HEADER_SIZE = 56 EXTENDED_IF_DATA_HEADER_SIZE = 76 def __init__(self, file_name): super.__init__(file_name) def read_next_global_frame_body_data_header(self): bytes = self.ammos_file.read(AmmosIFReader.STANDARD_IF_DATA_HEADER_SIZE) # print("\nReading global frame body standard data header\n") if ((not bytes) or (len(bytes) < AmmosIFReader.STANDARD_IF_DATA_HEADER_SIZE)): # print("Can not read all 56 bytes of global frame body data header") return None data_header = AmmosIFDataHeader.from_bytes(bytes) # print("Data header", data_header) return data_header def read_next_global_frame_body_extended_data_header(self): bytes = self.ammos_file.read(AmmosIFReader.EXTENDED_IF_DATA_HEADER_SIZE) # print("\nReading global frame body extended data header\n") if ((not bytes) or (len(bytes) < AmmosIFReader.EXTENDED_IF_DATA_HEADER_SIZE)): # print("Can not read all ", 76, "bytes of global frame extended data header") return None extended_data_header = AmmosExtendedIFDataHeader.from_bytes(bytes) # print("Extended data header", extended_data_header) return extended_data_header def read_next_if_data_blocks(self, n, length): # FIXME: Describe the parameters better data_blocks = [] block_length = 4 + length total = n*block_length byte_string = self.ammos_file.read(block_length) if len(byte_string) != total: # print("Can not read all", total, "bytes of data body") return None for i in range(0, n): result = byte_string[i*block_length:(i*block_length+block_length)] data_blocks.append(AmmosIFDataBlock(result[0:4], result[4:])) return data_blocks def read_next_global_frame_body(self, data_header_length): if_data_header = None if data_header_length == AmmosIFReader.STANDARD_IF_DATA_HEADER_SIZE: if_data_header = self.read_next_global_frame_body_data_header() if data_header_length == AmmosIFReader.EXTENDED_IF_DATA_HEADER_SIZE: if_data_header = self.read_next_global_frame_body_extended_data_header() if if_data_header is None: # print("Data header missing") return None if_data_body = self.read_next_if_data_blocks(if_data_header.block_count, if_data_header.block_length) if if_data_body is None: # print("Data body missing") return None return AmmosGlobalFrameBody(if_data_header, if_data_body)