Module ammosreader.AmmosExtendedIFDataHeader
Expand source code
import struct
import numpy as np
from ammosreader.AmmosIFDataHeader import AmmosIFDataHeader
class AmmosExtendedIFDataHeader():
"""I implement an Ammos extended data header for IF data frames."""
HEADER_SIZE = 76
@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
Classes
class 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)
-
I implement an Ammos extended data header for IF data frames.
I create a new instance of myself using the above parameters.
Expand source code
class AmmosExtendedIFDataHeader(): """I implement an Ammos extended data header for IF data frames.""" HEADER_SIZE = 76 @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
Class variables
var HEADER_SIZE
Static methods
def from_bytes(bytes)
-
I return an AMMOS extended data header from given bytes.
Expand source code
@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)