-def from_bytes(bytes)
+def from_bytes(in_bytes)
@classmethod
-def from_bytes(cls, bytes):
+def from_bytes(cls, in_bytes):
"""I return an AMMOS data header from given bytes."""
- assert len(bytes) == cls.HEADER_SIZE
- elements = struct.unpack('<IIQIIIII', bytes)
+ assert len(in_bytes) == cls.HEADER_SIZE
+ elements = struct.unpack('<IIQIIIII', in_bytes)
sample_rate = elements[0]
status = elements[1]
Index: doc/ammosreader/AmmosAudioReader.html
===================================================================
--- doc/ammosreader/AmmosAudioReader.html (revision 04097cdb708531c6549b286c360cd998d5deb618)
+++ doc/ammosreader/AmmosAudioReader.html (revision 7752d1f99e91b74a1fdc11cfd15ae6166803378e)
@@ -37,15 +37,7 @@
from ammosreader import logger
+
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):
@@ -60,11 +52,11 @@
header_size = AmmosAudioDataHeader.HEADER_SIZE
- bytes = self.ammos_file.read(header_size)
+ in_bytes = self.ammos_file.read(header_size)
logger.info("\nReading global frame body standard data header\n")
- if ((not bytes) or (len(bytes) < header_size)):
+ if ((not in_bytes) or (len(in_bytes) < header_size)):
logger.debug("Can not read all %s bytes of global frame body data header", header_size)
return None
- return AmmosAudioDataHeader.from_bytes(bytes)
+ return AmmosAudioDataHeader.from_bytes(in_bytes)
def read_next_global_frame_body_extended_data_header(self):
@@ -77,10 +69,10 @@
header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE
- bytes = self.ammos_file.read(header_size)
-
- if ((not bytes) or (len(bytes) < header_size)):
+ in_bytes = self.ammos_file.read(header_size)
+
+ if ((not in_bytes) or (len(in_bytes) < header_size)):
logger.debug("Can not read all %s bytes of global frame extended data header", header_size)
return None
- return AmmosExtendedAudioDataHeader.from_bytes(bytes)
+ return AmmosExtendedAudioDataHeader.from_bytes(in_bytes)
def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
@@ -150,5 +142,6 @@
:rtype: bytes
"""
- return b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel) for each in self.container.global_frames])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 +
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
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):
@@ -198,11 +183,11 @@
header_size = AmmosAudioDataHeader.HEADER_SIZE
- bytes = self.ammos_file.read(header_size)
+ in_bytes = self.ammos_file.read(header_size)
logger.info("\nReading global frame body standard data header\n")
- if ((not bytes) or (len(bytes) < header_size)):
+ if ((not in_bytes) or (len(in_bytes) < header_size)):
logger.debug("Can not read all %s bytes of global frame body data header", header_size)
return None
- return AmmosAudioDataHeader.from_bytes(bytes)
+ return AmmosAudioDataHeader.from_bytes(in_bytes)
def read_next_global_frame_body_extended_data_header(self):
@@ -215,10 +200,10 @@
header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE
- bytes = self.ammos_file.read(header_size)
-
- if ((not bytes) or (len(bytes) < header_size)):
+ in_bytes = self.ammos_file.read(header_size)
+
+ if ((not in_bytes) or (len(in_bytes) < header_size)):
logger.debug("Can not read all %s bytes of global frame extended data header", header_size)
return None
- return AmmosExtendedAudioDataHeader.from_bytes(bytes)
+ return AmmosExtendedAudioDataHeader.from_bytes(in_bytes)
def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
@@ -288,5 +273,6 @@
:rtype: bytes
"""
- return b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel) for each in self.container.global_frames])"""I read a Ammos datastream from a socket."""
-import select
+# import select
import socket
from collections import deque
@@ -37,8 +37,12 @@
from ammosreader.AmmosExtendedAudioDataHeader import AmmosExtendedAudioDataHeader
from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader
+from ammosreader.AmmosConstants import AmmosAudioDemodType
from ammosreader import logger
+
class AmmosAudioSocketReader:
- def __init__(self, socket:socket.socket):
+ """I read Ammos Audio data from a socket"""
+
+ def __init__(self, in_socket: socket.socket):
"""
Initializes the AmmosAudioSocketReader
@@ -48,9 +52,9 @@
"""
- #buffer for reading socket bytewise und check for the magic word
+ # buffer for reading socket bytewise und check for the magic word
self.__magic_word_buffer = deque(maxlen=4)
- #input socket to read from
- self.__socket = socket
+ # input socket to read from
+ self.__socket = in_socket
def __get_next_data(self, byte_count: int) -> bytearray:
@@ -69,7 +73,8 @@
byte_array = []
-
+
while len(b''.join(byte_array)) < byte_count:
- logger.info(f"Remaining Bytes: {byte_count - len(b''.join(byte_array))}")
+ logger.info("Remaining Bytes: %s", byte_count - len(b''.join(byte_array)))
+ # logger.info(f"Remaining Bytes: {byte_count - len(b''.join(byte_array))}")
self.__socket.settimeout(5)
new_bytes = self.__socket.recv(byte_count - len(b''.join(byte_array)), socket.MSG_WAITALL)
@@ -77,6 +82,6 @@
if not new_bytes:
raise TimeoutError("Socket timed out while reading data")
-
- logger.info(f"Got {len(new_bytes)} bytes of {byte_count - len(b''.join(byte_array))} ramining")
+ logger.info("Got %s bytes of %s remaining", len(new_bytes), byte_count - len(b''.join(byte_array)))
+ # logger.info(f"Got {len(new_bytes)} bytes of {byte_count - len(b''.join(byte_array))} remaining")
byte_array.append(new_bytes)
@@ -84,5 +89,5 @@
return b''.join(byte_array)
- def __audio_data_body_to_numpy(self, audio_data_body:bytearray) -> np.ndarray:
+ def __audio_data_body_to_numpy(self, audio_data_body: bytearray) -> np.ndarray:
"""
converts the audio data body to a numpy array
@@ -101,5 +106,5 @@
Raises:
- TimeoutError: Raisees TimeoutError if the socket does not serve data anymore
+ TimeoutError: Raises TimeoutError if the socket does not serve data anymore
Returns:
@@ -113,7 +118,7 @@
# raise Exception if socket does not return anything
if len(new_byte) < 1:
- raise TimeoutError
-
- #read loop
+ raise TimeoutError
+
+ # read loop
while new_byte:
#
@@ -121,44 +126,52 @@
byte_array = b''.join(self.__magic_word_buffer)
- if byte_array.hex() == '726574fb':
- #print(byte_array.hex())
+ if byte_array.hex() == AmmosGlobalFrameHeader.MAGIC_WORD:
+ # print(byte_array.hex())
ammos_global_header_buffer = list(self.__magic_word_buffer)
ammos_global_header_buffer.append(self.__get_next_data(20))
- #while len(b''.join(ammos_global_header_buffer)) < 24:
+ # while len(b''.join(ammos_global_header_buffer)) < 24:
# ammos_global_header_buffer.append(self.__socket.recv(24 - len(b''.join(ammos_global_header_buffer))))
-
+
ammos_global_header = AmmosGlobalFrameHeader.from_bytes(b''.join(ammos_global_header_buffer))
logger.info(ammos_global_header)
- if ammos_global_header.data_header_length == 44 and ammos_global_header.frame_type == 256:
- byte_array_header = self.__get_next_data(44)
- #while len(b''.join(byte_array_header)) < 44:
+ if (ammos_global_header.data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL):
+
+ byte_array_header = self.__get_next_data(AmmosExtendedAudioDataHeader.HEADER_SIZE)
+ # while len(b''.join(byte_array_header)) < 44:
# byte_array_header.append(self.__socket.recv(44 - len(b''.join(byte_array_header))))
ammos_extended_audio_data_header = AmmosExtendedAudioDataHeader.from_bytes(byte_array_header)
- logger.info(ammos_extended_audio_data_header.number_of_samples, ammos_extended_audio_data_header.number_of_channels, ammos_extended_audio_data_header.sample_size)
- audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples*
- ammos_extended_audio_data_header.number_of_channels*
+ logger.debug(str(ammos_extended_audio_data_header.number_of_samples or ''),
+ str(ammos_extended_audio_data_header.number_of_channels or ''),
+ str(ammos_extended_audio_data_header.sample_size or ''))
+
+ audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples *
+ ammos_extended_audio_data_header.number_of_channels *
ammos_extended_audio_data_header.sample_size)
audio_array = self.__audio_data_body_to_numpy(audio_body)
- logger.info(len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate)
+ # print("44,256", len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate)
+ logger.debug("%s, %s", len(audio_array), len(audio_array) / ammos_extended_audio_data_header.sample_rate)
return [audio_array, ammos_extended_audio_data_header.sample_rate]
- elif ammos_global_header.data_header_length == 36 and ammos_global_header.frame_type == 256:
- byte_array_header = self.__get_next_data(36)
- #while len(b''.join(byte_array_header)) < 36:
+ elif (ammos_global_header.data_header_length == AmmosAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL):
+ byte_array_header = self.__get_next_data(AmmosAudioDataHeader.HEADER_SIZE)
+ # while len(b''.join(byte_array_header)) < 36:
# byte_array_header.append(self.__socket.recv(36 - len(b''.join(byte_array_header))))
ammos_audio_data_header = AmmosAudioDataHeader.from_bytes(byte_array_header)
- logger.info(ammos_audio_data_header.number_of_samples, ammos_audio_data_header.number_of_channels, ammos_audio_data_header.sample_size)
- audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples*
- ammos_extended_audio_data_header.number_of_channels*
+ logger.debug(str(ammos_audio_data_header.number_of_samples or ''),
+ str(ammos_audio_data_header.number_of_channels or ''),
+ str(ammos_audio_data_header.sample_size or ''))
+
+ audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples *
+ ammos_extended_audio_data_header.number_of_channels *
ammos_extended_audio_data_header.sample_size)
audio_array = self.__audio_data_body_to_numpy(audio_body)
- logger.info(len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
+ logger.debug("%s, %s", len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
return [audio_array, ammos_audio_data_header.sample_rate]
@@ -170,5 +183,5 @@
# raise Exception if socket does not return anything
if len(new_byte) < 1:
- raise TimeoutError
+ raise TimeoutError
return None
class AmmosAudioSocketReader
-(socket: socket.socket)
+(in_socket: socket.socket)
Initializes the AmmosAudioSocketReader
+I read Ammos Audio data from a socket
+Initializes the AmmosAudioSocketReader
class AmmosAudioSocketReader:
- def __init__(self, socket:socket.socket):
+ """I read Ammos Audio data from a socket"""
+
+ def __init__(self, in_socket: socket.socket):
"""
Initializes the AmmosAudioSocketReader
@@ -208,9 +224,9 @@
"""
- #buffer for reading socket bytewise und check for the magic word
+ # buffer for reading socket bytewise und check for the magic word
self.__magic_word_buffer = deque(maxlen=4)
- #input socket to read from
- self.__socket = socket
+ # input socket to read from
+ self.__socket = in_socket
def __get_next_data(self, byte_count: int) -> bytearray:
@@ -229,7 +245,8 @@
byte_array = []
-
+
while len(b''.join(byte_array)) < byte_count:
- logger.info(f"Remaining Bytes: {byte_count - len(b''.join(byte_array))}")
+ logger.info("Remaining Bytes: %s", byte_count - len(b''.join(byte_array)))
+ # logger.info(f"Remaining Bytes: {byte_count - len(b''.join(byte_array))}")
self.__socket.settimeout(5)
new_bytes = self.__socket.recv(byte_count - len(b''.join(byte_array)), socket.MSG_WAITALL)
@@ -237,6 +254,6 @@
if not new_bytes:
raise TimeoutError("Socket timed out while reading data")
-
- logger.info(f"Got {len(new_bytes)} bytes of {byte_count - len(b''.join(byte_array))} ramining")
+ logger.info("Got %s bytes of %s remaining", len(new_bytes), byte_count - len(b''.join(byte_array)))
+ # logger.info(f"Got {len(new_bytes)} bytes of {byte_count - len(b''.join(byte_array))} remaining")
byte_array.append(new_bytes)
@@ -244,5 +261,5 @@
return b''.join(byte_array)
- def __audio_data_body_to_numpy(self, audio_data_body:bytearray) -> np.ndarray:
+ def __audio_data_body_to_numpy(self, audio_data_body: bytearray) -> np.ndarray:
"""
converts the audio data body to a numpy array
@@ -261,5 +278,5 @@
Raises:
- TimeoutError: Raisees TimeoutError if the socket does not serve data anymore
+ TimeoutError: Raises TimeoutError if the socket does not serve data anymore
Returns:
@@ -273,7 +290,7 @@
# raise Exception if socket does not return anything
if len(new_byte) < 1:
- raise TimeoutError
-
- #read loop
+ raise TimeoutError
+
+ # read loop
while new_byte:
#
@@ -281,44 +298,52 @@
byte_array = b''.join(self.__magic_word_buffer)
- if byte_array.hex() == '726574fb':
- #print(byte_array.hex())
+ if byte_array.hex() == AmmosGlobalFrameHeader.MAGIC_WORD:
+ # print(byte_array.hex())
ammos_global_header_buffer = list(self.__magic_word_buffer)
ammos_global_header_buffer.append(self.__get_next_data(20))
- #while len(b''.join(ammos_global_header_buffer)) < 24:
+ # while len(b''.join(ammos_global_header_buffer)) < 24:
# ammos_global_header_buffer.append(self.__socket.recv(24 - len(b''.join(ammos_global_header_buffer))))
-
+
ammos_global_header = AmmosGlobalFrameHeader.from_bytes(b''.join(ammos_global_header_buffer))
logger.info(ammos_global_header)
- if ammos_global_header.data_header_length == 44 and ammos_global_header.frame_type == 256:
- byte_array_header = self.__get_next_data(44)
- #while len(b''.join(byte_array_header)) < 44:
+ if (ammos_global_header.data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL):
+
+ byte_array_header = self.__get_next_data(AmmosExtendedAudioDataHeader.HEADER_SIZE)
+ # while len(b''.join(byte_array_header)) < 44:
# byte_array_header.append(self.__socket.recv(44 - len(b''.join(byte_array_header))))
ammos_extended_audio_data_header = AmmosExtendedAudioDataHeader.from_bytes(byte_array_header)
- logger.info(ammos_extended_audio_data_header.number_of_samples, ammos_extended_audio_data_header.number_of_channels, ammos_extended_audio_data_header.sample_size)
- audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples*
- ammos_extended_audio_data_header.number_of_channels*
+ logger.debug(str(ammos_extended_audio_data_header.number_of_samples or ''),
+ str(ammos_extended_audio_data_header.number_of_channels or ''),
+ str(ammos_extended_audio_data_header.sample_size or ''))
+
+ audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples *
+ ammos_extended_audio_data_header.number_of_channels *
ammos_extended_audio_data_header.sample_size)
audio_array = self.__audio_data_body_to_numpy(audio_body)
- logger.info(len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate)
+ # print("44,256", len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate)
+ logger.debug("%s, %s", len(audio_array), len(audio_array) / ammos_extended_audio_data_header.sample_rate)
return [audio_array, ammos_extended_audio_data_header.sample_rate]
- elif ammos_global_header.data_header_length == 36 and ammos_global_header.frame_type == 256:
- byte_array_header = self.__get_next_data(36)
- #while len(b''.join(byte_array_header)) < 36:
+ elif (ammos_global_header.data_header_length == AmmosAudioDataHeader.HEADER_SIZE and ammos_global_header.frame_type == AmmosAudioDemodType.DIGITAL):
+ byte_array_header = self.__get_next_data(AmmosAudioDataHeader.HEADER_SIZE)
+ # while len(b''.join(byte_array_header)) < 36:
# byte_array_header.append(self.__socket.recv(36 - len(b''.join(byte_array_header))))
ammos_audio_data_header = AmmosAudioDataHeader.from_bytes(byte_array_header)
- logger.info(ammos_audio_data_header.number_of_samples, ammos_audio_data_header.number_of_channels, ammos_audio_data_header.sample_size)
- audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples*
- ammos_extended_audio_data_header.number_of_channels*
+ logger.debug(str(ammos_audio_data_header.number_of_samples or ''),
+ str(ammos_audio_data_header.number_of_channels or ''),
+ str(ammos_audio_data_header.sample_size or ''))
+
+ audio_body = self.__get_next_data(ammos_extended_audio_data_header.number_of_samples *
+ ammos_extended_audio_data_header.number_of_channels *
ammos_extended_audio_data_header.sample_size)
audio_array = self.__audio_data_body_to_numpy(audio_body)
- logger.info(len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
+ logger.debug("%s, %s", len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
return [audio_array, ammos_audio_data_header.sample_rate]
@@ -330,5 +355,5 @@
# raise Exception if socket does not return anything
if len(new_byte) < 1:
- raise TimeoutError
+ raise TimeoutError
return NoneTimeoutErrorI store the content of an R&S Ammos file in a more accessible way.
-"""I store the content of an R&S Ammos file in a more accessible way."""
+I provide an Ammos container.
+
+
+Expand source code
+
+"""I provide an Ammos container."""
+
class AmmosContainer():
-
+ """I store the content of an R&S Ammos file in a more accessible way."""
def __init__(self, name, frames):
self.__name = name
@@ -56,5 +57,6 @@
def size(self):
- return sum([each.global_frame_header.frame_length for each in self.__global_frames])
+ # return sum([each.global_frame_header.frame_length for each in self.__global_frames])
+ return sum(each.global_frame_header.frame_length for each in self.__global_frames)
def frequencies(self):
@@ -101,5 +103,5 @@
-
-
+
I store the content of an R&S Ammos file in a more accessible way.
@@ -107,5 +109,5 @@
class AmmosContainer():
-
+ """I store the content of an R&S Ammos file in a more accessible way."""
def __init__(self, name, frames):
self.__name = name
@@ -132,5 +134,6 @@
def size(self):
- return sum([each.global_frame_header.frame_length for each in self.__global_frames])
+ # return sum([each.global_frame_header.frame_length for each in self.__global_frames])
+ return sum(each.global_frame_header.frame_length for each in self.__global_frames)
def frequencies(self):
@@ -293,5 +296,6 @@
def size(self):
- return sum([each.global_frame_header.frame_length for each in self.__global_frames])
+ # return sum([each.global_frame_header.frame_length for each in self.__global_frames])
+ return sum(each.global_frame_header.frame_length for each in self.__global_frames)
Index: doc/ammosreader/AmmosExtendedAudioDataHeader.html
===================================================================
--- doc/ammosreader/AmmosExtendedAudioDataHeader.html (revision 04097cdb708531c6549b286c360cd998d5deb618)
+++ doc/ammosreader/AmmosExtendedAudioDataHeader.html (revision 7752d1f99e91b74a1fdc11cfd15ae6166803378e)
@@ -39,9 +39,9 @@
@classmethod
- def from_bytes(cls, bytes):
+ def from_bytes(cls, in_bytes):
"""I return a new AMMOS extended data header for audio frames built from given bytes."""
- assert len(bytes) == cls.HEADER_SIZE
- standard_header = AmmosAudioDataHeader.from_bytes(bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
- extended_header_elements = struct.unpack('<Q', bytes[AmmosAudioDataHeader.HEADER_SIZE:])
+ assert len(in_bytes) == cls.HEADER_SIZE
+ standard_header = AmmosAudioDataHeader.from_bytes(in_bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
+ extended_header_elements = struct.unpack('<Q', in_bytes[AmmosAudioDataHeader.HEADER_SIZE:])
timestamp = extended_header_elements[0]
sample_rate = standard_header.sample_rate
@@ -94,9 +94,9 @@
@classmethod
- def from_bytes(cls, bytes):
+ def from_bytes(cls, in_bytes):
"""I return a new AMMOS extended data header for audio frames built from given bytes."""
- assert len(bytes) == cls.HEADER_SIZE
- standard_header = AmmosAudioDataHeader.from_bytes(bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
- extended_header_elements = struct.unpack('<Q', bytes[AmmosAudioDataHeader.HEADER_SIZE:])
+ assert len(in_bytes) == cls.HEADER_SIZE
+ standard_header = AmmosAudioDataHeader.from_bytes(in_bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
+ extended_header_elements = struct.unpack('<Q', in_bytes[AmmosAudioDataHeader.HEADER_SIZE:])
timestamp = extended_header_elements[0]
sample_rate = standard_header.sample_rate
@@ -134,5 +134,5 @@
-def from_bytes(bytes)
+def from_bytes(in_bytes)
-
@@ -143,9 +143,9 @@
@classmethod
-def from_bytes(cls, bytes):
+def from_bytes(cls, in_bytes):
"""I return a new AMMOS extended data header for audio frames built from given bytes."""
- assert len(bytes) == cls.HEADER_SIZE
- standard_header = AmmosAudioDataHeader.from_bytes(bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
- extended_header_elements = struct.unpack('<Q', bytes[AmmosAudioDataHeader.HEADER_SIZE:])
+ assert len(in_bytes) == cls.HEADER_SIZE
+ standard_header = AmmosAudioDataHeader.from_bytes(in_bytes[0:AmmosAudioDataHeader.HEADER_SIZE])
+ extended_header_elements = struct.unpack('<Q', in_bytes[AmmosAudioDataHeader.HEADER_SIZE:])
timestamp = extended_header_elements[0]
sample_rate = standard_header.sample_rate
Index: doc/ammosreader/AmmosExtendedIFDataHeader.html
===================================================================
--- doc/ammosreader/AmmosExtendedIFDataHeader.html (revision 04097cdb708531c6549b286c360cd998d5deb618)
+++ doc/ammosreader/AmmosExtendedIFDataHeader.html (revision 7752d1f99e91b74a1fdc11cfd15ae6166803378e)
@@ -38,8 +38,8 @@
@classmethod
- def from_bytes(cls, bytes):
+ def from_bytes(cls, in_bytes):
"""I return an AMMOS extended data header from given bytes."""
- standard_header = AmmosIFDataHeader.from_bytes(bytes[0:AmmosIFDataHeader.HEADER_SIZE])
- extended_header_elements = struct.unpack('<QQI', bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
+ standard_header = AmmosIFDataHeader.from_bytes(in_bytes[0:AmmosIFDataHeader.HEADER_SIZE])
+ extended_header_elements = struct.unpack('<QQI', in_bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
block_count = standard_header.block_count
block_length = standard_header.block_length
@@ -57,5 +57,5 @@
sample_counter = extended_header_elements[1]
antenna_correction = extended_header_elements[2]
- size = len(bytes)
+ size = len(in_bytes)
return AmmosExtendedIFDataHeader(size, block_count, block_length, timestamp, status, source_id,
source_state, frequency, bandwidth, sample_rate, interpolation,
@@ -126,8 +126,8 @@
@classmethod
- def from_bytes(cls, bytes):
+ def from_bytes(cls, in_bytes):
"""I return an AMMOS extended data header from given bytes."""
- standard_header = AmmosIFDataHeader.from_bytes(bytes[0:AmmosIFDataHeader.HEADER_SIZE])
- extended_header_elements = struct.unpack('<QQI', bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
+ standard_header = AmmosIFDataHeader.from_bytes(in_bytes[0:AmmosIFDataHeader.HEADER_SIZE])
+ extended_header_elements = struct.unpack('<QQI', in_bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
block_count = standard_header.block_count
block_length = standard_header.block_length
@@ -145,5 +145,5 @@
sample_counter = extended_header_elements[1]
antenna_correction = extended_header_elements[2]
- size = len(bytes)
+ size = len(in_bytes)
return AmmosExtendedIFDataHeader(size, block_count, block_length, timestamp, status, source_id,
source_state, frequency, bandwidth, sample_rate, interpolation,
@@ -197,5 +197,5 @@
-def from_bytes(bytes)
+def from_bytes(in_bytes)
-
@@ -206,8 +206,8 @@
@classmethod
-def from_bytes(cls, bytes):
+def from_bytes(cls, in_bytes):
"""I return an AMMOS extended data header from given bytes."""
- standard_header = AmmosIFDataHeader.from_bytes(bytes[0:AmmosIFDataHeader.HEADER_SIZE])
- extended_header_elements = struct.unpack('<QQI', bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
+ standard_header = AmmosIFDataHeader.from_bytes(in_bytes[0:AmmosIFDataHeader.HEADER_SIZE])
+ extended_header_elements = struct.unpack('<QQI', in_bytes[AmmosIFDataHeader.HEADER_SIZE:cls.HEADER_SIZE])
block_count = standard_header.block_count
block_length = standard_header.block_length
@@ -225,5 +225,5 @@
sample_counter = extended_header_elements[1]
antenna_correction = extended_header_elements[2]
- size = len(bytes)
+ size = len(in_bytes)
return AmmosExtendedIFDataHeader(size, block_count, block_length, timestamp, status, source_id,
source_state, frequency, bandwidth, sample_rate, interpolation,
Index: doc/ammosreader/AmmosGlobalFrameHeader.html
===================================================================
--- doc/ammosreader/AmmosGlobalFrameHeader.html (revision 04097cdb708531c6549b286c360cd998d5deb618)
+++ doc/ammosreader/AmmosGlobalFrameHeader.html (revision 7752d1f99e91b74a1fdc11cfd15ae6166803378e)
@@ -39,9 +39,9 @@
@classmethod
- def from_bytes(cls, bytes):
+ def from_bytes(cls, in_bytes):
"""I create a new AmmosGlobalFrameHeader from bytes."""
- assert len(bytes) == cls.HEADER_SIZE
-
- elements = struct.unpack('<4s4s4s4s4s4s', bytes)
+ assert len(in_bytes) == cls.HEADER_SIZE
+
+ elements = struct.unpack('<4s4s4s4s4s4s', in_bytes)
magic_word = elements[0].hex()
@@ -116,9 +116,9 @@
@classmethod
- def from_bytes(cls, bytes):
+ def from_bytes(cls, in_bytes):
"""I create a new AmmosGlobalFrameHeader from bytes."""
- assert len(bytes) == cls.HEADER_SIZE
-
- elements = struct.unpack('<4s4s4s4s4s4s', bytes)
+ assert len(in_bytes) == cls.HEADER_SIZE
+
+ elements = struct.unpack('<4s4s4s4s4s4s', in_bytes)
magic_word = elements[0].hex()
@@ -179,5 +179,5 @@
-def from_bytes(bytes)
+def from_bytes(in_bytes)
-
@@ -188,9 +188,9 @@
@classmethod
-def from_bytes(cls, bytes):
+def from_bytes(cls, in_bytes):
"""I create a new AmmosGlobalFrameHeader from bytes."""
- assert len(bytes) == cls.HEADER_SIZE
-
- elements = struct.unpack('<4s4s4s4s4s4s', bytes)
+ assert len(in_bytes) == cls.HEADER_SIZE
+
+ elements = struct.unpack('<4s4s4s4s4s4s', in_bytes)
magic_word = elements[0].hex()
Index: doc/ammosreader/AmmosIFDataBlockHeader.html
===================================================================
--- doc/ammosreader/AmmosIFDataBlockHeader.html (revision 04097cdb708531c6549b286c360cd998d5deb618)
+++ doc/ammosreader/AmmosIFDataBlockHeader.html (revision 7752d1f99e91b74a1fdc11cfd15ae6166803378e)
@@ -36,8 +36,8 @@
@classmethod
- def from_bytes(cls, bytes):
+ def from_bytes(cls, in_bytes):
"""I return an AMMOS IF data block header built from bytes."""
- assert len(bytes) == cls.HEADER_SIZE
- elements = struct.unpack('<ccH', bytes)
+ assert len(in_bytes) == cls.HEADER_SIZE
+ elements = struct.unpack('<ccH', in_bytes)
header = cls()
first_entry = int.from_bytes(elements[0], byteorder='little')
@@ -125,8 +125,8 @@
@classmethod
- def from_bytes(cls, bytes):
+ def from_bytes(cls, in_bytes):
"""I return an AMMOS IF data block header built from bytes."""
- assert len(bytes) == cls.HEADER_SIZE
- elements = struct.unpack('<ccH', bytes)
+ assert len(in_bytes) == cls.HEADER_SIZE
+ elements = struct.unpack('<ccH', in_bytes)
header = cls()
first_entry = int.from_bytes(elements[0], byteorder='little')
@@ -198,5 +198,5 @@
-def from_bytes(bytes)
+def from_bytes(in_bytes)
-
@@ -207,8 +207,8 @@
@classmethod
-def from_bytes(cls, bytes):
+def from_bytes(cls, in_bytes):
"""I return an AMMOS IF data block header built from bytes."""
- assert len(bytes) == cls.HEADER_SIZE
- elements = struct.unpack('<ccH', bytes)
+ assert len(in_bytes) == cls.HEADER_SIZE
+ elements = struct.unpack('<ccH', in_bytes)
header = cls()
first_entry = int.from_bytes(elements[0], byteorder='little')
Index: doc/ammosreader/AmmosIFDataHeader.html
===================================================================
--- doc/ammosreader/AmmosIFDataHeader.html (revision 04097cdb708531c6549b286c360cd998d5deb618)
+++ doc/ammosreader/AmmosIFDataHeader.html (revision 7752d1f99e91b74a1fdc11cfd15ae6166803378e)
@@ -40,8 +40,8 @@
@classmethod
- def from_bytes(cls, bytes):
+ def from_bytes(cls, in_bytes):
"""I return an AMMOS data header from given bytes."""
- assert len(bytes) == cls.HEADER_SIZE
- elements = struct.unpack('<IIQIIIQIIIIi', bytes)
+ assert len(in_bytes) == cls.HEADER_SIZE
+ elements = struct.unpack('<IIQIIIQIIIIi', in_bytes)
block_count = elements[0]
block_length = int(elements[1])*4
@@ -77,5 +77,5 @@
self.voltage_ref = voltage_ref
- def __str_(self):
+ def __str__(self):
output = ("\nGlobal frame body data header\n" +
"-----------------------------\n" +
@@ -115,8 +115,8 @@
@classmethod
- def from_bytes(cls, bytes):
+ def from_bytes(cls, in_bytes):
"""I return an AMMOS data header from given bytes."""
- assert len(bytes) == cls.HEADER_SIZE
- elements = struct.unpack('<IIQIIIQIIIIi', bytes)
+ assert len(in_bytes) == cls.HEADER_SIZE
+ elements = struct.unpack('<IIQIIIQIIIIi', in_bytes)
block_count = elements[0]
block_length = int(elements[1])*4
@@ -152,5 +152,5 @@
self.voltage_ref = voltage_ref
- def __str_(self):
+ def __str__(self):
output = ("\nGlobal frame body data header\n" +
"-----------------------------\n" +
@@ -173,5 +173,5 @@
-def from_bytes(bytes)
+def from_bytes(in_bytes)
-
@@ -182,8 +182,8 @@
@classmethod
-def from_bytes(cls, bytes):
+def from_bytes(cls, in_bytes):
"""I return an AMMOS data header from given bytes."""
- assert len(bytes) == cls.HEADER_SIZE
- elements = struct.unpack('<IIQIIIQIIIIi', bytes)
+ assert len(in_bytes) == cls.HEADER_SIZE
+ elements = struct.unpack('<IIQIIIQIIIIi', in_bytes)
block_count = elements[0]
block_length = int(elements[1])*4
Index: doc/ammosreader/AmmosIFReader.html
===================================================================
--- doc/ammosreader/AmmosIFReader.html (revision 04097cdb708531c6549b286c360cd998d5deb618)
+++ doc/ammosreader/AmmosIFReader.html (revision 7752d1f99e91b74a1fdc11cfd15ae6166803378e)
@@ -39,21 +39,19 @@
from ammosreader import logger
+
class AmmosIFReader(AbstractAmmosReader):
"""I read the IF data embedded in an R&S AMMOS recording."""
- def __init__(self, file_name):
- super().__init__(file_name)
-
def read_next_global_frame_body_data_header(self):
-
+ """I read the next data header of a global frame body from current position in file."""
header_size = AmmosIFDataHeader.HEADER_SIZE
- bytes = self.ammos_file.read(header_size)
+ in_bytes = self.ammos_file.read(header_size)
logger.info("\nReading global frame body standard data header\n")
- if ((not bytes) or (len(bytes) < header_size)):
+ if ((not in_bytes) or (len(in_bytes) < header_size)):
logger.debug("Can not read all %s bytes of global frame body data header", header_size)
return None
- return AmmosIFDataHeader.from_bytes(bytes)
+ return AmmosIFDataHeader.from_bytes(in_bytes)
def read_next_global_frame_body_extended_data_header(self):
@@ -66,11 +64,11 @@
header_size = AmmosExtendedIFDataHeader.HEADER_SIZE
- bytes = self.ammos_file.read(header_size)
+ in_bytes = self.ammos_file.read(header_size)
logger.info("\nReading global frame body extended data header\n")
- if ((not bytes) or (len(bytes) < header_size)):
+ if ((not in_bytes) or (len(in_bytes) < header_size)):
logger.debug("Can not read all %s bytes of global frame extended data header", header_size)
return None
- return AmmosExtendedIFDataHeader.from_bytes(bytes)
+ return AmmosExtendedIFDataHeader.from_bytes(in_bytes)
def read_next_if_data_body(self, number_of_data_blocks, data_length):
@@ -132,4 +130,5 @@
def payload(self):
+ """I return just the pure date (payload) from my container."""
return b"".join([each.global_frame_body.data_body.payload for each in self.container.global_frames])
@@ -161,18 +160,15 @@
"""I read the IF data embedded in an R&S AMMOS recording."""
- def __init__(self, file_name):
- super().__init__(file_name)
-
def read_next_global_frame_body_data_header(self):
-
+ """I read the next data header of a global frame body from current position in file."""
header_size = AmmosIFDataHeader.HEADER_SIZE
- bytes = self.ammos_file.read(header_size)
+ in_bytes = self.ammos_file.read(header_size)
logger.info("\nReading global frame body standard data header\n")
- if ((not bytes) or (len(bytes) < header_size)):
+ if ((not in_bytes) or (len(in_bytes) < header_size)):
logger.debug("Can not read all %s bytes of global frame body data header", header_size)
return None
- return AmmosIFDataHeader.from_bytes(bytes)
+ return AmmosIFDataHeader.from_bytes(in_bytes)
def read_next_global_frame_body_extended_data_header(self):
@@ -185,11 +181,11 @@
header_size = AmmosExtendedIFDataHeader.HEADER_SIZE
- bytes = self.ammos_file.read(header_size)
+ in_bytes = self.ammos_file.read(header_size)
logger.info("\nReading global frame body extended data header\n")
- if ((not bytes) or (len(bytes) < header_size)):
+ if ((not in_bytes) or (len(in_bytes) < header_size)):
logger.debug("Can not read all %s bytes of global frame extended data header", header_size)
return None
- return AmmosExtendedIFDataHeader.from_bytes(bytes)
+ return AmmosExtendedIFDataHeader.from_bytes(in_bytes)
def read_next_if_data_body(self, number_of_data_blocks, data_length):
@@ -251,4 +247,5 @@
def payload(self):
+ """I return just the pure date (payload) from my container."""
return b"".join([each.global_frame_body.data_body.payload for each in self.container.global_frames])I return just the pure date (payload) from my container.
def payload(self):
+ """I return just the pure date (payload) from my container."""
return b"".join([each.global_frame_body.data_body.payload for each in self.container.global_frames])I read the next data header of a global frame body from current position in file.
def read_next_global_frame_body_data_header(self):
-
+ """I read the next data header of a global frame body from current position in file."""
header_size = AmmosIFDataHeader.HEADER_SIZE
- bytes = self.ammos_file.read(header_size)
+ in_bytes = self.ammos_file.read(header_size)
logger.info("\nReading global frame body standard data header\n")
- if ((not bytes) or (len(bytes) < header_size)):
+ if ((not in_bytes) or (len(in_bytes) < header_size)):
logger.debug("Can not read all %s bytes of global frame body data header", header_size)
return None
- return AmmosIFDataHeader.from_bytes(bytes)@classmethod
def from_bytes(cls, byte_string):
- pdw = PDW.from_bytes(bytes[0:32])
- return IQDWTXBLOCK(pdw, bytes[32:])import numpy as np
-from datetime import datetime
-
+from datetime import datetime
+import numpy as np
class PPDWContainer():
@@ -93,8 +92,8 @@
def start_time(self):
- return min([each.time_of_arrival for each in self.signals])
+ return min(each.time_of_arrival for each in self.signals)
def end_time(self):
- return max([each.time_of_arrival for each in self.signals])
+ return max(each.time_of_arrival for each in self.signals)
def to_json(self):
@@ -196,8 +195,8 @@
def start_time(self):
- return min([each.time_of_arrival for each in self.signals])
+ return min(each.time_of_arrival for each in self.signals)
def end_time(self):
- return max([each.time_of_arrival for each in self.signals])
+ return max(each.time_of_arrival for each in self.signals)
def to_json(self):
@@ -295,5 +294,5 @@
def end_time(self):
- return max([each.time_of_arrival for each in self.signals])
+ return max(each.time_of_arrival for each in self.signals)
@@ -362,5 +361,5 @@
def start_time(self):
- return min([each.time_of_arrival for each in self.signals])
+ return min(each.time_of_arrival for each in self.signals)from pathlib import Path
+from ammosreader.PDW import PDW
+from ammosreader.PPDWContainer import PPDWContainer
+from ammosreader import logger
-from ammosreader.PDW import PDW
-from ammosreader.PPDWContainer import PPDWContainer
class PPDWReader():
+ """I read data from a file and return a PPDWContainer."""
def __init__(self, a_file):
@@ -48,8 +49,8 @@
assert len(current_bytes) == 32
if not current_bytes:
- # print('End of file detected')
+ logger.info('End of file detected')
break
if self.cursor + 32 >= len(self.content):
- # print('Can not read all 32 bytes of next PDW. EOF')
+ logger.info('Can not read all 32 bytes of next PDW. EOF')
break
self.container.add(PDW.from_bytes(current_bytes))
@@ -74,5 +75,5 @@
-
-
+
I read data from a file and return a PPDWContainer.
@@ -80,4 +81,5 @@
class PPDWReader():
+ """I read data from a file and return a PPDWContainer."""
def __init__(self, a_file):
@@ -95,8 +97,8 @@
assert len(current_bytes) == 32
if not current_bytes:
- # print('End of file detected')
+ logger.info('End of file detected')
break
if self.cursor + 32 >= len(self.content):
- # print('Can not read all 32 bytes of next PDW. EOF')
+ logger.info('Can not read all 32 bytes of next PDW. EOF')
break
self.container.add(PDW.from_bytes(current_bytes))
@@ -123,8 +125,8 @@
assert len(current_bytes) == 32
if not current_bytes:
- # print('End of file detected')
+ logger.info('End of file detected')
break
if self.cursor + 32 >= len(self.content):
- # print('Can not read all 32 bytes of next PDW. EOF')
+ logger.info('Can not read all 32 bytes of next PDW. EOF')
break
self.container.add(PDW.from_bytes(current_bytes))
Index: doc/ammosreader/index.html
===================================================================
--- doc/ammosreader/index.html (revision 04097cdb708531c6549b286c360cd998d5deb618)
+++ doc/ammosreader/index.html (revision 7752d1f99e91b74a1fdc11cfd15ae6166803378e)
@@ -81,5 +81,5 @@
ammosreader.AmmosContainer-
-
I store the content of an R&S Ammos file in a more accessible way.
+I provide an Ammos container.
ammosreader.AmmosExtendedAudioDataHeader