Changeset 4455f2b in ammosreader

Timestamp:

06/29/22 15:35:25 (3 years ago)

Author:

Enrico Schwass <ennoausberlin@…>

Branches:

AmmosSource, guix

Children:

2a9faed

Parents:

83e6c89

Message:

another massive refactoring for IF frames

Location:

ammosreader

Files:

• AmmosAudioDataHeader.py (modified) (1 diff)
• AmmosIFDataBlockHeader.py (modified) (1 diff)
• AmmosIFDataBody.py (modified) (1 diff)
• AmmosIFDataHeader.py (modified) (1 diff)
• AmmosIFReader.py (modified) (2 diffs)

ammosreader/AmmosAudioDataHeader.py

@@ -5 +5 @@
 
 
-class AmmosAudioDataHeader():
+class AmmosAudioDataHeader:
     """I implement an AMMOS data header for audio data frames."""
 

ammosreader/AmmosIFDataBlockHeader.py

@@ -1 +1 @@
 import struct
+
 
 class AmmosIFDataBlockHeader:

ammosreader/AmmosIFDataBody.py

@@ -7 +7 @@
 class AmmosIFDataBody:
     """I implement an AMMOS data body for IF data."""
-    def __init__(self, data_blocks):
-        self.__data_blocks = data_blocks
+    def __init__(self, data_blocks=None):
+        if data_blocks is None:
+            self.__data_blocks = []
+        else:
+            self.__data_blocks = data_blocks
 
     @property

ammosreader/AmmosIFDataHeader.py

@@ -6 +6 @@
 
 class AmmosIFDataHeader():
-    """I implement a Ammos data header for IF data frames."""
+    """I implement an Ammos data header for IF data frames."""
 
     HEADER_SIZE = 56

ammosreader/AmmosIFReader.py

@@ -1 +1 @@
 """I provide a specialized Ammos Reader for IF data."""
+import logging
 
 from ammosreader.AmmosGlobalFrameBody import AmmosGlobalFrameBody
 from ammosreader.AmmosIFDataHeader import AmmosIFDataHeader
 from ammosreader.AmmosExtendedIFDataHeader import AmmosExtendedIFDataHeader
-from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader
-from ammosreader.AmmosSingleFrame import AmmosSingleFrame
+from ammosreader.AmmosIFDataBody import AmmosIFDataBody
 from ammosreader.AmmosIFDataBlock import AmmosIFDataBlock
-from ammosreader.AmmosContainer import AmmosContainer
+from ammosreader.AmmosIFDataBlockHeader import AmmosIFDataBlockHeader
 
 
 class AmmosIFReader():
-
-    STANDARD_IF_DATA_HEADER_SIZE = 56
-    EXTENDED_IF_DATA_HEADER_SIZE = 76
+    """I read the IF data embedded in an R&S AMMOS recording."""
 
     def __init__(self, file_name):
@@ -20 +18 @@
     def read_next_global_frame_body_data_header(self):
 
-        bytes = self.ammos_file.read(AmmosIFReader.STANDARD_IF_DATA_HEADER_SIZE)
+        header_size = AmmosIFDataHeader.HEADER_SIZE
 
-        # print("\nReading global frame body standard data header\n")
-        if ((not bytes) or (len(bytes) < AmmosIFReader.STANDARD_IF_DATA_HEADER_SIZE)):
-            # print("Can not read all 56 bytes of global frame body data header")
+        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
-
-        data_header = AmmosIFDataHeader.from_bytes(bytes)
-        # print("Data header", data_header)
-        return data_header
+        return AmmosIFDataHeader.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.
 
-        bytes = self.ammos_file.read(AmmosIFReader.EXTENDED_IF_DATA_HEADER_SIZE)
-        # print("\nReading global frame body extended data header\n")
+        :return: the next Ammos Extended IF data header or None if incomplete
+        :rtype: AmmosExtendedIFDataHeader
+        """
+        header_size = AmmosExtendedIFDataHeader.HEADER_SIZE
 
-        if ((not bytes) or (len(bytes) < AmmosIFReader.EXTENDED_IF_DATA_HEADER_SIZE)):
-            # print("Can not read all ", 76, "bytes of global frame extended data header")
+        bytes = self.ammos_file.read(header_size)
+
+        logging.info("\nReading global frame body extended data header\n")
+        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
-        extended_data_header = AmmosExtendedIFDataHeader.from_bytes(bytes)
-        # print("Extended data header", extended_data_header)
-        return extended_data_header
+        return AmmosExtendedIFDataHeader.from_bytes(bytes)
 
-    def read_next_if_data_blocks(self, n, length):
+    def read_next_if_data_body(self, number_of_data_blocks, data_length):
+        """
+        I return the next data body read from current position in file.
 
-        # FIXME: Describe the parameters better
+        :param number_of_data_blocks: the number of data blocks inside the body
+        :type number_of_data_blocks: int
 
-        data_blocks = []
+        :param data_length: the length of the raw data inside a single block
+        :type data_length: int
+        """
+        header_size = AmmosIFDataBlockHeader.HEADER_SIZE
 
-        block_length = 4 + length
+        data_body = AmmosIFDataBody()
 
-        total = n*block_length
+        block_length = header_size + data_length
+
+        total = number_of_data_blocks*block_length
 
         byte_string = self.ammos_file.read(block_length)
 
         if len(byte_string) != total:
-            # print("Can not read all", total, "bytes of data body")
+            logging.debug("Can not read all %s bytes of data body", total)
             return None
 
-        for i in range(0, n):
+        for i in range(0, number_of_data_blocks):
             result = byte_string[i*block_length:(i*block_length+block_length)]
-            data_blocks.append(AmmosIFDataBlock(result[0:4], result[4:]))
+            data_body.add_data_block(AmmosIFDataBlock(AmmosIFDataBlockHeader.from_bytes(result[0:header_size]),
+                                                      result[header_size:]))
 
-        return data_blocks
+        return data_body
 
     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
+        """
         if_data_header = None
 
-        if data_header_length == AmmosIFReader.STANDARD_IF_DATA_HEADER_SIZE:
+        if data_header_length == AmmosIFDataHeader.HEADER_SIZE:
             if_data_header = self.read_next_global_frame_body_data_header()
-        if data_header_length == AmmosIFReader.EXTENDED_IF_DATA_HEADER_SIZE:
+        if data_header_length == AmmosExtendedIFDataHeader.HEADER_SIZE:
             if_data_header = self.read_next_global_frame_body_extended_data_header()
 
         if if_data_header is None:
-            # print("Data header missing")
+            logging.debug("Data header missing")
             return None