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
- AbstractAmmosReader
- abc.ABC
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