Module ammosreader.AmmosAudioReader

I provide a specialized Ammos Reader for audio data.

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

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

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)

        logger.info("\nReading global frame body standard data header\n")
        if ((not bytes) or (len(bytes) < header_size)):
            logger.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)):
            logger.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:
            logger.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:
            logger.info("Read standard data header")
            audio_data_header = self.read_next_global_frame_body_data_header()

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

        if audio_data_header is None:
            logger.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:
            logger.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)

        logger.info("\nReading global frame body standard data header\n")
        if ((not bytes) or (len(bytes) < header_size)):
            logger.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)):
            logger.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:
            logger.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:
            logger.info("Read standard data header")
            audio_data_header = self.read_next_global_frame_body_data_header()

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

        if audio_data_header is None:
            logger.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:
            logger.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:
        logger.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:
        logger.info("Read standard data header")
        audio_data_header = self.read_next_global_frame_body_data_header()

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

    if audio_data_header is None:
        logger.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:
        logger.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)

    logger.info("\nReading global frame body standard data header\n")
    if ((not bytes) or (len(bytes) < header_size)):
        logger.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)):
        logger.debug("Can not read all %s bytes of global frame extended data header", header_size)
        return None
    return AmmosExtendedAudioDataHeader.from_bytes(bytes)

Inherited members