Module ammosreader.AmmosAudioReader

I provide a specialized Ammos Reader for audio data.

Expand source code
"""I provide a specialized Ammos Reader for audio data."""
import logging

from ammosreader.AbstractAmmosReader import AbstractAmmosReader
from ammosreader.AmmosGlobalFrameBody import AmmosGlobalFrameBody
from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader
from ammosreader.AmmosExtendedAudioDataHeader import AmmosExtendedAudioDataHeader
from ammosreader.AmmosAudioDataBody import AmmosAudioDataBody


class AmmosAudioReader(AbstractAmmosReader):
    """I read the audio data embedded in an R&S AMMOS recording."""

    def __init__(self, file_name):
        """
        I return an instance of AmmosAudioReader initialized with a given file name.

        :param file_name: the file to read from
        :type file_name: str
        """
        super().__init__(file_name)

    def read_next_global_frame_body_data_header(self):
        """
        I return the next global frame body data header from current position in file.

        :param data_header_size: the number of bytes to read
        :type data_header_size: int
        :return: the next Ammos Audio Data header or None if incomplete
        :rtype: AmmosAudioDataHeader
        """
        header_size = AmmosAudioDataHeader.HEADER_SIZE

        bytes = self.ammos_file.read(header_size)

        logging.info("\nReading global frame body standard data header\n")
        if ((not bytes) or (len(bytes) < header_size)):
            logging.debug("Can not read all %s bytes of global frame body data header", header_size)
            return None
        return AmmosAudioDataHeader.from_bytes(bytes)

    def read_next_global_frame_body_extended_data_header(self):
        """
        I return the next global frame body extended data header from current position in file.

        :return: the next Ammos Audio Extended Data header or None if incomplete
        :rtype: AmmosExtendedAudioDataHeader
        """
        header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE

        bytes = self.ammos_file.read(header_size)

        if ((not bytes) or (len(bytes) < header_size)):
            logging.debug("Can not read all %s bytes of global frame extended data header", header_size)
            return None
        return AmmosExtendedAudioDataHeader.from_bytes(bytes)

    def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
        """
        I return the next audio data read from current position in file.

        :param number_of_samples: the number of samples per channel inside data body
        :type number_of_samples: int

        :param number_of_channels: number of channels (e.g. mono, stereo or even more)
        :type number_of_channels: int

        :param sample_size: sample size in bytes (1, 2 or 4 bytes)
        :type sample_size: int

        :return: the next audio data or None if incomplete
        :rtype: bytes
        """
        total = number_of_samples*number_of_channels*sample_size

        byte_string = self.ammos_file.read(total)

        if len(byte_string) != total:
            logging.debug("Can not read all %s bytes of data body", total)
            return None
        return AmmosAudioDataBody(byte_string, number_of_channels, number_of_samples, sample_size)

    def read_next_global_frame_body(self, data_header_length):
        """
        I return the next global frame body read from current position in file.

        :param data_header_length: the length of the data header
        :type data_header_length: int
        """
        audio_data_header = None

        if data_header_length == AmmosAudioDataHeader.HEADER_SIZE:
            logging.info("Read standard data header")
            audio_data_header = self.read_next_global_frame_body_data_header()

        logging.info("Data header length %s", data_header_length)
        if data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE:
            logging.info("Read extended data header")
            audio_data_header = self.read_next_global_frame_body_extended_data_header()

        if audio_data_header is None:
            logging.debug("Data header missing or format unknown")
            return None

        audio_data_body = self.read_next_audio_data_body(audio_data_header.number_of_samples,
                                                         audio_data_header.number_of_channels,
                                                         audio_data_header.sample_size)

        if audio_data_body is None:
            logging.debug("Data body missing")
            return None

        return AmmosGlobalFrameBody(audio_data_header, audio_data_body)

    def pcm_for_channel(self, a_channel):
        """
        I return the raw pcm audio data for a given channel.

        :param a_channel: the channel I have to extract
        :type a_channel: int

        :rtype: bytes
        """
        return b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel) for each in self.container.global_frames])

Classes

class AmmosAudioReader (file_name)

