-import struct
-import numpy
+"""I provide an AMMOS data header for audio data frames."""
-[docs]class AmmosAudioDemodType():
+import struct
+from ammosreader.AmmosConstants import AmmosAudioDemodType
- @classmethod
- @property
- def mapping(cls):
- return {0: 'FM', 1: 'AM', 5: 'ISB', 6: 'CW',
- 7: 'USB', 8: 'LSB', 256: 'DIGITAL',
- 0xFFFFFFFF: 'UNKNOWN'}
- def __init__(self, demod_type):
- self.demod_type = demod_type
+[docs]class AmmosAudioDataHeader:
+ """I implement an AMMOS data header for audio data frames."""
- def __str__(self):
- return AmmosAudioDemodType.mapping[self.demod_type]
-
-[docs]class AmmosAudioDataHeader():
+ HEADER_SIZE = 36 # 9 words
[docs] @classmethod
def from_bytes(cls, bytes):
+ """I return an AMMOS data header from given bytes."""
+ assert len(bytes) == cls.HEADER_SIZE
elements = struct.unpack('<IIQIIIII', bytes)
sample_rate = elements[0]
@@ -91,12 +84,14 @@
frequency = elements[2]
demod_bandwidth = elements[3]
- demod_type = elements[4]
- sample_count = elements[5]
- channel_count = elements[6]
+ demod_type = AmmosAudioDemodType(elements[4])
+ number_of_samples = elements[5]
+ number_of_channels = elements[6]
sample_size = elements[7]
return AmmosAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
- sample_count, channel_count, sample_size)
+ number_of_samples, number_of_channels, sample_size)
- def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type, sample_count, channel_count, sample_size):
+ def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type, number_of_samples,
+ number_of_channels, sample_size):
+ """I create a new instance of myself using the above parameters."""
self.sample_rate = sample_rate
self.status = status
@@ -104,9 +99,10 @@
self.demod_bandwidth = demod_bandwidth
self.demod_type = AmmosAudioDemodType(demod_type)
- self.sample_count = sample_count
- self.channel_count = channel_count
+ self.number_of_samples = number_of_samples
+ self.number_of_channels = number_of_channels
self.sample_size = sample_size
def __str__(self):
+ """I return the string representation of myself."""
return ("\nAmmosAudioDataHeader\n" +
"Sample rate:" + str(self.sample_rate) + "\n" +
@@ -115,6 +111,6 @@
"Demodulation bandwidth:" + str(self.demod_bandwidth) + "\n" +
"Demodulation type:" + str(self.demod_type) + "\n" +
- "Sample count:" + str(self.sample_count) + "\n" +
- "Channel count:" + str(self.channel_count) + "\n" +
+ "Sample count:" + str(self.number_of_samples) + "\n" +
+ "Channel count:" + str(self.number_of_channels) + "\n" +
"Sample size:" + str(self.sample_size) + "\n")
-"""I parse an R&S AMMOS recording."""
-
-import os
-
+"""I provide a specialized Ammos Reader for audio data."""
+import logging
+
+from ammosreader.AbstractAmmosReader import AbstractAmmosReader
from ammosreader.AmmosGlobalFrameBody import AmmosGlobalFrameBody
from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader
from ammosreader.AmmosExtendedAudioDataHeader import AmmosExtendedAudioDataHeader
-from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader
-from ammosreader.AmmosSingleFrame import AmmosSingleFrame
-from ammosreader.AmmosContainer import AmmosContainer
-
-
-[docs]class AmmosAudioReader():
+from ammosreader.AmmosAudioDataBody import AmmosAudioDataBody
+
+
+[docs]class AmmosAudioReader(AbstractAmmosReader):
"""I read the audio data embedded in an R&S AMMOS recording."""
-
- GLOBAL_HEADER_SIZE = 24 # 8 words
- STANDARD_AUDIO_DATA_HEADER_SIZE = 36 # 9 words
- EXTENDED_AUDIO_DATA_HEADER_SIZE = 44 # 11 words
def __init__(self, file_name):
@@ -90,65 +84,5 @@
:type file_name: str
"""
- self.file_name = file_name
- self.file = open(self.file_name, "rb")
- self.file_size = os.path.getsize(self.file_name)
-
- self.container = AmmosContainer(self.file_name, [])
-
- self.tags = {}
-
-[docs] def rewind_to_start(self):
- """I set the file pointer to the beginning of the file for the next operation."""
- self.file.seek(0)
-
-[docs] def add_tag(self, tag):
- """
- I add a tag to my tag list.
-
- :param tag: The tag to add to my tag list
- :type tag: dict
- """
- self.tags[tag.key] = tag.value
-
-[docs] 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:
- print("Reading single frame", frames_read, '...')
- current_frame = self.read_next_single_frame()
- if current_frame is not None:
- frames_read += 1
- self.container.add_frame(current_frame)
- if frames_read % 10000 == 0:
- print("#", end="")
- else:
- print("Frame:", frames_read+1, " incomplete")
- break
-
- print(len(self.container.global_frames), "frames read")
- return self.container
-
-[docs] 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
- """
- bytes = self.file.read(AmmosAudioReader.GLOBAL_HEADER_SIZE)
- print("Reading next global frame header")
- if ((not bytes) or (len(bytes) < AmmosAudioReader.GLOBAL_HEADER_SIZE)):
- print("Can not read all", AmmosAudioReader.GLOBAL_HEADER_SIZE, "bytes of global frame header")
- return None
-
- # FIXME: Catch exceptions and add some asserts
- current_global_frame_header = AmmosGlobalFrameHeader.from_bytes(bytes)
- # print("Current global frame header", current_global_frame_header)
- return current_global_frame_header
+ super().__init__(file_name)
[docs] def read_next_global_frame_body_data_header(self):
@@ -161,10 +95,11 @@
:rtype: AmmosAudioDataHeader
"""
- bytes = self.file.read(AmmosAudioReader.STANDARD_AUDIO_DATA_HEADER_SIZE)
-
- # print("\nReading global frame body standard data header\n")
- if ((not bytes) or (len(bytes) < AmmosAudioReader.STANDARD_AUDIO_DATA_HEADER_SIZE)):
- print("Can not read all", AmmosAudioReader.STANDARD_AUDIO_DATA_HEADER_SIZE,
- "bytes of global frame body data header")
+ header_size = AmmosAudioDataHeader.HEADER_SIZE
+
+ 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
return AmmosAudioDataHeader.from_bytes(bytes)
@@ -177,21 +112,22 @@
:rtype: AmmosExtendedAudioDataHeader
"""
- bytes = self.file.read(AmmosAudioReader.EXTENDED_AUDIO_DATA_HEADER_SIZE)
-
- if ((not bytes) or (len(bytes) < AmmosAudioReader.EXTENDED_AUDIO_DATA_HEADER_SIZE)):
- print("Can not read all ", AmmosAudioReader.EXTENDED_AUDIO_DATA_HEADER_SIZE,
- " bytes of global frame extended data header")
+ header_size = AmmosExtendedAudioDataHeader.HEADER_SIZE
+
+ bytes = self.ammos_file.read(header_size)
+
+ 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
return AmmosExtendedAudioDataHeader.from_bytes(bytes)
-[docs] def read_next_audio_data_body(self, sample_count, channel_count, sample_size):
+[docs] def read_next_audio_data_body(self, number_of_samples, number_of_channels, sample_size):
"""
I return the next audio data read from current position in file.
- :param sample_count: the number of samples per channel inside data body
- :type sample_count: int
-
- :param channel_count: number of channels (e.g. mono, stereo or even more)
- :type channel_count: int
+ :param number_of_samples: the number of samples per channel inside data body
+ :type number_of_samples: int
+
+ :param number_of_channels: number of channels (e.g. mono, stereo or even more)
+ :type number_of_channels: int
:param sample_size: sample size in bytes (1, 2 or 4 bytes)
@@ -201,67 +137,54 @@
:rtype: bytes
"""
- # FIXME: Describe the parameters better
-
- total = sample_count*channel_count*sample_size
-
- byte_string = self.file.read(total)
+ total = number_of_samples*number_of_channels*sample_size
+
+ byte_string = self.ammos_file.read(total)
if len(byte_string) != total:
- print("Can not read all", total, "bytes of data body")
- return None
- print([hex(c) for c in byte_string])
- return byte_string
-
-[docs] def read_next_global_frame_body(self, global_frame_header):
-
+ logging.debug("Can not read all %s bytes of data body", total)
+ return None
+ return AmmosAudioDataBody(byte_string, number_of_channels, number_of_samples, sample_size)
+
+[docs] def read_next_global_frame_body(self, data_header_length):
+ """
+ I return the next global frame body read from current position in file.
+
+ :param global_frame_header:
+ """
audio_data_header = None
- if global_frame_header.data_header_length == AmmosAudioReader.STANDARD_AUDIO_DATA_HEADER_SIZE:
- print("Read standard data header")
+ if data_header_length == AmmosAudioDataHeader.HEADER_SIZE:
+ logging.info("Read standard data header")
audio_data_header = self.read_next_global_frame_body_data_header()
- if global_frame_header.data_header_length == AmmosAudioReader.EXTENDED_AUDIO_DATA_HEADER_SIZE:
- print("Read extended data header")
+ logging.info("Data header length %s", data_header_length)
+ if data_header_length == AmmosExtendedAudioDataHeader.HEADER_SIZE:
+ logging.info("Read extended data header")
audio_data_header = self.read_next_global_frame_body_extended_data_header()
if audio_data_header is None:
- print("Data header missing or format unknown")
- return None
-
- audio_data_body = self.read_next_audio_data_body(audio_data_header.sample_count,
- audio_data_header.channel_count,
+ logging.debug("Data header missing or format unknown")
+ return None
+
+ audio_data_body = self.read_next_audio_data_body(audio_data_header.number_of_samples,
+ audio_data_header.number_of_channels,
audio_data_header.sample_size)
if audio_data_body is None:
- print("Data body missing")
+ logging.debug("Data body missing")
return None
return AmmosGlobalFrameBody(audio_data_header, audio_data_body)
-[docs] def read_next_single_frame(self):
-
- global_frame_header = self.read_next_global_frame_header()
-
- print(global_frame_header)
-
- if global_frame_header is None:
- print("Global frame header missing")
- return None
-
- if global_frame_header.data_header_length is None:
- print("Data header length empty")
- return None
-
- if global_frame_header.frame_type == 256:
- print("Audio Datastream found")
- global_frame_body = self.read_next_global_frame_body(global_frame_header)
- if global_frame_body is None:
- return None
- else:
- print("Unsupported frame type", global_frame_header.frame_type, "found")
- return None
-
- ammos_single_frame = AmmosSingleFrame(global_frame_header, global_frame_body)
- return ammos_single_frame
+[docs] def pcm_for_channel(self, a_channel):
+ """
+ I return the raw pcm audio data for a given channel.
+
+ :param a_channel: the channel I have to extract
+ :type a_channel: int
+
+ :rtype: bytes
+ """
+ return b"".join([each.global_frame_body.data_body.pcm_for_channel(a_channel) for each in self.container.global_frames])
-import socket
+"""I read a Ammos datastream from a socket."""
+
+import select
+import socket
from collections import deque
-
import numpy as np
from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader
@@ -74,5 +76,5 @@
[docs]class AmmosAudioSocketReader:
- def __init__(self, socket:socket.socket):
+ def __init__(self, socket:socket.socket, debug=True):
"""
Initializes the AmmosAudioSocketReader
@@ -80,4 +82,5 @@
Args:
socket (socket.socket): socket to read from
+ debug (bool): if true, prints debug information
"""
@@ -88,26 +91,37 @@
self.__socket = socket
- def __read_next_audio_data_body(self, sample_count:int, channel_count:int, sample_size:int) -> bytearray:
+ #
+ self.DEBUG_MODE = debug
+
+ def __get_next_data(self, byte_count: int) -> bytearray:
"""
- reads the next audio data body
+ Gets the next bytes from the socket, for example headers and body data.
Args:
- sample_count (int): amount of samples
- channel_count (int): amount of channels
- sample_size (int): size of a sample in bytes
+ byte_count (int): number of bytes to read
+
+ Raises:
+ TimeoutError: Raises TimeoutError if the socket does not serve data anymore
Returns:
- bytearray: contains the audio data
- """
-
- total = sample_count*channel_count*sample_size
+ bytearray: data from socket as bytearray
+ """
+
byte_array = []
-
- while len(b''.join(byte_array)) < total:
- byte_array.append(self.__socket.recv(total - len(b''.join(byte_array))))
-
- if len(b''.join(byte_array)) != total:
- print("Can not read all", total, "bytes of data body")
- return None
+
+ while len(b''.join(byte_array)) < byte_count:
+ if self.DEBUG_MODE:
+ print(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)
+
+ if not new_bytes:
+ raise TimeoutError("Socket timed out while reading data")
+
+ if self.DEBUG_MODE:
+ print(f"Got {len(new_bytes)} bytes of {byte_count - len(b''.join(byte_array))} ramining")
+
+ byte_array.append(new_bytes)
+
return b''.join(byte_array)
@@ -125,17 +139,26 @@
return np.frombuffer(audio_data_body, dtype=np.int16)
-[docs] def read_next_frame(self) -> tuple[bytearray, int]:
- """
- reads the next ammos audio frame
+[docs] def read_next_frame(self) -> tuple[np.ndarray, int]:
+ """Reads the next ammos audio frame from socket
+
+ Raises:
+ TimeoutError: Raisees TimeoutError if the socket does not serve data anymore
Returns:
- tuple[bytearray, int]: contains the audio data and the sample rate
- """
+ tuple[np.ndarray, int]: Contains the audio data and the sample rate
+ """
+
+ # get first byte of the day
+ self.__socket.settimeout(5)
+
+ new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
+ # raise Exception if socket does not return anything
+ if len(new_byte) < 1:
+ raise TimeoutError
+
#read loop
- byte = self.__socket.recv(1)
-
- while byte:
+ while new_byte:
#
- self.__magic_word_buffer.append(byte)
+ self.__magic_word_buffer.append(new_byte)
byte_array = b''.join(self.__magic_word_buffer)
@@ -144,43 +167,55 @@
ammos_global_header_buffer = list(self.__magic_word_buffer)
- 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_buffer.append(self.__get_next_data(20))
+ #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))
- print(ammos_global_header)
+ if self.DEBUG_MODE:
+ print(ammos_global_header)
if ammos_global_header.data_header_length == 44 and ammos_global_header.frame_type == 256:
- byte_array_header = []
- 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(b''.join(byte_array_header))
- print(ammos_extended_audio_data_header.sample_count, ammos_extended_audio_data_header.channel_count, ammos_extended_audio_data_header.sample_size)
- audio_body = self.__read_next_audio_data_body(ammos_extended_audio_data_header.sample_count,
- ammos_extended_audio_data_header.channel_count,
- ammos_extended_audio_data_header.sample_size)
+ byte_array_header = self.__get_next_data(44)
+ #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)
+ if self.DEBUG_MODE:
+ print(ammos_extended_audio_data_header.sample_count, ammos_extended_audio_data_header.channel_count, ammos_extended_audio_data_header.sample_size)
+ audio_body = self.__get_next_data(ammos_extended_audio_data_header.sample_count*
+ ammos_extended_audio_data_header.channel_count*
+ ammos_extended_audio_data_header.sample_size)
audio_array = self.__audio_data_body_to_numpy(audio_body)
- print(len(audio_array), len(audio_array)/ammos_extended_audio_data_header.sample_rate)
+ if self.DEBUG_MODE:
+ print(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 = []
- 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(b''.join(byte_array_header))
- print(ammos_audio_data_header.sample_count, ammos_audio_data_header.channel_count, ammos_audio_data_header.sample_size)
- audio_body = self.__read_next_audio_data_body(ammos_audio_data_header.sample_count,
- ammos_audio_data_header.channel_count,
- ammos_audio_data_header.sample_size)
+ byte_array_header = self.__get_next_data(36)
+ #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)
+ if self.DEBUG_MODE:
+ print(ammos_audio_data_header.sample_count, ammos_audio_data_header.channel_count, ammos_audio_data_header.sample_size)
+ audio_body = self.__get_next_data(ammos_extended_audio_data_header.sample_count*
+ ammos_extended_audio_data_header.channel_count*
+ ammos_extended_audio_data_header.sample_size)
audio_array = self.__audio_data_body_to_numpy(audio_body)
- print(len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
+ if self.DEBUG_MODE:
+ print(len(audio_array), len(audio_array)/ammos_audio_data_header.sample_rate)
return [audio_array, ammos_audio_data_header.sample_rate]
- byte = self.__socket.recv(1)
+ # get the next byte
+ self.__socket.settimeout(5)
+
+ new_byte = self.__socket.recv(1, socket.MSG_WAITALL)
+ # raise Exception if socket does not return anything
+ if len(new_byte) < 1:
+ raise TimeoutError
return None
@@ -193,5 +228,5 @@
© Copyright 2022, KIDZg.
- Created using Sphinx 4.4.0 ADC Theme .
+ Created using Sphinx 4.5.0 ADC Theme .
(Revision )
Index: doc/html/_modules/ammosreader/AmmosContainer.html
===================================================================
--- doc/html/_modules/ammosreader/AmmosContainer.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19)
+++ doc/html/_modules/ammosreader/AmmosContainer.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5)
@@ -69,28 +69,43 @@
def __init__(self, name, frames):
- self.name = name
- self.global_frames = frames
- self.tags = []
+ self.__name = name
+ self.__global_frames = frames
+ self.__tags = []
+
+ @property
+ def name(self):
+ return self.__name
+
+ @name.setter
+ def name(self, a_name):
+ self.__name = a_name
+
+ @property
+ def global_frames(self):
+ return self.__global_frames
+ self.__tags.append(tag)
+ self.__global_frames.append(frame)© Copyright 2022, KIDZg. - Created using Sphinx 4.4.0 ADC Theme . + Created using Sphinx 4.5.0 ADC Theme . (Revision )
Index: doc/html/_modules/ammosreader/AmmosExtendedAudioDataHeader.html =================================================================== --- doc/html/_modules/ammosreader/AmmosExtendedAudioDataHeader.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19) +++ doc/html/_modules/ammosreader/AmmosExtendedAudioDataHeader.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5) @@ -64,15 +64,20 @@
-import struct
-import numpy as np
+"""I provide an Ammos extended data header for audio data frames."""
+
+import struct
from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader
-[docs]class AmmosExtendedAudioDataHeader():
+[docs]class AmmosExtendedAudioDataHeader:
+
+ HEADER_SIZE = 44 # 11 words
[docs] @classmethod
def from_bytes(cls, bytes):
- standard_header = AmmosAudioDataHeader.from_bytes(bytes[0:36])
- extended_header_elements = struct.unpack('Q', bytes[36:])
+ """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:])
timestamp = extended_header_elements[0]
sample_rate = standard_header.sample_rate
@@ -81,12 +86,13 @@
demod_bandwidth = standard_header.demod_bandwidth
demod_type = standard_header.demod_type
- sample_count = standard_header.sample_count
- channel_count = standard_header.channel_count
+ number_of_samples = standard_header.number_of_samples
+ number_of_channels = standard_header.number_of_channels
sample_size = standard_header.sample_size
return AmmosExtendedAudioDataHeader(sample_rate, status, frequency, demod_bandwidth, demod_type,
- sample_count, channel_count, sample_size, timestamp)
+ number_of_samples, number_of_channels, sample_size, timestamp)
def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type,
- sample_count, channel_count, sample_size, timestamp):
+ number_of_samples, number_of_channels, sample_size, timestamp):
+ """I return a new AMMOS extended data header for audio frames built from given parameters."""
self.sample_rate = sample_rate
self.status = status
@@ -94,6 +100,6 @@
self.demod_bandwidth = demod_bandwidth
self.demod_type = demod_type
- self.sample_count = sample_count
- self.channel_count = channel_count
+ self.number_of_samples = number_of_samples
+ self.number_of_channels = number_of_channels
self.sample_size = sample_size
self.timestamp = timestamp
Index: doc/html/_modules/ammosreader/AmmosExtendedIFDataHeader.html
===================================================================
--- doc/html/_modules/ammosreader/AmmosExtendedIFDataHeader.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19)
+++ doc/html/_modules/ammosreader/AmmosExtendedIFDataHeader.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5)
@@ -70,9 +70,13 @@
[docs]class AmmosExtendedIFDataHeader():
+ """I implement an Ammos extended data header for IF data frames."""
+
+ HEADER_SIZE = 76
[docs] @classmethod
def from_bytes(cls, bytes):
- standard_header = AmmosIFDataHeader.from_bytes(bytes[0:56])
- extended_header_elements = struct.unpack('QQI', bytes[56:76])
+ """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])
block_count = standard_header.block_count
block_length = standard_header.block_length
@@ -99,5 +103,5 @@
bandwidth, sample_rate, interpolation, decimation, voltage_ref, stream_start, sample_counter,
antenna_correction):
-
+ """I create a new instance of myself using the above parameters."""
self.size = size
self.block_count = block_count
@@ -129,7 +133,5 @@
"Stream start:" + str(self.stream_start) + "\n" +
"Sample counter:" + str(self.sample_counter) + "\n" +
- "Antenna correction:" + str(self.antenna_correction) + "\n"
- )
-
+ "Antenna correction:" + str(self.antenna_correction) + "\n")
return output
Index: doc/html/_modules/ammosreader/AmmosGlobalFrameBody.html
===================================================================
--- doc/html/_modules/ammosreader/AmmosGlobalFrameBody.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19)
+++ doc/html/_modules/ammosreader/AmmosGlobalFrameBody.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5)
@@ -64,13 +64,37 @@
Source code for ammosreader.AmmosGlobalFrameBody
-[docs]class AmmosGlobalFrameBody():
+"""I provide an AMMOS global frame body."""
+[docs]class AmmosGlobalFrameBody():
+ """
+ I implement an AMMOS global frame body.
+
+ AMMOS frames can store data of various types.
+ The data is described in the data header.
+ The raw data is then stored in the data body.
+ """
def __init__(self, data_header, data_body):
- self.data_header = data_header
- self.data_body = data_body
+ self.__data_header = data_header
+ self.__data_body = data_body
+
+ @property
+ def data_header(self):
+ return self.__data_header
+
+ @data_header.setter
+ def data_header(self, a_data_header):
+ self.__data_header = a_data_header
+
+ @property
+ def data_body(self):
+ return self.__data_body
+
+ @data_body.setter
+ def data_body(self, data_bytes):
+ self.__data_body = data_bytes
[docs] def data_bytes_only(self):
- byte_string = ""
+ byte_string = b""
for each_block in self.data_body:
@@ -78,5 +102,5 @@
print("Block is nil")
- byte_string += each_block.if_data
+ byte_string += each_block.data
return byte_string
Index: doc/html/_modules/ammosreader/AmmosGlobalFrameHeader.html
===================================================================
--- doc/html/_modules/ammosreader/AmmosGlobalFrameHeader.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19)
+++ doc/html/_modules/ammosreader/AmmosGlobalFrameHeader.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5)
@@ -74,5 +74,5 @@
def from_bytes(cls, bytes):
- elements = struct.unpack('4s4s4s4s4s4s', bytes)
+ elements = struct.unpack('<4s4s4s4s4s4s', bytes)
magic_word = elements[0].hex()
@@ -93,8 +93,7 @@
def __init__(self, magic_word, frame_length, running_frame_number, frame_type, data_header_length, reserved):
- if magic_word != self.MAGIC_WORD:
+ if magic_word != type(self).MAGIC_WORD:
print("Wrong magic word")
self.magic_word = magic_word
- return None
else:
self.magic_word = magic_word
Index: doc/html/_modules/ammosreader/AmmosIFDataBlock.html
===================================================================
--- doc/html/_modules/ammosreader/AmmosIFDataBlock.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19)
+++ doc/html/_modules/ammosreader/AmmosIFDataBlock.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5)
@@ -64,9 +64,17 @@
Source code for ammosreader.AmmosIFDataBlock
-[docs]class AmmosIFDataBlock():
+[docs]class AmmosIFDataBlock:
- def __init__(self, if_datablock_header, if_data):
- self.if_datablock_header = if_datablock_header
- self.if_data = if_data
+ def __init__(self, header, data):
+ self.__header = header
+ self.__data = data
+
+ @property
+ def header(self):
+ return self.__header
+
+ @property
+ def data(self):
+ return self.__data
Index: doc/html/_modules/ammosreader/AmmosIFDataHeader.html
===================================================================
--- doc/html/_modules/ammosreader/AmmosIFDataHeader.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19)
+++ doc/html/_modules/ammosreader/AmmosIFDataHeader.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5)
@@ -64,12 +64,19 @@
Source code for ammosreader.AmmosIFDataHeader
-import struct
+"""I provide a Ammos data header for IF data frames."""
+
+import struct
import numpy as np
[docs]class AmmosIFDataHeader():
+ """I implement an Ammos data header for IF data frames."""
+
+ HEADER_SIZE = 56
[docs] @classmethod
def from_bytes(cls, bytes):
+ """I return an AMMOS data header from given bytes."""
+ assert len(bytes) == cls.HEADER_SIZE
elements = struct.unpack('<IIQIIIQIIIIi', bytes)
block_count = elements[0]
@@ -92,4 +99,5 @@
def __init__(self, block_count, block_length, timestamp, status, source_id, source_state, frequency,
bandwidth, sample_rate, interpolation, decimation, voltage_ref):
+ """I create a new instance of myself using the above parameters."""
self.block_count = block_count
self.block_length = block_length
@@ -104,7 +112,4 @@
self.decimation = decimation
self.voltage_ref = voltage_ref
-
-
def __str_(self):
Index: doc/html/_modules/ammosreader/AmmosIFReader.html
===================================================================
--- doc/html/_modules/ammosreader/AmmosIFReader.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19)
+++ doc/html/_modules/ammosreader/AmmosIFReader.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5)
@@ -64,155 +64,110 @@
Source code for ammosreader.AmmosIFReader
-import math
-import os
+"""I provide a specialized Ammos Reader for IF data."""
+import logging
+from ammosreader.AbstractAmmosReader import AbstractAmmosReader
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
-[docs]class AmmosIFReader():
+[docs]class AmmosIFReader(AbstractAmmosReader):
+ """I read the IF data embedded in an R&S AMMOS recording."""
def __init__(self, file_name):
-
- self.file_name = file_name
- self.file = open(self.file_name, "rb")
- self.file_size = os.path.getsize(self.file_name)
-
- self.container = AmmosContainer(self.file_name, [])
-
- self.tags = []
-
-
-
-
-
-[docs] def read_all_frames_left(self):
-
- frames_read = 0
-
- while True:
- print("Reading single frame", frames_read, '...')
- current_frame = self.read_next_single_frame()
- if current_frame is not None:
- frames_read += 1
- self.container.add_frame(current_frame)
- if frames_read % 10000 == 0:
- print("#", end="")
- else:
- print("Frame:", frames_read+1, " incomplete")
- break
-
- print(len(self.container.global_frames), "frames read")
-
-[docs] def read_next_global_frame_header(self):
- bytes = self.file.read(24)
- print("Reading next global frame header")
- if ((not bytes) or (len(bytes) < 24)):
- print("Can not read all 24 bytes of global frame header")
- return None
-
- return AmmosGlobalFrameHeader.from_bytes(bytes)
+ super().__init__(file_name)
[docs] def read_next_global_frame_body_data_header(self):
- bytes = self.file.read(56)
+ header_size = AmmosIFDataHeader.HEADER_SIZE
- # print("\nReading global frame body standard data header\n")
- if ((not bytes) or (len(bytes) < 56)):
- 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
+ return AmmosIFDataHeader.from_bytes(bytes)
+
+[docs] 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.
+
+ :return: the next Ammos Extended IF data header or None if incomplete
+ :rtype: AmmosExtendedIFDataHeader
+ """
+ header_size = AmmosExtendedIFDataHeader.HEADER_SIZE
+
+ 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
+ return AmmosExtendedIFDataHeader.from_bytes(bytes)
+
+[docs] 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.
+
+ :param number_of_data_blocks: the number of data blocks inside the body
+ :type number_of_data_blocks: int
+
+ :param data_length: the length of the raw data inside a single block
+ :type data_length: int
+ """
+ header_size = AmmosIFDataBlockHeader.HEADER_SIZE
+
+ data_body = AmmosIFDataBody()
+
+ 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:
+ logging.debug("Can not read all %s bytes of data body", total)
return None
- data_header = AmmosIFDataHeader.from_bytes(bytes)
- # print("Data header", data_header)
- return data_header
+ for i in range(0, number_of_data_blocks):
+ result = byte_string[i*block_length:(i*block_length+block_length)]
+ data_body.add_data_block(AmmosIFDataBlock(AmmosIFDataBlockHeader.from_bytes(result[0:header_size]),
+ result[header_size:]))
-
- bytes = self.file.read(76)
- # print("\nReading global frame body extended data header\n")
+[docs] def read_next_global_frame_body(self, data_header_length):
+ """
+ I return the next global frame body read from current position in file.
- if ((not bytes) or (len(bytes) < 76)):
- print("Can not read all ", 76, "bytes of global frame extended data header")
- return None
- extended_data_header = AmmosExtendedIFDataHeader.from_bytes(bytes)
- # print("Extended data header", extended_data_header)
- return extended_data_header
-
-[docs] def read_next_if_data_blocks(self, n, length):
-
- # FIXME: Describe the parameters better
-
- data_blocks = []
-
- block_length = 4 + length
-
- total = n*block_length
-
- byte_string = self.file.read(block_length)
-
- if len(byte_string) != total:
- print("Can not read all", total, "bytes of data body")
- return None
-
- for i in range(0, n):
- result = byte_string[i*block_length:(i*block_length+block_length)]
- data_blocks.append(AmmosIFDataBlock(result[0:4], result[4:]))
-
- return data_blocks
-
-[docs] def read_next_global_frame_body(self, global_frame_header):
-
+ :param data_header_length: the length of the data header
+ :type data_header_length: int
+ """
if_data_header = None
- if global_frame_header.data_header_length == 56:
+ if data_header_length == AmmosIFDataHeader.HEADER_SIZE:
if_data_header = self.read_next_global_frame_body_data_header()
- else:
+ 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
- if_data_body = self.read_next_if_data_blocks(if_data_header.block_count, if_data_header.block_length)
+ if_data_body = self.read_next_if_data_body(if_data_header.block_count, if_data_header.block_length)
if if_data_body is None:
- print("Data body missing")
+ logging.debug("Data body missing")
return None
return AmmosGlobalFrameBody(if_data_header, if_data_body)
-[docs] def read_next_single_frame(self):
-
- global_frame_header = self.read_next_global_frame_header()
-
- print("\nReading next global frame header\n", global_frame_header)
- # print("File pointer", self.file.tell())
-
- if global_frame_header is None:
- print("Global frame header missing")
- return None
-
- if global_frame_header.data_header_length is None:
- print("Data header length empty")
- return None
-
- if global_frame_header.frame_type == 2:
-
- global_frame_body = self.read_next_global_frame_body(global_frame_header)
- if global_frame_body is None:
- return None
-
- else:
- print("Unsupported frame type", global_frame_header.frame_type, "found")
- return None
-
- return AmmosSingleFrame(global_frame_header, global_frame_body)
+[docs] def data(self):
+ return b"".join([each.global_frame_body.data_body.data for each in self.container.global_frames])
Index: doc/html/_modules/ammosreader/AmmosSingleFrame.html
===================================================================
--- doc/html/_modules/ammosreader/AmmosSingleFrame.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19)
+++ doc/html/_modules/ammosreader/AmmosSingleFrame.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5)
@@ -67,6 +67,14 @@
def __init__(self, global_frame_header, global_frame_body):
- self.global_frame_header = global_frame_header
- self.global_frame_body = global_frame_body
+ self.__global_frame_header = global_frame_header
+ self.__global_frame_body = global_frame_body
+
+ @property
+ def global_frame_header(self):
+ return self.__global_frame_header
+
+ @property
+ def global_frame_body(self):
+ return self.__global_frame_body
[docs] def data(self):
@@ -83,5 +91,5 @@
"Global frame header\n" +
"-------------------\n" +
- "Frame tyoe:" + str(self.global_frame_header.frame_type) + "\n" +
+ "Frame type:" + str(self.global_frame_header.frame_type) + "\n" +
"Frame count:" + str(self.global_frame_header.running_frame_number) + "\n" +
"Data header length:" + str(self.global_frame_header.data_header_length) + " bytes\n" +
Index: doc/html/_modules/ammosreader/PDW.html
===================================================================
--- doc/html/_modules/ammosreader/PDW.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19)
+++ doc/html/_modules/ammosreader/PDW.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5)
@@ -64,9 +64,9 @@
Source code for ammosreader.PDW
-import struct
+"""I store the information of a single PDW block."""
+
+import struct
import math
import numpy as np
-
-# TODO: Use BitArray module in future versions
@@ -76,13 +76,10 @@
.. automethod:: __init__
-
"""
[docs] @classmethod
def from_bytes(cls, byte_string):
-
"""
-
- I create an instance of class PDW from data body (8 * 32 bits)
+ I create an instance of class PDW from data body (8 * 32 bits).
:param byte_string: a byte string containing a single data body read from a ppdw file
@@ -91,12 +88,10 @@
:return: an instance of class PDW with attributes set according to the data of a data body
:rtype: PDW
-
"""
-
assert(len(byte_string) == 32)
- parts = struct.unpack('Q4s4s4s4s4s4s', byte_string)
+ parts = struct.unpack('<Q4s4s4s4s4s4s', byte_string)
nanoseconds = (parts[0])
- time_of_arrival = np.datetime64(nanoseconds, 'ns') #datetime.datetime.utcfromtimestamp(seconds / 1000000000)
+ time_of_arrival = np.datetime64(nanoseconds, 'ns')
third_entry = bin(int.from_bytes(parts[1], byteorder='little'))
@@ -159,6 +154,6 @@
pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
sector, polarity, df_quality, elevation, azimuth, channel):
-
- """
+ r"""
+ I return an instance of an Pulse Data word.
:param time_of_arrival: nanoseconds since 1970-01-01 00:00:00
@@ -207,29 +202,32 @@
:return: An instance of class PDW with attributes set according to the data of a data body
:rtype: PDW
-
"""
-
- self.time_of_arrival = time_of_arrival
+ self.time_of_arrival = time_of_arrival #
self.pdw_format_identifier = pdw_format_identifier
- self.center_frequency = center_frequency
- self.is_valid = is_valid
- self.is_pulse = is_pulse
- self.level_unit = level_unit
+ self.center_frequency = center_frequency #
+ self.is_valid = is_valid #
+ self.is_pulse = is_pulse #
+ self.level_unit = level_unit #
self.signal_start_missing = signal_start_missing
self.signal_end_missing = signal_end_missing
- self.pulse_width = pulse_width
- self.frequency_shift_or_bandwidth = frequency_shift_or_bandwidth
- self.pulse_level_or_pulse_field_strength = pulse_level_or_pulse_field_strength
+ self.pulse_width = pulse_width #
+ self.frequency_shift_or_bandwidth = frequency_shift_or_bandwidth #
+ self.pulse_level_or_pulse_field_strength = pulse_level_or_pulse_field_strength #
self.region_of_interest = region_of_interest
self.azimuth_confidence = azimuth_confidence
- self.modulation = modulation
+ self.modulation = modulation #
self.sector = sector
self.polarity = polarity
- self.df_quality = df_quality
- self.elevation = elevation
+ self.df_quality = df_quality #
+ self.elevation = elevation #
self.azimuth = azimuth
- self.channel = channel
+ self.channel = channel #
def __str__(self):
+ """
+ I return the string representation of myself.
+
+ :rtype: str
+ """
output = ("Time of arrival: " + str(self.time_of_arrival) + "\n" +
"PDW Format identifier: " + str(self.pdw_format_identifier) + "\n" +
@@ -287,6 +285,20 @@
output += "Channel: " + str(self.channel) + "\n"
- return output
-
+ return output
+
+[docs] def to_json(self):
+ return {'TIMEOFARRIVAL': self.time_of_arrival.item(),
+ 'CENTERFREQUENCY': self.center_frequency,
+ 'VALID': self.is_valid,
+ 'PULSE': self.is_pulse,
+ 'PULSEWIDTH': self.pulse_width,
+ 'LEVELUNIT': self.level_unit,
+ 'FREQUENCYSHIFTORBANDWIDTH': self.frequency_shift_or_bandwidth,
+ 'PULSELEVELORFIELDSTRENGTH': self.pulse_level_or_pulse_field_strength,
+ 'MODULATION': self.modulation,
+ 'ELEVATION': self.elevation,
+ 'AZIMUTH': self.azimuth,
+ 'CHANNEL': self.channel
+ }
if __name__ == '__main__':
Index: doc/html/_modules/ammosreader/PPDWContainer.html
===================================================================
--- doc/html/_modules/ammosreader/PPDWContainer.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19)
+++ doc/html/_modules/ammosreader/PPDWContainer.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5)
@@ -64,18 +64,61 @@
Source code for ammosreader.PPDWContainer
-[docs]class PPDWContainer():
+import numpy as np
+from datetime import datetime
+
+[docs]class PPDWContainer():
"""
-
I store multiple signals imported from one or more .ppdw files
-
.. automethod:: __init__
-
"""
-[docs] def __init__(self, signals):
+ def __init__(self, name, signals=None):
- self.signals = signals
+ if signals is None:
+ self.signals = []
+ else:
+ self.signals = signals
+ def __str__(self):
+ return "\n".join(["Number of pulses:" + str(len(self.signals)),
+ "Start time:" + str(self.start_time()),
+ "End time:" + str(self.end_time())])
+
+
+
+[docs] def julian_date_string(self):
+ ts = (self.start_time() - np.datetime64('1970-01-01T00:00:00')) / np.timedelta64(1, 's')
+ time_tuple = datetime.utcfromtimestamp(ts).timetuple()
+ return str(time_tuple.tm_year)[2:] + str(time_tuple.tm_yday).zfill(3)
+
+[docs] def center_frequencies(self):
+ return list({each.center_frequency for each in self.signals})
+
+
+
+
+
+[docs] def bandwidths(self):
+ return list({each.frequency_shift_or_bandwidth for each in self.signals})
+
+
+
+
+
+[docs] def to_json(self):
+ return {'JULIANDATE': self.julian_date_string(),
+ 'STARTTIME': str(self.start_time()),
+ 'ENDTIME': str(self.end_time()),
+ # 'CENTERFREQUENCIES': self.center_frequencies(),
+ 'CHANNELS': self.channels()
+ # 'MODULATIONS': self.modulations(),
+ # 'BANDWIDTHS': self.bandwidths()
+ }
if __name__ == '__main__':
Index: doc/html/_modules/index.html
===================================================================
--- doc/html/_modules/index.html (revision a4a96fdcee5eddf20bb3729345b6f90bea726a19)
+++ doc/html/_modules/index.html (revision 299c79c00241e6f0fbb1e0e96048951cc5b871f5)
@@ -86,5 +86,5 @@
© Copyright 2022, KIDZg.
- Created using Sphinx 4.4.0 ADC Theme .
+ Created using Sphinx 4.5.0 ADC Theme .
(Revision )