Index: ammosreader/AbstractAmmosReader.py =================================================================== --- ammosreader/AbstractAmmosReader.py (revision ef16c0b623d6d10da5f763419185ae9e5e77929b) +++ ammosreader/AbstractAmmosReader.py (revision ef16c0b623d6d10da5f763419185ae9e5e77929b) @@ -0,0 +1,150 @@ +"""I provide a base class for specialized AmmosReaders.""" +import logging + +from abc import ABC, abstractmethod + +from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader +from ammosreader.AmmosSingleFrame import AmmosSingleFrame +from ammosreader.AmmosContainer import AmmosContainer +from ammosreader.CAConstants import FrameType + + +class AbstractAmmosReader(ABC): + """I implement a base class for specialized AmmosReaders.""" + + GLOBAL_HEADER_SIZE = 24 # 8 words + + def __init__(self, file_name): + """ + I am the standard constructor for Ammos Readers. + + Additional information about the file can be added as key/value pairs in tags + + :param file_name: The file to read Ammos data from + :type file_name: str + """ + self.__file_name = file_name + self.__ammos_file = open(self.file_name, "rb") + self.__container = AmmosContainer(self.file_name, []) + self.__tags = {} + + @property + def file_name(self): + return self.__file_name + + @property + def ammos_file(self): + return self.__ammos_file + + @property + def container(self): + return self.__container + + @property + def tags(self): + return self.__tags + + def add_tag(self, a_key, a_value): + """I add information to tags using a key/value pair.""" + assert a_key not in self.__tags + self.__tags[a_key] = a_value + + def rewind_to_start(self): + """I set the file pointer to the beginning of the file for the next operation.""" + self.ammos_file.seek(0) + + def read_all_frames_left(self): + """ + I read all remaining frames into my container until end of file is reached. + + :return: a container containing all frames read + :rtype: AmmosContainer + """ + frames_read = 0 + while True: + logging.info("Reading single frame %s ...", frames_read) + current_frame = self.read_next_single_frame() + if current_frame is not None: + frames_read += 1 + self.container.add_frame(current_frame) + else: + logging.debug("Frame: %s incomplete", frames_read+1) + break + logging.info("%s frames read", len(self.container.global_frames)) + return self.container + + def read_next_global_frame_header(self): + """ + I return the next global frame header read from current position in file. + + :return: the next global frame header or None if incomplete + :rtype: AmmosGlobalFrameHeader + """ + bytes = self.ammos_file.read(AbstractAmmosReader.GLOBAL_HEADER_SIZE) + logging.info("Reading next global frame header") + if ((not bytes) or (len(bytes) < AbstractAmmosReader.GLOBAL_HEADER_SIZE)): + logging.debug("Can not read all %s bytes of global frame header", AbstractAmmosReader.GLOBAL_HEADER_SIZE) + return None + + # FIXME: Catch exceptions and add some asserts + current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(bytes) + logging.info("Current global frame header %s", current_global_frame_header) + return current_global_frame_header + + @abstractmethod + def read_next_global_frame_body(self, data_header_length): + """My descendents have to implement this.""" + pass + + def read_next_single_frame(self): + """ + I read and return a single global frame. + + :return: a single global frame + :rtype: AmmosSingleFrame + """ + global_frame_header = self.read_next_global_frame_header() + + # print(global_frame_header) + + if global_frame_header is None: + logging.debug("Global frame header missing") + return None + + if global_frame_header.data_header_length is None: + logging.debug("Data header length empty") + return None + + if global_frame_header.frame_type not in list(FrameType): + logging.info("Unknown frame type %s found", global_frame_header.frame_type) + return None + + # FIXME: Refactor duplicate logic - This code stinks + + global_frame_body = None + + if_data_types = [FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT.value, + FrameType.IF_DATA_16BIT_REAL_IMAGINARY_FIXEDPOINT.value, + FrameType.IF_DATA_16BIT_REAL_REAL.value, + FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FIXEDPOINT_RESCALED.value, + FrameType.IF_DATA_32BIT_REAL_IMAGINARY_FLOATINGPOINT_RESCALED.value] + + if global_frame_header.frame_type == FrameType.AUDIO_DATA.value: + logging.info("Audio Datastream found") + global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length) + if global_frame_body is None: + return None + return AmmosSingleFrame(global_frame_header, global_frame_body) + + if global_frame_header.frame_type in if_data_types: + logging.info("IF Datastream found") + global_frame_body = self.read_next_global_frame_body(global_frame_header.data_header_length) + if global_frame_body is None: + return None + return AmmosSingleFrame(global_frame_header, global_frame_body) + + if global_frame_body is None: + return None + + logging.info("Unsupported frame type %s found", global_frame_header.frame_type) + return None Index: ammosreader/AmmosAudioReader.py =================================================================== --- ammosreader/AmmosAudioReader.py (revision b41e9756e2adb758e54e40cd73c97eeae3787875) +++ ammosreader/AmmosAudioReader.py (revision ef16c0b623d6d10da5f763419185ae9e5e77929b) @@ -1,19 +1,14 @@ -"""I parse an R&S AMMOS recording.""" +"""I provide a specialized Ammos Reader for audio data.""" -import os - +from ammosreader.AbstractAmmosReader import AbstractAmmosReader from ammosreader.AmmosGlobalFrameBody import AmmosGlobalFrameBody from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader from ammosreader.AmmosExtendedAudioDataHeader import AmmosExtendedAudioDataHeader -from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader -from ammosreader.AmmosSingleFrame import AmmosSingleFrame -from ammosreader.AmmosContainer import AmmosContainer from ammosreader.AmmosAudioDataBlock import AmmosAudioDataBlock -class AmmosAudioReader(): +class AmmosAudioReader(AbstractAmmosReader): """I read the audio data embedded in an R&S AMMOS recording.""" - GLOBAL_HEADER_SIZE = 24 # 8 words STANDARD_AUDIO_DATA_HEADER_SIZE = 36 # 9 words EXTENDED_AUDIO_DATA_HEADER_SIZE = 44 # 11 words @@ -26,65 +21,5 @@ :type file_name: str """ - self.file_name = file_name - self.file = open(self.file_name, "rb") - self.file_size = os.path.getsize(self.file_name) - - self.container = AmmosContainer(self.file_name, []) - - self.tags = {} - - def rewind_to_start(self): - """I set the file pointer to the beginning of the file for the next operation.""" - self.file.seek(0) - - def add_tag(self, tag): - """ - I add a tag to my tag list. - - :param tag: The tag to add to my tag list - :type tag: dict - """ - self.tags[tag.key] = tag.value - - def read_all_frames_left(self): - """ - I read all remaining frames into my container until end of file is reached. - - :return: a container containing all frames read - :rtype: AmmosContainer - """ - frames_read = 0 - while True: - print("Reading single frame", frames_read, '...') - current_frame = self.read_next_single_frame() - if current_frame is not None: - frames_read += 1 - self.container.add_frame(current_frame) - if frames_read % 10000 == 0: - print("#", end="") - else: - print("Frame:", frames_read+1, " incomplete") - break - - print(len(self.container.global_frames), "frames read") - return self.container - - def read_next_global_frame_header(self): - """ - I return the next global frame header read from current position in file. - - :return: the next global frame header or None if incomplete - :rtype: AmmosGlobalFrameHeader - """ - bytes = self.file.read(AmmosAudioReader.GLOBAL_HEADER_SIZE) - print("Reading next global frame header") - if ((not bytes) or (len(bytes) < AmmosAudioReader.GLOBAL_HEADER_SIZE)): - print("Can not read all", AmmosAudioReader.GLOBAL_HEADER_SIZE, "bytes of global frame header") - return None - - # FIXME: Catch exceptions and add some asserts - current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(bytes) - # print("Current global frame header", current_global_frame_header) - return current_global_frame_header + super().__init__(file_name) def read_next_global_frame_body_data_header(self): @@ -97,5 +32,5 @@ :rtype: AmmosAudioDataHeader """ - bytes = self.file.read(AmmosAudioReader.STANDARD_AUDIO_DATA_HEADER_SIZE) + bytes = self.ammos_file.read(AmmosAudioReader.STANDARD_AUDIO_DATA_HEADER_SIZE) # print("\nReading global frame body standard data header\n") @@ -113,5 +48,5 @@ :rtype: AmmosExtendedAudioDataHeader """ - bytes = self.file.read(AmmosAudioReader.EXTENDED_AUDIO_DATA_HEADER_SIZE) + bytes = self.ammos_file.read(AmmosAudioReader.EXTENDED_AUDIO_DATA_HEADER_SIZE) if ((not bytes) or (len(bytes) < AmmosAudioReader.EXTENDED_AUDIO_DATA_HEADER_SIZE)): @@ -141,5 +76,5 @@ total = sample_count*channel_count*sample_size - byte_string = self.file.read(total) + byte_string = self.ammos_file.read(total) if len(byte_string) != total: @@ -149,13 +84,18 @@ return AmmosAudioDataBlock(byte_string, channel_count, sample_count, sample_size) - def read_next_global_frame_body(self, global_frame_header): + def read_next_global_frame_body(self, data_header_length): + """ + I return the next global frame body read from current position in file. + :param global_frame_header: + """ audio_data_header = None - if global_frame_header.data_header_length == AmmosAudioReader.STANDARD_AUDIO_DATA_HEADER_SIZE: + if data_header_length == AmmosAudioReader.STANDARD_AUDIO_DATA_HEADER_SIZE: print("Read standard data header") audio_data_header = self.read_next_global_frame_body_data_header() - if global_frame_header.data_header_length == AmmosAudioReader.EXTENDED_AUDIO_DATA_HEADER_SIZE: + print("Data header length", data_header_length) + if data_header_length == AmmosAudioReader.EXTENDED_AUDIO_DATA_HEADER_SIZE: print("Read extended data header") audio_data_header = self.read_next_global_frame_body_extended_data_header() @@ -175,30 +115,14 @@ return AmmosGlobalFrameBody(audio_data_header, audio_data_body) - def read_next_single_frame(self): + def pcm_for_channel(self, a_channel): + """ + I return the raw pcm audio data for a given channel. - global_frame_header = self.read_next_global_frame_header() + :param a_channel: the channel I have to extract + :type a_channel: int - print(global_frame_header) - - if global_frame_header is None: - print("Global frame header missing") - return None - - if global_frame_header.data_header_length is None: - print("Data header length empty") - return None - - if global_frame_header.frame_type == 256: - print("Audio Datastream found") - global_frame_body = self.read_next_global_frame_body(global_frame_header) - if global_frame_body is None: - return None - else: - print("Unsupported frame type", global_frame_header.frame_type, "found") - return None - - ammos_single_frame = AmmosSingleFrame(global_frame_header, global_frame_body) - return ammos_single_frame - - def pcm_for_channel(self, a_channel): + :rtype: bytes + """ + for each in self.container.global_frames: + print(each.global_frame_body) return b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel) for each in self.container.global_frames]) Index: ammosreader/AmmosIFReader.py =================================================================== --- ammosreader/AmmosIFReader.py (revision b41e9756e2adb758e54e40cd73c97eeae3787875) +++ ammosreader/AmmosIFReader.py (revision ef16c0b623d6d10da5f763419185ae9e5e77929b) @@ -1,4 +1,3 @@ -import math -import os +"""I provide a specialized Ammos Reader for IF data.""" from ammosreader.AmmosGlobalFrameBody import AmmosGlobalFrameBody @@ -13,49 +12,16 @@ class AmmosIFReader(): + STANDARD_IF_DATA_HEADER_SIZE = 56 + EXTENDED_IF_DATA_HEADER_SIZE = 76 + def __init__(self, file_name): - - self.file_name = file_name - self.file = open(self.file_name, "rb") - - self.container = AmmosContainer(self.file_name, []) - - self.tags = [] - - def add_tag(self, tag): - self.tags.append(tag) - - def read_all_frames_left(self): - - frames_read = 0 - - while True: - # print("Reading single frame", frames_read, '...') - current_frame = self.read_next_single_frame() - if current_frame is not None: - frames_read += 1 - self.container.add_frame(current_frame) - # if frames_read % 10000 == 0: - # print("#", end="") - else: - # print("Frame:", frames_read+1, " incomplete") - break - - # print(len(self.container.global_frames), "frames read") - - def read_next_global_frame_header(self): - bytes = self.file.read(24) - # print("Reading next global frame header") - if ((not bytes) or (len(bytes) < 24)): - # print("Can not read all 24 bytes of global frame header") - return None - - return AmmosGlobalFrameHeader.from_bytes(bytes) + super.__init__(file_name) def read_next_global_frame_body_data_header(self): - bytes = self.file.read(56) + 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) < 56)): + 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 @@ -67,8 +33,8 @@ def read_next_global_frame_body_extended_data_header(self): - bytes = self.file.read(76) + 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) < 76)): + 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 @@ -87,5 +53,5 @@ total = n*block_length - byte_string = self.file.read(block_length) + byte_string = self.ammos_file.read(block_length) if len(byte_string) != total: @@ -99,11 +65,11 @@ return data_blocks - def read_next_global_frame_body(self, global_frame_header): + def read_next_global_frame_body(self, data_header_length): if_data_header = None - if global_frame_header.data_header_length == 56: + if data_header_length == AmmosIFReader.STANDARD_IF_DATA_HEADER_SIZE: if_data_header = self.read_next_global_frame_body_data_header() - else: + if data_header_length == AmmosIFReader.EXTENDED_IF_DATA_HEADER_SIZE: if_data_header = self.read_next_global_frame_body_extended_data_header() @@ -119,29 +85,2 @@ return AmmosGlobalFrameBody(if_data_header, if_data_body) - - def read_next_single_frame(self): - - global_frame_header = self.read_next_global_frame_header() - - # print("\nReading next global frame header\n", global_frame_header) - # print("File pointer", self.file.tell()) - - if global_frame_header is None: - # print("Global frame header missing") - return None - - if global_frame_header.data_header_length is None: - # print("Data header length empty") - return None - - if global_frame_header.frame_type == 2: - - global_frame_body = self.read_next_global_frame_body(global_frame_header) - if global_frame_body is None: - return None - - else: - # print("Unsupported frame type", global_frame_header.frame_type, "found") - return None - - return AmmosSingleFrame(global_frame_header, global_frame_body) Index: ammosreader/CAConstants.py =================================================================== --- ammosreader/CAConstants.py (revision b41e9756e2adb758e54e40cd73c97eeae3787875) +++ ammosreader/CAConstants.py (revision ef16c0b623d6d10da5f763419185ae9e5e77929b) @@ -1,5 +1,5 @@ """I provide several constants used in R&S software.""" -import Enum +from enum import Enum @@ -40,3 +40,4 @@ SCAN_TUNING = 0x4002 SCAN_LEVEL_TUNING = 0x4003 - DDF_RESERVED_RANGE = range(0x5000, 0x5100) + DDF_RESERVED_START = 0x5000 + DDF_RESERVED_END = 0x50FF