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 NoneClasses
--
-
-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 NoneAncestors
--
-
- 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)
-
-