I read the audio data embedded in an R&S AMMOS recording.

I return an instance of AmmosAudioReader initialized with a given file name.

:param file_name: the file to read from :type file_name: str

Expand source code
class AmmosAudioReader(AbstractAmmosReader):
    """I read the audio data embedded in an R&S AMMOS recording."""

    def __init__(self, file_name):
        """
        I return an instance of AmmosAudioReader initialized with a given file name.

        :param file_name: the file to read from
        :type file_name: str
        """
        super().__init__(file_name)

    def read_next_global_frame_body_data_header(self):
        """
        I return the next global frame body data header from current position in file.

        :param data_header_size: the number of bytes to read
        :type data_header_size: int
        :return: the next Ammos Audio Data header or None if incomplete
        :rtype: AmmosAudioDataHeader
        """
        header_size = AmmosAudioDataHeader.HEADER_SIZE

        bytes = self.ammos_file.read(header_size)

        logging.info("\nReading global frame body standard data header\n")
        if ((not bytes) or (len(bytes) < header_size)):
            logging.debug("Can not read all %s bytes of global frame body data header", header_size)
            return None
        return AmmosAudioDataHeader.from_bytes(bytes)

    def read_next_global_frame_body_extended_data_header(self):
        """
        I return the next global frame body extended data header from current position in file.

        :return: the next Ammos Audio Extended Data header or None if incomplete
        :rtype: AmmosExtendedAudioDataHeader
        """
        header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE

        bytes = self.ammos_file.read(header_size)

        if ((not bytes) or (len(bytes) < header_size)):
            logging.debug("Can not read all %s bytes of global frame extended data header", header_size)
            return None
        return AmmosExtendedAudioDataHeader.from_bytes(bytes)

    def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
        """
        I return the next audio data read from current position in file.

        :param number_of_samples: the number of samples per channel inside data body
        :type number_of_samples: int

        :param number_of_channels: number of channels (e.g. mono, stereo or even more)
        :type number_of_channels: int

        :param sample_size: sample size in bytes (1, 2 or 4 bytes)
        :type sample_size: int

        :return: the next audio data or None if incomplete
        :rtype: bytes
        """
        total = number_of_samples*number_of_channels*sample_size

        byte_string = self.ammos_file.read(total)

        if len(byte_string) != total:
            logging.debug("Can not read all %s bytes of data body", total)
            return None
        return AmmosAudioDataBody(byte_string, number_of_channels, number_of_samples, sample_size)

    def read_next_global_frame_body(self, data_header_length):
        """
        I return the next global frame body read from current position in file.

        :param data_header_length: the length of the data header
        :type data_header_length: int
        """
        audio_data_header = None

        if data_header_length == AmmosAudioDataHeader.HEADER_SIZE:
            logging.info("Read standard data header")
            audio_data_header = self.read_next_global_frame_body_data_header()

        logging.info("Data header length %s", data_header_length)
        if data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE:
            logging.info("Read extended data header")
            audio_data_header = self.read_next_global_frame_body_extended_data_header()

        if audio_data_header is None:
            logging.debug("Data header missing or format unknown")
            return None

        audio_data_body = self.read_next_audio_data_body(audio_data_header.number_of_samples,
                                                         audio_data_header.number_of_channels,
                                                         audio_data_header.sample_size)

        if audio_data_body is None:
            logging.debug("Data body missing")
            return None

        return AmmosGlobalFrameBody(audio_data_header, audio_data_body)

    def pcm_for_channel(self, a_channel):
        """
        I return the raw pcm audio data for a given channel.

        :param a_channel: the channel I have to extract
        :type a_channel: int

        :rtype: bytes
        """
        return b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel) for each in self.container.global_frames])

Ancestors

Methods

def pcm_for_channel(self, a_channel)

I return the raw pcm audio data for a given channel.

:param a_channel: the channel I have to extract :type a_channel: int

:rtype: bytes

Expand source code
def pcm_for_channel(self, a_channel):
    """
    I return the raw pcm audio data for a given channel.

    :param a_channel: the channel I have to extract
    :type a_channel: int

    :rtype: bytes
    """
    return b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel) for each in self.container.global_frames])
