Module ammosreader.AbstractAmmosReader

I provide a base class for specialized AmmosReaders.

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