Module ammosreader.AbstractAmmosReader

I provide a base class for specialized AmmosReaders.

"""I provide a base class for specialized AmmosReaders."""

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
from ammosreader import logger


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:
            logger.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:
                logger.debug("Frame: %s incomplete", frames_read+1)
                break
        logger.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

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

        current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(in_bytes)

        if current_global_frame_header is None:
            return None

        # logger.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:
            logger.debug("Global frame header missing")
            return None

        if global_frame_header.data_header_length is None:
            logger.debug("Data header length empty")
            return None

        if global_frame_header.frame_type not in list(FrameType):
            logger.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:
            logger.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:
            logger.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

        logger.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:
            logger.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:
                logger.debug("Frame: %s incomplete", frames_read+1)
                break
        logger.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

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

        current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(in_bytes)

        if current_global_frame_header is None:
            return None

        # logger.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:
            logger.debug("Global frame header missing")
            return None

        if global_frame_header.data_header_length is None:
            logger.debug("Data header length empty")
            return None

        if global_frame_header.frame_type not in list(FrameType):
            logger.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:
            logger.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:
            logger.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

        logger.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:
        logger.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:
            logger.debug("Frame: %s incomplete", frames_read+1)
            break
    logger.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

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

    current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(in_bytes)

    if current_global_frame_header is None:
        return None

    # logger.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:
        logger.debug("Global frame header missing")
        return None

    if global_frame_header.data_header_length is None:
        logger.debug("Data header length empty")
        return None

    if global_frame_header.frame_type not in list(FrameType):
        logger.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:
        logger.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:
        logger.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

    logger.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)