+ ammosreader.AmmosAudioDataHeader
+ AmmosReader 0.314 documentation
+
+"""I provide an AMMOS data header for audio data frames."""
+
+import struct
+from ammosreader.AmmosConstants import AmmosAudioDemodType
+
+
+[docs]class AmmosAudioDataHeader:
+ """I implement an AMMOS data header for audio data frames."""
+
+ 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]
+ status = elements[1]
+ frequency = elements[2]
+ demod_bandwidth = elements[3]
+ 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,
+ number_of_samples, number_of_channels, 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
+ self.frequency = frequency
+ self.demod_bandwidth = demod_bandwidth
+ self.demod_type = AmmosAudioDemodType(demod_type)
+ 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" +
+ "Status:" + str(self.status) + "\n" +
+ "Frequency:" + str(self.frequency) + "\n" +
+ "Demodulation bandwidth:" + str(self.demod_bandwidth) + "\n" +
+ "Demodulation type:" + str(self.demod_type) + "\n" +
+ "Sample count:" + str(self.number_of_samples) + "\n" +
+ "Channel count:" + str(self.number_of_channels) + "\n" +
+ "Sample size:" + str(self.sample_size) + "\n")
+
+ ammosreader.AmmosAudioDataHeader
+ AmmosReader 0.314 documentation
+
+"""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.AmmosAudioDataBody import AmmosAudioDataBody
+
+
+[docs]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)
+
+[docs] def read_next_global_frame_body_data_header(self):
+ """
+ I return the next global frame body data header from current position in file.
+
+ :param data_header_size: the number of bytes to read
+ :type data_header_size: int
+ :return: the next Ammos Audio Data header or None if incomplete
+ :rtype: AmmosAudioDataHeader
+ """
+ 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)
+
+[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 Audio Extended Data header or None if incomplete
+ :rtype: AmmosExtendedAudioDataHeader
+ """
+ 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, number_of_samples, number_of_channels, sample_size):
+ """
+ I return the next audio data read from current position in file.
+
+ :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)
+ :type sample_size: int
+
+ :return: the next audio data or None if incomplete
+ :rtype: bytes
+ """
+ total = number_of_samples*number_of_channels*sample_size
+
+ byte_string = self.ammos_file.read(total)
+
+ if len(byte_string) != total:
+ 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 data_header_length == AmmosAudioDataHeader.HEADER_SIZE:
+ logging.info("Read standard data header")
+ audio_data_header = self.read_next_global_frame_body_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:
+ 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:
+ logging.debug("Data body missing")
+ return None
+
+ return AmmosGlobalFrameBody(audio_data_header, audio_data_body)
+
+[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])
+
+ ammosreader.AmmosAudioReader
+ AmmosReader 0.314 documentation
+
+"""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
+from ammosreader.AmmosExtendedAudioDataHeader import AmmosExtendedAudioDataHeader
+from ammosreader.AmmosGlobalFrameHeader import AmmosGlobalFrameHeader
+
+
+[docs]class AmmosAudioSocketReader:
+ def __init__(self, socket:socket.socket, debug=True):
+ """
+ Initializes the AmmosAudioSocketReader
+
+ Args:
+ socket (socket.socket): socket to read from
+ debug (bool): if true, prints debug information
+ """
+
+ #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
+
+ #
+ self.DEBUG_MODE = debug
+
+ def __get_next_data(self, byte_count: int) -> bytearray:
+ """
+ Gets the next bytes from the socket, for example headers and body data.
+
+ Args:
+ byte_count (int): number of bytes to read
+
+ Raises:
+ TimeoutError: Raises TimeoutError if the socket does not serve data anymore
+
+ Returns:
+ bytearray: data from socket as bytearray
+ """
+
+ byte_array = []
+
+ 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)
+
+ def __audio_data_body_to_numpy(self, audio_data_body:bytearray) -> np.ndarray:
+ """
+ converts the audio data body to a numpy array
+
+ Args:
+ audio_data_body (bytearray): audio data from audio data body
+
+ Returns:
+ np.ndarray: audio data as numpy array
+ """
+
+ return np.frombuffer(audio_data_body, dtype=np.int16)
+
+[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[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
+ while new_byte:
+ #
+ self.__magic_word_buffer.append(new_byte)
+ byte_array = b''.join(self.__magic_word_buffer)
+
+ if byte_array.hex() == '726574fb':
+ #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:
+ # 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))
+ 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 = 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)
+ 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 = 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)
+ 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]
+
+ # 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
+
+ ammosreader.AmmosAudioSocketReader
+ AmmosReader 0.314 documentation
+
+"""I store the content of an R&S Ammos file in a more accessible way."""
+
+[docs]class AmmosContainer():
+
+ def __init__(self, name, frames):
+ 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
+
+
+
+
+
+[docs] def size(self):
+ return sum([each.global_frame_header.frame_length for each in self.__global_frames])
+
+[docs] def frequencies(self):
+ return set(list(filter(lambda frame: frame.global_frame_body.data_header.frequency, self.__global_frames)))
+
+ def __str__(self):
+ start_time = self.__global_frames[0].global_frame_body.data_header.timestamp
+ end_time = self.__global_frames[-1].global_frame_body.data_header.timestamp
+
+ frq = str(self.__global_frames[0].global_frame_body.data_header.frequency)
+
+ return ("Start time: " + str(start_time) +
+ "\nEnd time : " + str(end_time) + "\nFrequencies: " + frq)
+
+[docs] def data_only(self):
+ return [each.global_frame_body.data_body.data for each in self.__global_frames]
+
+ ammosreader.AmmosContainer
+ AmmosReader 0.314 documentation
+
+"""I provide an Ammos extended data header for audio data frames."""
+
+import struct
+from ammosreader.AmmosAudioDataHeader import AmmosAudioDataHeader
+
+
+[docs]class AmmosExtendedAudioDataHeader:
+
+ HEADER_SIZE = 44 # 11 words
+
+[docs] @classmethod
+ def from_bytes(cls, 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:])
+ timestamp = extended_header_elements[0]
+ sample_rate = standard_header.sample_rate
+ status = standard_header.status
+ frequency = standard_header.frequency
+ demod_bandwidth = standard_header.demod_bandwidth
+ demod_type = standard_header.demod_type
+ 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,
+ number_of_samples, number_of_channels, sample_size, timestamp)
+
+ def __init__(self, sample_rate, status, frequency, demod_bandwidth, demod_type,
+ 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
+ self.frequency = frequency
+ self.demod_bandwidth = demod_bandwidth
+ self.demod_type = demod_type
+ self.number_of_samples = number_of_samples
+ self.number_of_channels = number_of_channels
+ self.sample_size = sample_size
+ self.timestamp = timestamp
+
+ ammosreader.AmmosExtendedAudioDataHeader
+ AmmosReader 0.314 documentation
+
+import struct
+import numpy as np
+from ammosreader.AmmosIFDataHeader import AmmosIFDataHeader
+
+
+[docs]class AmmosExtendedIFDataHeader():
+ """I implement an Ammos extended data header for IF data frames."""
+
+ HEADER_SIZE = 76
+
+[docs] @classmethod
+ def from_bytes(cls, 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])
+ block_count = standard_header.block_count
+ block_length = standard_header.block_length
+ timestamp = standard_header.timestamp
+ status = standard_header.status
+ source_id = standard_header.source_id
+ source_state = standard_header.source_state
+ frequency = standard_header.frequency
+ bandwidth = standard_header.bandwidth
+ sample_rate = standard_header.sample_rate
+ interpolation = standard_header.interpolation
+ decimation = standard_header.decimation
+ voltage_ref = standard_header.voltage_ref
+ stream_start = np.datetime64(int(extended_header_elements[0]), 'ns')
+ sample_counter = extended_header_elements[1]
+ antenna_correction = extended_header_elements[2]
+ size = len(bytes)
+ return AmmosExtendedIFDataHeader(size, block_count, block_length, timestamp, status, source_id,
+ source_state, frequency, bandwidth, sample_rate, interpolation,
+ decimation, voltage_ref, stream_start, sample_counter,
+ antenna_correction)
+
+ def __init__(self, size, block_count, block_length, timestamp, status, source_id, source_state, frequency,
+ 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
+ self.block_length = block_length
+ self.timestamp = timestamp
+ self.status = status
+ self.source_id = source_id
+ self.source_state = source_state
+ self.frequency = frequency
+ self.bandwidth = bandwidth
+ self.sample_rate = sample_rate
+ self.interpolation = interpolation
+ self.decimation = decimation
+ self.voltage_ref = voltage_ref
+ self.stream_start = stream_start
+ self.sample_counter = sample_counter
+ self.antenna_correction = antenna_correction
+
+ def __str__(self):
+ output = (
+ "\nGlobal frame body data header\n" +
+ "-----------------------------\n" +
+ "Block count:" + str(self.block_count) + "\n" +
+ "Block length:" + str(self.block_length) + "\n" +
+ "Time stamp:" + str(self.timestamp) + "\n" +
+ "Frequency:" + str(self.frequency) + "\n" +
+ "Bandwidth:" + str(self.bandwidth) + "\n" +
+ "Sample rate:" + str(self.sample_rate) + "\n" +
+ "Stream start:" + str(self.stream_start) + "\n" +
+ "Sample counter:" + str(self.sample_counter) + "\n" +
+ "Antenna correction:" + str(self.antenna_correction) + "\n")
+ return output
+
+ ammosreader.AmmosExtendedIFDataHeader
+ AmmosReader 0.314 documentation
+
+"""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
+
+ @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 = b""
+
+ for each_block in self.data_body:
+ if not each_block:
+ print("Block is nil")
+
+ byte_string += each_block.data
+
+ return byte_string
+
+ ammosreader.AmmosGlobalFrameBody
+ AmmosReader 0.314 documentation
+
+import struct
+
+
+[docs]class AmmosGlobalFrameHeader():
+
+ MAGIC_WORD = "726574fb"
+
+[docs] @classmethod
+ def from_bytes(cls, bytes):
+
+ elements = struct.unpack('<4s4s4s4s4s4s', bytes)
+
+ magic_word = elements[0].hex()
+
+ frame_length = (int.from_bytes(elements[1], byteorder='little')*4)
+
+ running_frame_number = (int.from_bytes(elements[2], byteorder='little'))
+
+ frame_type = (int.from_bytes(elements[3], byteorder='little'))
+
+ data_header_length = 4 * int.from_bytes(elements[4], byteorder='little')
+
+ reserved = elements[5]
+
+ return AmmosGlobalFrameHeader(magic_word, frame_length, running_frame_number,
+ frame_type, data_header_length, reserved)
+
+ def __init__(self, magic_word, frame_length, running_frame_number, frame_type, data_header_length, reserved):
+
+ if magic_word != type(self).MAGIC_WORD:
+ print("Wrong magic word")
+ self.magic_word = magic_word
+ else:
+ self.magic_word = magic_word
+ self.frame_length = frame_length
+ self.running_frame_number = running_frame_number
+ self.frame_type = frame_type
+ self.data_header_length = data_header_length
+ self.reserved = reserved
+
+ def __str__(self):
+ output = ("Global frame header info\n" +
+ "------------------------\n" +
+ "Magic word:" + str(self.magic_word) + "\n" +
+ "Frame length:" + str(self.frame_length) + "\n" +
+ "Running frame:" + str(self.running_frame_number) + "\n" +
+ "Frame Type:" + str(self.frame_type) + "\n" +
+ "Data header length:" + str(self.data_header_length) + "\n")
+ return output
+
+
+
+ ammosreader.AmmosGlobalFrameHeader
+ AmmosReader 0.314 documentation
+
+[docs]class AmmosIFDataBlock:
+
+ 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
+
+ ammosreader.AmmosIFDataBlock
+ AmmosReader 0.314 documentation
+
+"""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]
+ block_length = int(elements[1])*4
+ timestamp = np.datetime64(int(elements[2])*1000, 'ns')
+ status = elements[3]
+ source_id = elements[4]
+ source_state = elements[5]
+ frequency = elements[6]
+ bandwidth = elements[7]
+ sample_rate = elements[8]
+ interpolation = elements[9]
+ decimation = elements[10]
+ voltage_ref = elements[11]
+
+ return AmmosIFDataHeader(block_count, block_length, timestamp, status, source_id,
+ source_state, frequency, bandwidth, sample_rate,
+ interpolation, decimation, voltage_ref)
+
+ 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
+ self.timestamp = timestamp
+ self.status = status
+ self.source_id = source_id
+ self.source_state = source_state
+ self.frequency = frequency
+ self.bandwidth = bandwidth
+ self.sample_rate = sample_rate
+ self.interpolation = interpolation
+ self.decimation = decimation
+ self.voltage_ref = voltage_ref
+
+ def __str_(self):
+ output = ("\nGlobal frame body data header\n" +
+ "-----------------------------\n" +
+ "Block count:" + str(self.block_count) + "\n" +
+ "Block length:" + str(self.block_length) + "\n" +
+ "Time stamp:" + str(self.timestamp) + "\n" +
+ "Frequency:" + str(self.frequency) + "\n" +
+ "Bandwidth:" + str(self.bandwidth) + "\n" +
+ "Sample rate:" + str(self.sample_rate) + "\n")
+ return output
+
+ ammosreader.AmmosIFDataHeader
+ AmmosReader 0.314 documentation
+
+"""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.AmmosIFDataBody import AmmosIFDataBody
+from ammosreader.AmmosIFDataBlock import AmmosIFDataBlock
+from ammosreader.AmmosIFDataBlockHeader import AmmosIFDataBlockHeader
+
+
+[docs]class AmmosIFReader(AbstractAmmosReader):
+ """I read the IF data embedded in an R&S AMMOS recording."""
+
+ def __init__(self, file_name):
+ super().__init__(file_name)
+
+[docs] def read_next_global_frame_body_data_header(self):
+
+ header_size = AmmosIFDataHeader.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 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
+
+ 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:]))
+
+ return data_body
+
+[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 data_header_length: the length of the data header
+ :type data_header_length: int
+ """
+ if_data_header = None
+
+ if data_header_length == AmmosIFDataHeader.HEADER_SIZE:
+ if_data_header = self.read_next_global_frame_body_data_header()
+ 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:
+ logging.debug("Data header missing")
+ return None
+
+ 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:
+ logging.debug("Data body missing")
+ return None
+
+ return AmmosGlobalFrameBody(if_data_header, if_data_body)
+
+[docs] def data(self):
+ return b"".join([each.global_frame_body.data_body.data for each in self.container.global_frames])
+
+ ammosreader.AmmosIFReader
+ AmmosReader 0.314 documentation
+
+[docs]class AmmosSingleFrame():
+
+ def __init__(self, global_frame_header, 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 size_correct(self):
+ return (self.global_frame_header.frame_length == (24 + self.global_frame_header.data_header_length +
+ (self.global_frame_body.data_header.block_count *
+ (self.global_frame_body.data_header.block_length + 4))))
+ # FIXME: Use str method instead
+
+ def __str__(self):
+ output = (
+ "Global frame header\n" +
+ "-------------------\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" +
+ "Frame length:" + str(self.global_frame_header.frame_length) + " bytes\n"
+ )
+ return output
+
+ ammosreader.AmmosSingleFrame
+ AmmosReader 0.314 documentation
+
+ ammosreader.IQDWTXBlock
+ AmmosReader 0.314 documentation
+
+"""I store the information of a single PDW block."""
+
+import struct
+import math
+import numpy as np
+
+
+[docs]class PDW():
+ """
+ I store information from a single ppdw data block.
+
+ .. automethod:: __init__
+ """
+
+[docs] @classmethod
+ def from_bytes(cls, byte_string):
+ """
+ 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
+ :type byte_string: byte string
+
+ :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)
+ nanoseconds = (parts[0])
+ time_of_arrival = np.datetime64(nanoseconds, 'ns')
+
+ third_entry = bin(int.from_bytes(parts[1], byteorder='little'))
+ padding = 32-len(str(third_entry)[2:])
+ third_entry_bit_string = "0" * padding + str(third_entry)[2:]
+ pdw_format_identifier = int(third_entry_bit_string[0:6], 2)
+ center_frequency = int(third_entry_bit_string[5:32], 2)
+
+ fourth_entry = bin(int.from_bytes(parts[2], byteorder='little'))
+ padding = 32-len(str(fourth_entry)[2:])
+ fourth_entry_bit_string = "0" * padding + str(fourth_entry)[2:]
+ is_valid = bool(int(fourth_entry_bit_string[0]))
+ is_pulse = bool(int(fourth_entry_bit_string[1]))
+ level_unit = int(fourth_entry_bit_string[2])
+ signal_start_missing = bool(int(fourth_entry_bit_string[3]))
+ signal_end_missing = bool(int(fourth_entry_bit_string[4]))
+ pulse_width = int(fourth_entry_bit_string[7:33], 2)
+
+ fifth_entry = bin(int.from_bytes(parts[3], byteorder='little'))
+ padding = 32-len(str(fifth_entry)[2:])
+ fifth_entry_bit_string = "0" * padding + str(fifth_entry)[2:]
+ frequency_shift_or_bandwidth = int(fifth_entry_bit_string[0:20], 2)
+ # FIXME: You have to scale me to the range from -200.0 to 200.0 in 0.1 steps
+ pulse_level_or_pulse_field_strength = math.ceil(int(fifth_entry_bit_string[20:32], 2)) / 10
+
+ sixth_entry = bin(int.from_bytes(parts[4], byteorder='little'))
+ padding = 32-len(str(sixth_entry)[2:])
+ sixth_entry_bit_string = "0" * padding + str(sixth_entry)[2:]
+ region_of_interest = bool(int(sixth_entry_bit_string[0]))
+ # FIXME: You have to scale me to a range from 0.0 to 6.2 in steps of 0.1 - 6.3 means unknown
+ azimuth_confidence = math.ceil(int(sixth_entry_bit_string[1:7], 2)) / 10
+ modulations = {0: 'Unknown', 1: 'Unmodulated', 2: 'FM', 3: 'LFM', 4: 'PSK-2', 5: 'PSK-3', 6: 'PSK-4',
+ 7: 'PSK-m', 8: 'NLFM', 9: 'SFM', 10: 'TFM', 11: 'Pulse too short'}
+ modulation = modulations[int(sixth_entry_bit_string[7:12], 2)]
+ sector = int(sixth_entry_bit_string[28:32], 2)
+
+ seventh_entry = bin(int.from_bytes(parts[5], byteorder='little'))
+ padding = 32-len(str(seventh_entry)[2:])
+ seventh_entry_bit_string = "0" * padding + str(seventh_entry)[2:]
+ polarities = {0: 'Horizontal/Unknown', 1: 'Vertical', 2: 'Counter clockwise', 3: 'Clockwise'}
+ polarity = polarities[int(seventh_entry_bit_string[0:2], 2)]
+ df_quality = int(seventh_entry_bit_string[2:9], 2)
+ # FIXME: You have to scale me from -90 to 90 in 0.1 degree steps
+ elevation = int(seventh_entry_bit_string[9:20], 2)
+ # FIXME: You have to check me for a range from 0.0 to 359.9 in steps of 0.1
+ azimuth = 0.1 * (int(seventh_entry_bit_string[20:32], 2))
+
+ eighth_entry = bin(int.from_bytes(parts[5], byteorder='little'))
+ padding = 32-len(str(eighth_entry)[2:])
+ eighth_entry_bit_string = "0" * padding + str(eighth_entry)[2:]
+ channel = int(eighth_entry_bit_string[0:4], 2)
+
+ return PDW(time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse, level_unit,
+ signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
+ pulse_level_or_pulse_field_strength, region_of_interest, azimuth_confidence, modulation,
+ sector, polarity, df_quality, elevation, azimuth, channel)
+
+[docs] def __init__(self, time_of_arrival, pdw_format_identifier, center_frequency, is_valid, is_pulse,
+ level_unit, signal_start_missing, signal_end_missing, pulse_width, frequency_shift_or_bandwidth,
+ 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
+ :type time_of_arrival: Integer
+ :param pdw_format: format code
+ :type pdw_format: Integer
+ :param center_frequency: center frequency in KHz
+ :type center_frequency: Integer
+ :param is_valid: flag to mark if pdw data body is valid
+ :type is_valid: Boolean
+ :param is_pulse: flag to mark if pdw data body contains a pulse or a continuous wave signal
+ :type is_pulse: Boolean
+ :param level_unit: 0 means dBµV - 1 means dBµV/m
+ :type level_unit: Integer
+ :param signal_start_missing: signal started before time of arrival
+ :type signal_start_missing: Boolean
+ :param signal_end_missing: signal stops after time of arrival
+ :type signal_end_missing: Boolean
+ :param pulse_width: pulse width in nanoseconds - Zero if no valid pulse detected
+ :type pulse_width: Integer
+ :param frequency_shift_or_bandwidth: Value in KHz - Value set to 1048575 means Unknown
+ :type frequency_shift_or_bandwidth: Integer
+ :param pulse_level_or_pulse_field_strength: Pulse level or Pulse Field Strength depending on level_unit \
+ (-200.0...200.0) in 0.1 steps / minus 204.8 means no valid level detected
+ :type pulse_level_or_pulse_field_strength: Float
+ :param region_of_interest: Marks if signal is from region of interest
+ :type region_of_interest: Boolean
+ :param azimuth_confidence: degree in steps of 0.1 (0.0-6.2) / 6.3 means confidence unknown
+ :type azimuth_confidence: Float
+ :param modulation: type of modulation (e.g. PSK-2, PSK-4, FM etc.)
+ :type modulation: String
+ :param sector: reference antenna sector (0-15)
+ :type sector: Integer
+ :param polarity: Horizontal, Vertical, Clockwise, Counter clockwise
+ :type polarity: String
+ :param df_quality: Direction finding quality in percent (0-100) - Zero means unknown
+ :type df_quality: Integer
+ :param elevation: elevation of incoming signal (from -90 to 90 degree) in steps of 0.1 degree \
+ minus 102.4 means unknown
+ :type elevation: Float
+ :param azimuth: azimuth of incoming signal (from 0 to 359.9 degree) in steps of 0.1 degree \
+ plus 409.5 means unknown
+ :type azimuth: Float
+ :param channel: detecting channel (0-16) - Zero means unknown
+ :type channel: Integer
+ :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.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.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.region_of_interest = region_of_interest
+ self.azimuth_confidence = azimuth_confidence
+ self.modulation = modulation #
+ self.sector = sector
+ self.polarity = polarity
+ self.df_quality = df_quality #
+ self.elevation = elevation #
+ self.azimuth = azimuth
+ 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" +
+ "Center frequency: " + str(self.center_frequency) + " KHz\n")
+
+ if self.is_valid:
+ output += "Signal: Valid\n"
+ else:
+ output += "Signal: Invalid\n"
+
+ if self.is_pulse:
+ output += "Signal type: Pulse\n"
+ else:
+ output += "Signal type: Continuous wave\n"
+
+ if self.level_unit == 1:
+ output += "Pulse level: " + str(self.pulse_level_or_pulse_field_strength) + " dbµV\n"
+ else:
+ output += "Pulse field strength: " + str(self.pulse_level_or_pulse_field_strength) + " dbµV/meter\n"
+
+ output += ("Pulse width: " + str(self.pulse_width) + " nanoseconds\n" +
+ "Frequency shift or bandwidth: " + str(self.frequency_shift_or_bandwidth) + " KHz\n")
+
+ if self.region_of_interest:
+ output += "Region of interest: Yes\n"
+ else:
+ output += "Region of interest: No\n"
+
+ if self.azimuth_confidence == 6.3:
+ output += "Azimuth confidence: Invalid\n"
+ else:
+ output += "Azimuth confidence: " + str(self.azimuth_confidence) + " degree\n"
+
+ output += "Modulation: " + str(self.modulation) + "\n"
+
+ if self.sector == 0:
+ output += "Sector: Unknown\n"
+ else:
+ output += "Sector:" + str(self.sector) + "\n"
+
+ output += "Polarity: " + str(self.polarity) + "\n"
+
+ output += "DF quality: " + str(self.df_quality) + " %\n"
+
+ if self.elevation == 1024:
+ output += "Elevation: Unknown\n"
+ else:
+ output += "Elevation: " + str(self.elevation) + " degree\n"
+
+ if self.azimuth == 409.5:
+ output += "Azimuth: Unknown\n"
+ else:
+ output += "Azimuth: " + str(self.azimuth) + " degree\n"
+
+ output += "Channel: " + str(self.channel) + "\n"
+
+ 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__':
+ pass
+
+ ammosreader.PDW
+ AmmosReader 0.314 documentation
+
+import numpy as np
+from datetime import datetime
+
+
+[docs]class PPDWContainer():
+ """
+ I store multiple signals imported from one or more .ppdw files
+ .. automethod:: __init__
+ """
+
+ def __init__(self, name, signals=None):
+
+ 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__':
+ pass
+
+ ammosreader.PPDWContainer
+ AmmosReader 0.314 documentation
+