# Module `ammosreader.AbstractAmmosReader`

I provide a base class for specialized AmmosReaders.

Expand source code

    """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.AmmosConstants import FrameType

    logging.basicConfig(filename='ammos.log', level=logging.DEBUG)

    class AbstractAmmosReader(ABC):
        """I implement a base class for specialized AmmosReaders."""

        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):
            """I return the name of the original file."""
            return self.__file_name

        @property
        def ammos_file(self):
            """I return the file to read the data from."""
            return self.__ammos_file

        @property
        def container(self):
            """I return the container which stores the data read."""
            return self.__container

        @property
        def tags(self):
            """I return all the tags."""
            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
            """
            header_size = AmmosGlobalFrameHeader.HEADER_SIZE

            bytes = self.ammos_file.read(header_size)
            logging.info("Reading next global frame header")
            if ((not bytes) or (len(bytes) < header_size)):
                logging.debug("Can not read all %s bytes of global frame header", header_size)
                return None

            current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(bytes)

            if current_global_frame_header is None:
                return None

            # 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()

            if global_frame_header is None:
                return None

            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

## Classes

`class AbstractAmmosReader (file_name)`

  I implement a base class for specialized AmmosReaders.

  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

  Expand source code

      class AbstractAmmosReader(ABC):
          """I implement a base class for specialized AmmosReaders."""

          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):
              """I return the name of the original file."""
              return self.__file_name

          @property
          def ammos_file(self):
              """I return the file to read the data from."""
              return self.__ammos_file

          @property
          def container(self):
              """I return the container which stores the data read."""
              return self.__container

          @property
          def tags(self):
              """I return all the tags."""
              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
              """
              header_size = AmmosGlobalFrameHeader.HEADER_SIZE

              bytes = self.ammos_file.read(header_size)
              logging.info("Reading next global frame header")
              if ((not bytes) or (len(bytes) < header_size)):
                  logging.debug("Can not read all %s bytes of global frame header", header_size)
                  return None

              current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(bytes)

              if current_global_frame_header is None:
                  return None

              # 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()

              if global_frame_header is None:
                  return None

              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

  ### Ancestors

  - abc.ABC

  ### Subclasses

  - AmmosAudioReader
  - AmmosIFReader

  ### Instance variables

  `var ammos_file`

    I return the file to read the data from.

    Expand source code

        @property
        def ammos_file(self):
            """I return the file to read the data from."""
            return self.__ammos_file

  `var container`

    I return the container which stores the data read.

    Expand source code

        @property
        def container(self):
            """I return the container which stores the data read."""
            return self.__container

  `var file_name`

    I return the name of the original file.

    Expand source code

        @property
        def file_name(self):
            """I return the name of the original file."""
            return self.__file_name

  `var tags`

    I return all the tags.

    Expand source code

        @property
        def tags(self):
            """I return all the tags."""
            return self.__tags

  ### Methods

  `def add_tag(self, a_key, a_value)`

    I add information to tags using a key/value pair.

    Expand source code

        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 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

    Expand source code

        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_body(self, data_header_length)`

    My descendents have to implement this.

    Expand source code

        @abstractmethod
        def read_next_global_frame_body(self, data_header_length):
            """My descendents have to implement this."""
            pass

  `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

    Expand source code

        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
            """
            header_size = AmmosGlobalFrameHeader.HEADER_SIZE

            bytes = self.ammos_file.read(header_size)
            logging.info("Reading next global frame header")
            if ((not bytes) or (len(bytes) < header_size)):
                logging.debug("Can not read all %s bytes of global frame header", header_size)
                return None

            current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(bytes)

            if current_global_frame_header is None:
                return None

            # logging.info("Current global frame header %s", current_global_frame_header)
            return current_global_frame_header

  `def read_next_single_frame(self)`

    I read and return a single global frame.

    :return: a single global frame :rtype: AmmosSingleFrame

    Expand source code

        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()

            if global_frame_header is None:
                return None

            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

  `def rewind_to_start(self)`

    I set the file pointer to the beginning of the file for the next operation.

    Expand source code

        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)