def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size)

I return the next audio data read from current position in file.

:param number_of_samples: the number of samples per channel inside data body :type number_of_samples: int

:param number_of_channels: number of channels (e.g. mono, stereo or even more) :type number_of_channels: int

:param sample_size: sample size in bytes (1, 2 or 4 bytes) :type sample_size: int

:return: the next audio data or None if incomplete :rtype: bytes

Expand source code
def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
    """
    I return the next audio data read from current position in file.

    :param number_of_samples: the number of samples per channel inside data body
    :type number_of_samples: int

    :param number_of_channels: number of channels (e.g. mono, stereo or even more)
    :type number_of_channels: int

    :param sample_size: sample size in bytes (1, 2 or 4 bytes)
    :type sample_size: int

    :return: the next audio data or None if incomplete
    :rtype: bytes
    """
    total = number_of_samples*number_of_channels*sample_size

    byte_string = self.ammos_file.read(total)

    if len(byte_string) != total:
        logging.debug("Can not read all %s bytes of data body", total)
        return None
    return AmmosAudioDataBody(byte_string, number_of_channels, number_of_samples, sample_size)
def read_next_global_frame_body(self, data_header_length)

I return the next global frame body read from current position in file.

:param data_header_length: the length of the data header :type data_header_length: int

Expand source code
def read_next_global_frame_body(self, data_header_length):
    """
    I return the next global frame body read from current position in file.

    :param data_header_length: the length of the data header
    :type data_header_length: int
    """
    audio_data_header = None

    if data_header_length == AmmosAudioDataHeader.HEADER_SIZE:
        logging.info("Read standard data header")
        audio_data_header = self.read_next_global_frame_body_data_header()

    logging.info("Data header length %s", data_header_length)
    if data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE:
        logging.info("Read extended data header")
        audio_data_header = self.read_next_global_frame_body_extended_data_header()

    if audio_data_header is None:
        logging.debug("Data header missing or format unknown")
        return None

    audio_data_body = self.read_next_audio_data_body(audio_data_header.number_of_samples,
                                                     audio_data_header.number_of_channels,
                                                     audio_data_header.sample_size)

    if audio_data_body is None:
        logging.debug("Data body missing")
        return None

    return AmmosGlobalFrameBody(audio_data_header, audio_data_body)
def read_next_global_frame_body_data_header(self)

I return the next global frame body data header from current position in file.

:param data_header_size: the number of bytes to read :type data_header_size: int :return: the next Ammos Audio Data header or None if incomplete :rtype: AmmosAudioDataHeader

Expand source code
def read_next_global_frame_body_data_header(self):
    """
    I return the next global frame body data header from current position in file.

    :param data_header_size: the number of bytes to read
    :type data_header_size: int
    :return: the next Ammos Audio Data header or None if incomplete
    :rtype: AmmosAudioDataHeader
    """
    header_size = AmmosAudioDataHeader.HEADER_SIZE

    bytes = self.ammos_file.read(header_size)

    logging.info("\nReading global frame body standard data header\n")
    if ((not bytes) or (len(bytes) < header_size)):
        logging.debug("Can not read all %s bytes of global frame body data header", header_size)
        return None
    return AmmosAudioDataHeader.from_bytes(bytes)
def read_next_global_frame_body_extended_data_header(self)

I return the next global frame body extended data header from current position in file.

:return: the next Ammos Audio Extended Data header or None if incomplete :rtype: AmmosExtendedAudioDataHeader

Expand source code
def read_next_global_frame_body_extended_data_header(self):
    """
    I return the next global frame body extended data header from current position in file.

    :return: the next Ammos Audio Extended Data header or None if incomplete
    :rtype: AmmosExtendedAudioDataHeader
    """
    header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE

    bytes = self.ammos_file.read(header_size)

    if ((not bytes) or (len(bytes) < header_size)):
        logging.debug("Can not read all %s bytes of global frame extended data header", header_size)
        return None
    return AmmosExtendedAudioDataHeader.from_bytes(bytes)

Inherited members