Module ammosreader.AbstractAmmosReader
I provide a base class for specialized AmmosReaders.
Expand source code
"""I provide a base class for specialized AmmosReaders."""
import logging
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
logging.basicConfig(filename='ammos.log', level=logging.DEBUG)
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:
logging.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:
logging.debug("Frame: %s incomplete", frames_read+1)
break
logging.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)
logging.info("Reading next global frame header")
if ((not bytes) or (len(bytes) < header_size)):
logging.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
# logging.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:
logging.debug("Global frame header missing")
return None
if global_frame_header.data_header_length is None:
logging.debug("Data header length empty")
return None
if global_frame_header.frame_type not in list(FrameType):
logging.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:
logging.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:
logging.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
logging.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: logging.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: logging.debug("Frame: %s incomplete", frames_read+1) break logging.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) logging.info("Reading next global frame header") if ((not bytes) or (len(bytes) < header_size)): logging.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 # logging.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: logging.debug("Global frame header missing") return None if global_frame_header.data_header_length is None: logging.debug("Data header length empty") return None if global_frame_header.frame_type not in list(FrameType): logging.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: logging.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: logging.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 logging.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: logging.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: logging.debug("Frame: %s incomplete", frames_read+1) break logging.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) logging.info("Reading next global frame header") if ((not bytes) or (len(bytes) < header_size)): logging.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 # logging.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: logging.debug("Global frame header missing") return None if global_frame_header.data_header_length is None: logging.debug("Data header length empty") return None if global_frame_header.frame_type not in list(FrameType): logging.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: logging.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: logging.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 logging.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)