Module ammosreader.AbstractAmmosReader
I provide a base class for specialized AmmosReaders.
Expand source code
"""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
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
bytes = self.ammos_file.read(header_size)
logger.info("Reading next global frame header")
if ((not bytes) or (len(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(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 bytes = self.ammos_file.read(header_size) logger.info("Reading next global frame header") if ((not bytes) or (len(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(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
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
-
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 bytes = self.ammos_file.read(header_size) logger.info("Reading next global frame header") if ((not bytes) or (len(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(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)