Different classes ( Blocks ) are defined here, given the type of block shown in the `.bin` file
- Use DataClass as Basic Block
- Simplification of methods that just return properties.
- Elimination of Code Repetition
- Loops Vectoring using numpy
- Change from Inheritance to Composition and attribute delegation
BaseBlock - Common Attributes for all Blocks
A block is a piece of the .bin file with a known beginning and end and that contains different types of information.
It has several fields: type, header, data and footer.
Each field has lengths and information defined in the documentation.
It is common to all blocks and all classes that define the following blocks receive this base block in the constructor.
class TimedNonSpectral(GetAttr):
"""Block with time attributes applied to non spectral blocks
Implementa o mapeamento de funções dos blocos que possuem os seguintes campos
(na mesma posição do vetor binario):F0 a F2
F0 = (4 bytes) _walldate = Wall Clock Start Date of measurements
F1 = (4 bytes) WALLTIME = Wall Clock Start Time
F2 = (4u bytes) WALLNANO = Wall Clock Start Time Nanoseconds
"""
def __init__(self, block: BaseBlock):
"""This implementation substitues inheritance of the class Block by Composition
The attributes which belong to Block are accessed normally as if it was Inherited
"""
self.default = block
@cached
def _walldate(self) -> str:
"""F0 = (4 bytes) _walldate = Wall Clock Start Date of measurements"""
date = bin2date(self.data[:4])
return (
f"{str(date[2]).zfill(4)}-{str(date[1]).zfill(2)}-{str(date[0]).zfill(2)}"
)
@cached
def _walltime(self) -> str:
"""F1 = (4 bytes) WALLTIME = Wall Clock Start Time"""
time = bin2time(self.data[4:8])
return (
f"{str(time[0]).zfill(2)}:{str(time[1]).zfill(2)}:{str(time[2]).zfill(2)}"
)
@cached
def _wallnano(self) -> int:
"""F2 = (4u bytes) WALLNANO = Wall Clock Start Time Nanoseconds"""
return bin2int(self.data[8:12], False)
@cached
def wallclock_datetime(self) -> np.datetime64:
"""Returns the wallclock datetime"""
return np.datetime64(
f"{self._walldate}T{self._walltime}.{int(self._wallnano/1000)}"
)
class TimedSpectral(GetAttr):
"""Common Atributes of Block with Time Info and Spectral Data
F0 = (4 bytes) _walldate = Wall Clock Start Date of measurements
F1 = (4 bytes) WALLTIME = Wall Clock Start Time
F2 = (4u bytes) WALLNANO = Wall Clock Start Time Nanoseconds
F3 = (2u bytes) STARTMEGA = Start Frequency MHz
F4 = (4 bytes) STARTMILLI = Start Frequency mHz
F5 = (2u bytes) STOPMEGA = Stop Frequency MHz
F6 = (4 bytes) STOPMILLI = Stop Frequency mHz
F7 = (2u bytes) STARTCHAN = Start Channel number
F8 = (2u bytes) STOPCHAN = Stop Channel number
"""
# The attributes which don't belong to this class are delegated to default: i.e Block
def __init__(self, block: BaseBlock):
self.default = TimedNonSpectral(block)
@cached
def start_mega(self):
return bin2int(self.data[12:14], False)
@cached
def start_mili(self):
return bin2int(self.data[14:18])
@cached
def stop_mega(self):
return bin2int(self.data[18:20], False)
@cached
def stop_mili(self):
return bin2int(self.data[20:24])
@cached
def start_channel(self):
return bin2int(self.data[24:26], False)
@cached
def stop_channel(self):
return bin2int(self.data[26:28], False)
class TimedVersion5(GetAttr):
"""Block with additional attributes common to Blocks from version 5 of Logger"""
def __init__(self, block: BaseBlock):
self.default = TimedNonSpectral(block)
@cached
def group_id(self) -> int:
return bin2int(self.data[BYTES_V5[3]])
@cached
def dynamic_id(self) -> int:
return bin2int(self.data[BYTES_V5[4]])
@cached
def desclen(self) -> int:
return bin2int(self.data[BYTES_V5[5]])
class DType1(GetAttr):
"""V.1 Unit Information"""
def __init__(self, block: BaseBlock):
self.default = block
@cached
def hostname(self) -> str:
return bin2str(self.data[:16])
@cached
def text_len(self) -> int:
return bin2int(self.data[16:20])
@cached
def text(self) -> str:
return bin2str(self.data[20:20 + self.text_len])
@cached
def ant_type1(self) -> int:
return bin2int(self.data[20 + self.text_len : 20 + self.text_len + 1])
@cached
def ant_type2(self) -> int:
return bin2int(self.data[20 + self.text_len + 1 : 20 + self.text_len + 2])
@cached
def ant_type3(self) -> int:
return bin2int(self.data[20 + self.text_len + 2 : 20 + self.text_len + 3])
@cached
def ant_type4(self) -> int:
return bin2int(self.data[20 + self.text_len + 3 : 20 + self.text_len + 4])
class DType2(GetAttr):
"""Data Type 40 – GPS Data"""
def __init__(self, block: BaseBlock):
self.default = block
@cached
def _gps_time(self) -> str:
"""GPS Time. Date from GPS reading (hh/mm/ss/null)"""
time = bin2time(self.data[:4])
return (
f"{str(time[0]).zfill(2)}:{str(time[1]).zfill(2)}:{str(time[2]).zfill(2)}"
)
@cached
def _gps_date(self) -> str:
"""GPS Date. Date from GPS reading (dd/mm/yy/null)"""
date = bin2date(self.data[4:8])
return (
f"{str(date[2]).zfill(2)}-{str(date[1]).zfill(2)}-{str(date[0]).zfill(2)}"
)
@cached
def gps_datetime(self) -> np.datetime64:
"""Returns the GPS datetime"""
return np.datetime64(f"{self._gps_date}T{self._gps_time}.000000")
@cached
def gps_posfix(self) -> int:
"""Positional Fix and status.
0=No Fix, 1=Standard GPS, 2=Differential GPS...n=others.
"""
return bin2int(self.data[8:10])
@cached
def gps_status(self) -> int:
"""0 = Bad, 1 = Good"""
return bin2int(self.data[10:12])
@cached
def num_satellites(self) -> int:
"""F6 - Satellites in view. 0=bad, 1+ better"""
return bin2int(self.data[12:14])
@cached
def latitude(self) -> float:
"""Latitude = Degrees * 1000000: +ve=N, -ve=S"""
return bin2int(self.data[14:18]) / 1000000
@cached
def longitude(self) -> float:
"""Longitude = Degrees * 1000000: +ve=E, -ve=W"""
return bin2int(self.data[18:22]) / 1000000
@cached
def speed(self) -> float:
"""Speed = kph * 1000"""
return bin2int(self.data[22:26]) / 1000
@cached
def heading(self) -> float:
"""F7 - Heading. Degrees * 100"""
return bin2int(self.data[26:28]) / 100
@cached
def altitude(self) -> float:
"""Altitude = Metres * 1000"""
return bin2int(self.data[28:32]) / 1000
class DType3(GetAttr):
def __init__(self, block: BaseBlock) -> None:
super().__init__()
self.default = block
@cached
def textlen(self) -> int:
return bin2int(self.data[:4])
@cached
def description(self) -> str:
return bin2str(self.data[4 : 4 + self.textlen])
class DType4(GetAttr):
def __init__(self, block: BaseBlock):
super().__init__()
self.default = block
# @cached
# def antuid(self)-> int:
# return 0 #compatibility with newer formats
@cached
def start_mega(self) -> int:
return bin2int(self.data[:2])
@cached
def stop_mega(self) -> int:
return bin2int(self.data[2:4])
@cached
def offset(self) -> int:
return bin2int(self.data[4:5])
@cached
def antuid(self) -> int:
return bin2int(self.data[5:6])
@cached
def processing(self) -> int:
d = {0: "Single Measurement", 1: "Average", 2: "Peak"}
return d.get(bin2int(self.data[6:7]), "Unknown")
@cached
def npad(self) -> int:
return bin2int(self.data[7:8])
@cached
def ndata(self) -> int:
return bin2int(self.data[8:12])
@cached
def raw_data(self) -> np.ndarray:
try:
return np.frombuffer(self.data[12 : 12 + self.ndata], dtype=np.uint8)
except ValueError:
return np.empty(self.ndata, dtype=np.uint8)
@cached
def levels(self) -> np.ndarray:
return -(self.raw_data / 2 - self.offset).astype(np.float32)
@cached
def padding(self) -> bytes:
return self.data[12 + self.ndata :]
class DType5(GetAttr):
"""Data Type 5 - Free Text Information"""
def __init__(self, block: BaseBlock):
self.default = block
@cached
def text_len(self) -> int:
return bin2int(self.data[:4])
@cached
def text(self) -> str:
return bin2str(self.data[4 : 4 + self.text_len])
class DType6(GetAttr):
"""Data Type 6 – 16 bit IQ Data.Logger versions 1 to 4"""
# The attributes which don't belong to this class are delegated to default: i.e Block
def __init__(self, block: BaseBlock):
self.default = block
@cached
def start_mega(self) -> str:
"""Start Frequency in MHz"""
return bin2int(self.data[BYTES_6[0]])
@cached
def stop_mega(self) -> int:
"""Stop Frequency in MHz"""
return bin2int(self.data[BYTES_6[1]])
@cached
def delta_freq(self) -> int:
"""Delta Frequency in MHz"""
return bin2int(self.data[BYTES_6[2]])
@cached
def antuid(self) -> int:
"""Antenna Input Number 1=ip1, 2=ip2, 3=ip3, 4=ip4"""
return bin2int(self.data[BYTES_6[3]])
@cached
def nsamples(self) -> int:
"""Number of samples per frequency"""
return bin2int(self.data[BYTES_6[4]])
@cached
def ngain(self) -> int:
"""Number of Gain Integers"""
return bin2int(self.data[BYTES_6[5]])
@cached
def gain(self) -> int:
"""Array of Gain (dB/16). Each gain is stored as a 32 bit integer"""
inicio = BYTES_6[5].stop
fim = inicio + self.ngain
return np.fromiter(self.data[inicio:fim], dtype=np.int32)
@cached
def n_iq_pairs(self) -> int:
"""Number of IQ Pairs"""
inicio = BYTES_6[5].stop + self.ngain
fim = inicio + 4
return bin2int(self.data[inicio:fim])
@cached
def iq_array(self) -> int:
"""Array of IQ data points stored as two signed twos complement 16 bit numbers packed into a 32 bit word"""
inicio = BYTES_6[5].stop + self.ngain + 4
return self.data[inicio:]
class DType7(GetAttr):
"""Data Type 7 - Threshold Compressed Data
Args:
Inherit from GetAttr to have attr access passed down to an instance attribute.
This makes it easy to create composites that don't require callers to know about their components.
For a more detailed discussion of how this works as well as relevant context,
we suggest reading the delegated composition section of this blog article.
"""
# The attributes which don't belong to this class are delegated to default: i.e Block
def __init__(self, block: BaseBlock):
self.default = block
self.minimum = self.offset - 127.5
@cached
def start_mega(self) -> int:
return bin2int(self.data[:4])
@cached
def stop_mega(self) -> int:
return bin2int(self.data[4:8])
@cached
def offset(self) -> int:
return bin2int(self.data[8:12])
@cached
def thresh(self) -> int:
return bin2int(self.data[12:16])
@cached
def antuid(self) -> int:
return bin2int(self.data[16:20])
@cached
def processing(self) -> int:
d = {0: "Single Measurement", 1: "Average", 2: "Peak"}
return d.get(bin2int(self.data[20:24]), "Unknown")
@cached
def namal(self) -> int:
return bin2int(self.data[24:28])
@cached
def ndata(self) -> int:
"""Number of original uncompressed data points"""
return bin2int(self.data[28:32])
@cached
def npad(self) -> int:
return bin2int(self.data[32:36])
@cached
def ncompressed(self) -> int:
return bin2int(self.data[36:40])
@cached
def levels(self) -> np.ndarray:
return np.frombuffer(self.data[40 : 40 + self.ncompressed], dtype=np.uint8)
@cached
def padding(self) -> bytes:
return self.data[40 + self.ndata:]
class DType8(GetAttr):
"""10.8 Data Type 8 – Frequency Channel Occupancy (ITU-R SM.1793)
Args:
GetAttr: Delegates attributes to self.default in case it doesn't belong to this class
"""
def __init__(self, block: BaseBlock):
self.default = block
@cached
def start_mega(self) -> int:
return bin2int(self.data[:4])
@cached
def stop_mega(self) -> int:
return bin2int(self.data[4:8])
@cached
def thresh(self) -> int:
return bin2int(self.data[8:12])
@cached
def _walldate(self) -> list:
date = bin2date(self.data[12:16])
return (
f"{str(date[2]).zfill(4)}-{str(date[1]).zfill(2)}-{str(date[0]).zfill(2)}"
)
@cached
def _walltime(self) -> list:
time = bin2time(self.data[16:20])
return (
f"{str(time[0]).zfill(2)}:{str(time[1]).zfill(2)}:{str(time[2]).zfill(2)}"
)
@cached
def wallclock_datetime(self) -> np.datetime64:
"""Returns the wallclock datetime"""
return np.datetime64(f"{self._walldate}T{self._walltime}.000000")
@cached
def sampling(self) -> int:
return bin2int(self.data[20:24])
@cached
def antuid(self) -> int:
return bin2int(self.data[24:28])
@cached
def namal(self) -> int:
return bin2int(self.data[28:32])
@cached
def npad(self) -> int:
return bin2int(self.data[32:36])
@cached
def ndata(self) -> int:
return bin2int(self.data[36:40])
@cached
def raw_data(self) -> np.ndarray:
try:
return np.frombuffer(self.data[40 : 40 + self.ndata], dtype=np.uint8)
except ValueError as e:
raise ValueError(f"{e} while retrieving raw_data in {self.filename}") from e
@cached
def levels(self) -> np.ndarray:
return (self.raw_data / 2).astype(np.float32)
@cached
def padding(self) -> bytes:
return self.data[40 + self.ndata :]
Data Type 20 - File Footer - Optional
An optional data block at the end of each file provides a summary of some aspects of the file content. The MBR is the Minimum Bounding Rectangle of all GPS records that have a fix. The start and end date/time is included, omitting the nanoseconds fields. If a file is used for the output from more than one unit, multiple footers may appear, up to one per unit.
class DType20(GetAttr):
"""Data Type 20 – File Footer - Optional"""
# The attributes which don't belong to this class are delegated to default: i.e Block
def __init__(self, block: BaseBlock):
self.default = block
@cached
def bottom_latitude(self) -> float:
return bin2int(self.data[BYTES_20[0]]) / 1e6
@cached
def left_longitude(self) -> float:
return bin2int(self.data[BYTES_20[1]]) / 1e6
@cached
def top_latitude(self) -> float:
return bin2int(self.data[BYTES_20[2]]) / 1e6
@cached
def right_longitude(self) -> float:
return bin2int(self.data[BYTES_20[3]]) / 1e6
@cached
def start_date(self) -> L:
"""Date of first data block (dd/mm/yy/null)"""
return bin2date(self.data[BYTES_20[4]])
@cached
def start_time(self) -> L:
"""Time of first data block (hh/mm/ss/cc)"""
return bin2time(self.data[BYTES_20[5]])
@cached
def end_date(self) -> L:
"""Date of last data block (dd/mm/yy/null)"""
return bin2date(self.data[BYTES_20[6]])
@cached
def end_time(self) -> L:
"""Time of last data block (hh/mm/ss/cc)"""
return bin2time(self.data[BYTES_20[7]])
@cached
def n_spectral_blocks(self) -> int:
"""Total number of blocks in threads with threadID>0"""
return bin2int(self.data[BYTES_20[8]])
@cached
def nddt(self) -> int:
"""Spectral data, i.e. ThreadID > 0.
Optionally can be set to a special value of zero, to omit the following information."""
return bin2int(self.data[BYTES_20[9]])
@cached
def spectral_blocks_info(self) -> L:
start = BYTES_20[9].stop
stop = start + self.nddt
step = 8
return L.range(start, stop, step).map(
lambda i: (bin2int(self.data[i : i + 4]), bin2int(self.data[i + 4 : i + 8]))
)
class DType21(GetAttr):
"""Data Type 21 – Unit and Job Information.Logger versions 3 to 5"""
# The attributes which don't belong to this class are delegated to default: i.e Block
def __init__(self, block: BaseBlock):
self.default = block
@cached
def hostname(self) -> str:
"""Retorna o campo HOSTNAME que contém o 'Unit Hostname'"""
return bin2str(self.data[:16])
@cached
def _text1_len(self) -> int:
"""Retorna o tamanho do campo TEXT1 que contém o ‘unit_info’ no arquivo cfg."""
return bin2int(self.data[16:20])
@cached
def unit_info(self) -> str:
"""Retorna o campo TEXT1 que contém o ‘unit_info’ no arquivo cfg."""
return bin2str(self.data[20 : 20 + self._text1_len])
@cached
def _text2_len(self) -> int:
"""Retorna o tamanho do campo TEXT2 que contém o ‘unit_info’ no arquivo cfg."""
return bin2int(self.data[20 + self._text1_len : 24 + self._text1_len])
@cached
def method(self) -> str:
"""Retorna o campo TEXT2 que contém o ‘method’ no arquivo cfg."""
start = 24 + self._text1_len
return bin2str(self.data[start : start + self._text2_len])
@cached
def file_number(self) -> int:
"""Return the File Number"""
start = 24 + self._text1_len + self._text2_len
stop = start + 4
return bin2int(self.data[start:stop])
class DType22(GetAttr):
"""Data Type 22 – Data Thread Information"""
def __init__(self, block: BaseBlock) -> None:
super().__init__()
self.default = block
@cached
def textlen(self) -> int:
return bin2int(self.data[:4])
@cached
def description(self) -> str:
return bin2str(self.data[4 : 4 + self.textlen])
class DType23(GetAttr):
"""Data Type 23 - Free Text Information"""
def __init__(self, block: BaseBlock):
super().__init__()
self.default = block
@cached
def textid(self) -> int:
return bin2str(self.data[:32])
@cached
def textlen(self) -> int:
return bin2int(self.data[32:36])
@cached
def text(self) -> str:
return bin2str(self.data[36 : 36 + self.textlen])
class DType24(GetAttr):
"""Data Type 24 – Data Thread Information"""
# The attributes which don't belong to this class are delegated to default: i.e Block
def __init__(self, block: BaseBlock):
self.default = block
@cached
def group_id(self) -> int:
"""F0 - Número identificador do grupo. Zero significa que não faz parte de nenhum grupo"""
return bin2int(self.data[BYTES_24[0]])
@cached
def text_len(self) -> int:
"""F1 - Comprimento do texto incluindo bytes nulos ao final (número inteiro de 4 bytes)."""
return bin2int(self.data[BYTES_24[1]])
@cached
def description(self) -> str:
"""F2 - Texto armazenado no bloco sem os bytes nulos ao final"""
start = BYTES_24[1].stop
stop = start + self.text_len
return bin2str(self.data[start:stop])
class DType40(GetAttr):
"""Data Type 40 – GPS Data"""
def __init__(self, block: BaseBlock):
"""This implementation substitues inheritance of the class Block by Composition
The attributes which belong to Block are accessed normally as if it was Inherited
"""
self.default = TimedNonSpectral(block)
@cached
def _gpsdate(self) -> str:
"""F3 = (4 bytes) _walldate = Wall Clock Start Date of measurements"""
date = bin2date(self.data[BYTES_40[3]])
return (
f"{str(date[2]).zfill(4)}-{str(date[1]).zfill(2)}-{str(date[0]).zfill(2)}"
)
@cached
def _gpstime(self) -> str:
"""F4 - GPS Date. Date from GPS reading (dd/mm/yy/null)"""
time = bin2time(self.data[BYTES_40[4]])
return (
f"{str(time[0]).zfill(2)}:{str(time[1]).zfill(2)}:{str(time[2]).zfill(2)}"
)
@cached
def gps_datetime(self) -> np.datetime64:
"""Returns the wallclock datetime"""
return np.datetime64(f"{self._gpsdate}T{self._gpstime}")
@cached
def gps_status(self) -> Union[str, int]:
"""F5 - Positional Fix and status.
If status=1: 0=No Fix, 1=Standard GPS, 2=Differential GPS. If status=0: set to zero.
"""
return bin2int(self.data[BYTES_40[5]])
@cached
def num_satellites(self) -> int:
"""F6 - Satellites in view. 0=bad, 1+ better"""
return bin2int(self.data[BYTES_40[6]])
@cached
def heading(self) -> float:
"""F7 - Heading. Degrees * 100"""
return bin2int(self.data[BYTES_40[7]]) / 100
@cached
def latitude(self) -> float:
"""F8 - Latitude = Degrees * 1000000: +ve=N, -ve=S"""
return bin2int(self.data[BYTES_40[8]]) / 1000000
@cached
def longitude(self) -> float:
"""F9 - (4 bytes) Longitude = Degrees * 1000000: +ve=E, -ve=W"""
return bin2int(self.data[BYTES_40[9]]) / 1000000
@cached
def speed(self) -> float:
"""F10 - (4 bytes) Speed = kph * 1000"""
return bin2int(self.data[BYTES_40[10]]) / 1000
@cached
def altitude(self) -> float:
"""F11 = (4 bytes) Altitude = Metres * 1000"""
return bin2int(self.data[BYTES_40[11]]) / 1000
class DType41(GetAttr):
"""Data Type 41 – Timed Free Text Information"""
def __init__(self, block: BaseBlock):
"""This implementation substitues inheritance of the class Block by Composition
The attributes which belong to Block are accessed normally as if it was Inherited
"""
self.default = TimedNonSpectral(block)
@cached
def identifier(self) -> str:
"""F3 = (32 bytes) Fixed Length Data Type Identifier, null terminated string
Tells the software how to deal with the free text.
Defined types:
- LOGGER_NAME: Application name
- LOGGER_VERSION: Application version
- AUDIT: Audit process output
- GPRS: GPRS message
- GSM: Cell survey information
- INFO: Voltages, currents and temperatures
- LED: LED status
- MASK: Mask status
- MESSAGE: Message text
- NMEA: GSM NMEA text
- SNMP: SNMP Message text
- CONF: NCPD Configuration
"""
return bin2str(self.data[BYTES_41[3]])
@cached
def textlen(self) -> int:
"""F4 - (4 bytes) NTEXT = Free Text Length. Including null termination and padding
(must be a whole number of 4 bytes)
"""
return bin2int(self.data[BYTES_41[4]])
@cached
def text(self) -> str:
"""F5 = (NTEXT bytes) Null terminated Free Text"""
return bin2str(self.data[BYTES_41[4].stop :])
class DType42(GetAttr):
"""Data Type 42 – Timed Free Text Information"""
def __init__(self, block: BaseBlock):
"""This implementation substitues inheritance of the class Block by Composition
The attributes which belong to Block are accessed normally as if it was Inherited
"""
self.default = TimedNonSpectral(block)
@cached
def group_id(self) -> int:
return bin2int(self.data[BYTES_42[3]])
@cached
def dynamic_id(self) -> int:
return bin2int(self.data[BYTES_42[4]])
@cached
def identifier(self) -> str:
"""F5 = (32 bytes) Fixed Length Data Type Identifier, null terminated string
Tells the software how to deal with the free text.
Defined types:
- LOGGER_NAME: Application name
- LOGGER_VERSION: Application version
- AUDIT: Audit process output
- GPRS: GPRS message
- GSM: Cell survey information
- INFO: Voltages, currents and temperatures
- LED: LED status
- MASK: Mask status
- MESSAGE: Message text
- NMEA: GSM NMEA text
- SNMP: SNMP Message text
- CONF: NCPD Configuration
"""
return bin2str(self.data[BYTES_42[5]])
@cached
def len_text(self) -> int:
"""F6 - (4 bytes) NTEXT = Free Text Length. Including null termination and padding
(must be a whole number of 4 bytes)
"""
return bin2int(self.data[BYTES_42[6]])
@cached
def text(self) -> str:
"""F5 = (NTEXT bytes) Null terminated Free Text"""
return bin2str(self.data[BYTES_42[6].stop :])
class DType51(GetAttr):
"""Classifier Data"""
def __init__(self, block: BaseBlock) -> None:
self.default = DType42(block)
@cached
def desclen(self):
return bin2int(self.data[BYTES_51[5]])
@cached
def description(self):
start = BYTES_51[5].stop
stop = start + self.desclen
return bin2str(self.data[start:stop])
@cached
def center_mega(self):
start = BYTES_51[5].stop + self.desclen
stop = start + 2
return bin2int(self.data[start:stop], False)
@cached
def center_mili(self):
start = BYTES_51[5].stop + self.desclen + 2
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def bw(self):
start = BYTES_51[5].stop + self.desclen + 6
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def sample_rate_numerator(self):
start = BYTES_51[5].stop + self.desclen + 10
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def sample_rate_denominator(self):
start = BYTES_51[5].stop + self.desclen + 14
stop = start + 2
return bin2int(self.data[start:stop])
@cached
def len_text(self) -> int:
start = BYTES_51[5].stop + self.desclen + 16
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def text(self) -> str:
start = BYTES_51[5].stop + self.desclen + 20
stop = start + self.len_text
return bin2str(self.data[start:stop])
# @cached_property
# def align1(self):
# start = BYTES_51[5].stop + self.desclen + 20 + self.len_text
# stop = start + 2
# return bin2dec(self.data[start:stop])
# @cached_property
# def align2(self):
# start = BYTES_51[5].stop + self.desclen + 22 + self.len_text
# stop = start + 1
# return bin2dec(self.data[start:stop])
@cached
def npad(self):
start = BYTES_51[5].stop + self.desclen + 23 + self.len_text
stop = start + 1
return bin2int(self.data[start:stop])
@cached
def npoints(self):
start = BYTES_51[5].stop + self.desclen + 24 + self.len_text
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def probabilites(self):
start = BYTES_51[5].stop + self.desclen + 28 + self.len_text
stop = start + self.npoints
return bin2int(self.data[start:stop])
@cached
def padding(self):
start = BYTES_51[5].stop + self.desclen + 28 + self.len_text + self.npoints
return bin2int(self.data[start:])
class DType60(GetAttr):
"""Data Type 60 – 8 bit Spectral Data"""
def __init__(self, block: BaseBlock) -> None:
self.default = block
@cached
def _walldate(self) -> L:
"F1 = (4 bytes) Wall clock date and time in seconds since 1/1/1970"
date = bin2date(self.data[:4])
return (
f"{str(date[2]).zfill(4)}-{str(date[1]).zfill(2)}-{str(date[0]).zfill(2)}"
)
@cached
def _walltime(self) -> str:
"""F1 = (4 bytes) WALLTIME = Wall Clock Start Time"""
time = bin2time(self.data[4:8])
return (
f"{str(time[0]).zfill(2)}:{str(time[1]).zfill(2)}:{str(time[2]).zfill(2)}"
)
@cached
def _wallnano(self) -> int:
return bin2int(self.data[8:12], False)
@cached
def wallclock_datetime(self) -> np.datetime64:
"""Returns the wallclock datetime"""
return np.datetime64(
f"{self._walldate}T{self._walltime}.{int(self._wallnano/1000)}"
)
@cached
def dtype(self): # To maintain compatibility with Other Blocks
return 'dBm'
@cached
def start_mega(self) -> int:
return bin2int(self.data[12:16])
@cached
def stop_mega(self) -> int:
return bin2int(self.data[16:20])
@cached
def freqexp(self) -> int:
return bin2int(self.data[20:21])
@cached
def antuid(self) -> int:
return bin2int(self.data[21:22])
@cached
def gerror(self) -> int:
return bin2int(self.data[22:23])
@cached
def gflags(self) -> int:
return self.data[23:24]
@cached
def parentid(self) -> int:
return bin2int(self.data[24:26])
@cached
def processing(self) -> int:
d = {0: "single measurement", 1: "average", 2: "peak", 3: "minimum"}
return d.get(bin2int(self.data[26:27]), "unknown")
@cached
def nloops(self) -> int:
"i.e. ‘number of loops’. Equal to 1 if single measurement. Values 1, 2, 4, 8, 16"
return bin2int(self.data[27:28])
@cached
def offset(self) -> int:
"Data level offset in dB 2’s Complement, range [-128, 127]."
return bin2int(self.data[28:29])
@cached
def minimum(self) -> float:
return self.offset - 127.5
@cached
def npad(self) -> int:
"Number of bytes of padding. 0 - 3"
return bin2int(self.data[29:30])
@cached
def ntun(self) -> int:
"Number of 4 byte Tuning info blocks. 0 or NAGC"
return bin2int(self.data[30:32])
@cached
def nagc(self) -> int:
"Number of single byte AGC values. 0 or number of tunings in the sweep"
return bin2int(self.data[32:34])
@cached
def reserved(self) -> int:
"For alignment of following fields. Set to 0."
return bin2int(self.data[34:36])
@cached
def ndata(self) -> int:
"Number of single byte data points in this block."
return bin2int(self.data[36:40])
@cached
def tunning(self) -> list:
"Array of 4 byte Tuning info blocks.One block per tuning (1 or 10 MHz)."
return L(self.data[i : i + 4] for i in range(40, 40 + 4 * self.ntun, 4))
@cached
def agc(self) -> list:
"Array of single unsigned byte AGC values. Current actual values are 0..63."
return np.frombuffer(
self.data[40 + 4 * self.ntun : 40 + 4 * self.ntun + self.nagc],
dtype=np.uint8,
)
@cached
def raw_data(self) -> np.ndarray:
"Array of single byte data points."
start = 40 + 4 * self.ntun + self.nagc
try:
return np.frombuffer(self.data[start : start + self.ndata], dtype=np.uint8)
except ValueError as e:
raise ValueError(f"{e} while retrieving raw_data in {self.filename}") from e
@cached
def levels(self) -> list:
"Array of single byte data points."
return (self.raw_data / 2 + self.minimum).astype(np.float32)
@cached
def padding(self) -> bytes:
"Padding to align to 4 byte boundary."
return self.data[40 + 4 * self.ntun + self.nagc + self.ndata :]
class DType61(GetAttr):
"""Data Type 61 – Threshold Compressed Data"""
def __init__(self, block: BaseBlock) -> None:
self.default = DType60(block)
self.size = (
48
+ (self.default.ntun * 4)
+ self.default.nagc
+ self.default.ndata
+ self.default.npad
)
@cached
def thresh(self) -> int:
"Threshold Level in dBm.All data below this level will be run length encoded."
return bin2int(self.data[40:44])
@cached
def ndata(self) -> int:
"Number of original data points in this block before compression."
return bin2int(self.data[44:48])
@cached
def tunning(self) -> list:
"Array of 4 byte Tuning info blocks.One block per tuning (1 or 10 MHz)."
return L(self.data[i : i + 4] for i in range(48, 48 + 4 * self.ntun, 4))
@cached
def agc(self) -> np.ndarray:
return np.frombuffer(
self.data[48 + 4 * self.ntun : 48 + 4 * self.ntun + self.nagc],
dtype=np.uint8,
)
@cached
def levels(self) -> bytes:
start = 48 + 4 * self.ntun + self.nagc
return np.frombuffer(self.data[start : start + self.ndata], dtype=np.uint8)
@cached
def padding(self) -> bytes:
"Padding to align to 4 byte boundary."
return self.data[48 + 4 * self.ntun + self.nagc + self.ndata:]
class DType62(GetAttr):
def __init__(self, block: BaseBlock) -> None:
self.default = block
@cached
def _walldate(self) -> L:
"F1 = (4 bytes) Wall clock date and time in seconds since 1/1/1970"
date = bin2date(self.data[:4])
return (
f"{str(date[2]).zfill(4)}-{str(date[1]).zfill(2)}-{str(date[0]).zfill(2)}"
)
@cached
def _walltime(self) -> str:
"""F1 = (4 bytes) WALLTIME = Wall Clock Start Time"""
time = bin2time(self.data[4:8])
return (
f"{str(time[0]).zfill(2)}:{str(time[1]).zfill(2)}:{str(time[2]).zfill(2)}"
)
@cached
def _wallnano(self) -> int:
return bin2int(self.data[8:12], False)
@cached
def wallclock_datetime(self) -> np.datetime64:
"""Returns the wallclock datetime"""
return np.datetime64(
f"{self._walldate}T{self._walltime}.{int(self._wallnano/1000)}"
)
@cached
def start_mega(self) -> int:
return bin2int(self.data[12:16])
@cached
def stop_mega(self) -> int:
return bin2int(self.data[16:20])
@cached
def freqexp(self) -> int:
return bin2int(self.data[20:21])
@cached
def antuid(self) -> int:
return bin2int(self.data[21:22])
@cached
def gerror(self) -> int:
return bin2int(self.data[22:23])
@cached
def gflags(self) -> int:
return self.data[23:24]
@cached
def parentid(self) -> int:
return bin2int(self.data[24:26])
@cached
def thresh(self) -> int:
"""Threshold Level in dBm"""
return bin2int(self.data[26:28])
@cached
def namal(self) -> int:
"""Amalgamated Results. 1 if single measurement. Should include aggregation over time."""
return bin2int(self.data[28:32])
@cached
def sampling(self) -> int:
"""Duration of sampling in seconds.i.e.300,900,1800,2600"""
return bin2int(self.data[32:36])
@cached
def npad(self) -> int:
"""Number of bytes of padding. 0 - 3"""
return bin2int(self.data[36:37])
@cached
def reserved(self) -> int:
"""For alignment of following fields. Set to 0."""
return bin2int(self.data[37:40])
@cached
def ndata(self) -> int:
"""Number of single byte data points in this block.
Number of equal width channels dividing the reported frequency width"""
return bin2int(self.data[40:44])
@cached
def levels(self) -> np.ndarray:
"""Array of single byte data points representing the percentage. (0...100% in 0.5 steps)"""
try:
return (np.frombuffer(self.data[44 : 44 + self.ndata], dtype=np.uint8) / 2).astype(np.float32)
except ValueError as e:
raise ValueError(f"{e} while retrieving levels in {self.type}") from e
@cached
def padding(self) -> bytes:
"""Padding to align to 4 byte boundary."""
return self.data[44 + self.ndata :]
# TODO: review agc and tunning
class DType63(GetAttr):
"""Data Type 63 – Spectral Data - Frequencies and its levels
The Following attributes are delegated to the `self.default` object
F0 = (4 bytes) _walldate = Wall Clock Start Date of measurements
F1 = (4 bytes) WALLTIME = Wall Clock Start Time
F2 = (4u bytes) WALLNANO = Wall Clock Start Time Nanoseconds
F3 = (2u bytes) STARTMEGA = Start Frequency MHz
F4 = (4 bytes) STARTMILLI = Start Frequency mHz
F5 = (2u bytes) STOPMEGA = Stop Frequency MHz
F6 = (4 bytes) STOPMILLI = Stop Frequency mHz
F7 = (2u bytes) STARTCHAN = Start Channel number
F8 = (2u bytes) STOPCHAN = Stop Channel number
Total Bytes = 52 + (4 * NTUN) + NAGC + NDATA + NPAD
"""
def __init__(self, block: BaseBlock) -> None:
self.default = TimedSpectral(block)
self.start = 52 + self.ntun * 4 + self.nagc
self.stop = self.start + self.ndata
@cached
def description(self) -> str:
return "Peak"
@cached
def sample(self) -> int:
"""F9 = (4 bytes) SAMPLE = Duration of sampling.Time taken by the FPGA and Radio to execute command in µs."""
return bin2int(self.data[28:32])
@cached
def namal(self) -> int:
"""F10 =(4 bytes) NAMAL = Amalgamated Results"""
return bin2int(self.data[32:36])
@cached
def antuid(self) -> int:
"""F11 = (1u byte) ANTUID Antenna number [0-255]"""
return bin2int(self.data[36:37], False)
@cached
def processing(self) -> Union[str, int]:
"""F12 = (1 byte) PROC = Processing
0 = single measurement,
1 = average,
2 = peak,
3 = minimum
"""
proc = bin2int(self.data[37:38])
return DICT_PROCESSING.get(proc, proc)
@cached
def dtype(self) -> Union[str, int]:
"""F13 = (1 byte) DTYPE = Data Type. 0 = dBm, 1 = dBuV/m"""
unit: int = bin2int(self.data[38:39])
return DICT_UNIT.get(unit, unit)
@cached
def offset(self) -> int:
"""F14 = (1 byte) OFFSET = Data level offset in DTYPE units 2’s Complement, range [-128, 127]."""
return bin2int(self.data[39:40])
@cached
def minimum(self) -> float:
return self.offset - 127.5
@cached
def thresh(self) -> float:
return self.minimum # redundancy for compatibility with Dtype68
@cached
def gerror(self) -> int:
"""F15 = (1 byte) GERROR = Global Error Code. Radio or processing global error code"""
return bin2int(self.data[40:41])
@cached
def gflags(self) -> int:
"""F16 - Códigos de alertas globais ou de processamento do radio."""
return bin2int(self.data[41:42])
@cached
def group_id(self) -> int:
"""F17 - O ID do grupo à qual a medida pertence.0 caso não pertença a nenhum grupo."""
return bin2int(self.data[42:43])
@cached
def ntun(self) -> int:
"""F18 - 0 ou igual à quantidade de valores de AGC usados na amostra"""
return bin2int(self.data[43:45])
@cached
def nagc(self) -> int:
"""F19 - 0 ou igual à quantidade de valores de "tunings" usados na amostra"""
return bin2int(self.data[45:47])
@cached
def npad(self) -> int:
"""F20 - Valor que varia de 0 a 3 indicando o preenchimento nulo para manter o tamanho do bloco (em bytes) fixo"""
return bin2int(self.data[47:48])
@cached
def ndata(self) -> int:
"""F21 - O número de canais (ou "steps") que dividem igualmente a largura de banda"""
return bin2int(self.data[48:52])
@cached
def frequencies(self) -> np.ndarray:
"""Retorna um numpy array com a faixa de frequências presentes no bloco"""
return np.linspace(
self.start_mega, self.stop_mega, num=self.ndata, dtype=np.float32
)
# @cached
# def bw(self) -> int:
# """Retorna o RBW calculado a partir de STARTMEGA, STOPMEGA e NDATA."""
# return int((self.stop_mega - self.start_mega) * 1000 / (self.ndata - 1))
@cached
def tunning(self) -> Tuple:
"""F22 - Informações do 'tunning'. One Block per tunning(1 or 10MHz)"""
start = 52
stop = start + (4 * self.ntun)
return L(self.data[i : i + 4] for i in range(start, stop, 4))
@cached
def agc(self) -> np.ndarray:
"""F23 - Array com AGC - Automatic Gain Control as dB in single unsigned byte: 0...63"""
start = 52 + (self.ntun * 4)
stop = start + self.nagc
return np.frombuffer(self.data[start:stop], np.uint8)
@cached
def raw_data(self) -> np.ndarray:
"""Spectrum Data in 'dB' with 0.5 dBm interval"""
try:
return np.frombuffer(self.data[self.start : self.stop], dtype=np.uint8)
except ValueError as e:
raise ValueError(f"{e} while retrieving raw_data in {self}") from e
@cached
def levels(self) -> np.ndarray:
return ((self.raw_data / 2) + self.minimum).astype(np.float32)
def __getitem__(self, i):
"""Return a tuple with frequency, spectrum_data"""
return self.frequencies[i], self.block_data[i]
def __len__(self):
return int(self.ndata)
def __iter__(self):
return iter(self[i] for i in range(len(self)))
class DType64(GetAttr):
"""This is very similar to type 63, with the addition of threshold power level and number of points before compression,
and obviously a compressed spectrum vector replacing the uncompressed
Fields 0 to 21 are the same from `DType63` and are delegated to this class
"""
def __init__(self, block: BaseBlock) -> None:
self.default = DType63(block)
self.minimum = self.offset - 127.5
self.start = 52 + 8 + (self.ntun * 4) + self.nagc
self.stop = self.start + self.default.ndata
@cached
def thresh(self):
"""F22 - THRESH = Threshold Level in dB All data below this level will be run length encoded"""
start = 52
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def ndata(self):
"""F23 - NORIG = Number of original data points4 Before compression"""
start = 52 + 4
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def tunning(self) -> Tuple:
"""F24 - Informações do 'tunning'. One Block per tunning(1 or 10MHz)"""
start = 52 + 8
stop = start + (4 * self.ntun)
return L(self.data[i : i + 4] for i in range(start, stop, 4))
@cached
def agc(self) -> bytes:
"""F23 - Array com AGC - Automatic Gain Control as dB in single unsigned byte: 0...63"""
start = 52 + 8 + (self.ntun * 4)
stop = start + (self.nagc)
return np.fromiter(self.data[start:stop], np.uint8)
# return '-'.join(L(self.data[start:stop]).map(str))
@cached
def levels(self) -> np.ndarray:
"""Spectrum Data in 'dB' with 0.5 dBm interval"""
return np.frombuffer(self.data[self.start : self.stop], dtype=np.uint8)
@cached
def padding(self):
return bin2int(self.data[self.stop :])
def __getitem__(self, i):
"""Return a tuple with frequency, spectrum_data"""
return self.frequencies[i], self.levels[i]
def __len__(self):
return len(self.raw_data)
def __iter__(self):
return iter(self[i] for i in range(len(self)))
class DType65(GetAttr):
"""
O Bloco do tipo 65 carrega dados de Taxa de Ocupação Espectral por Canal de Frequência (ITU-R SM.1880).
O tamanho total em bytes é (48 + NDATA + NPAD)
F0 a F16 e F19 já inicializados na classe TimedBlock.
*F0 = (4 bytes) _walldate = Wall Clock Start Date of measurements
*F1 = (4 bytes) WALLTIME = Wall Clock Start Time
*F2 = (4u bytes) WALLNANO = Wall Clock Start Time Nanoseconds
*F3 = (2u bytes) STARTMEGA = Start Frequency MHz
*F4 = (4 bytes) STARTMILLI = Start Frequency mHz
*F5 = (2u bytes) STOPMEGA = Stop Frequency MHz
*F6 = (4 bytes) STOPMILLI = Stop Frequency mHz
*F7 = (2u bytes) STARTCHAN = Start Channel number
*F8 = (2u bytes) STOPCHAN = Stop Channel number
F9 = (4 bytes) NAMAL = Amalgamated Results
i.e. ‘number of loops’. Equal to 1 if single measurement.
F10 = (1u bytes) ANTUID Antenna number [ 0- 255]
F11 = (1 byte) PROC = Processing
0 = single measurement,
1 = average,
2 = peak,
3 = minimum
F12 = (2 bytes) DTYPE = Data Type
1 = dBm,
2 = dBuV/m
F13 = (1 byte) GERROR = Global Error Code. Radio or processing global error code
F14 = (1 byte) GFLAGS = Global clipping flags etc. Radio or processing global flags:
F15 = (1 bytes) GROUPID = ID used to group sets of data
0 = not a member of a group
F16 = (1 byte) NPAD = Number of bytes of padding. 0-3
F17 = (2 bytes) THRESH = Threshold Level in DTYPE
F18 = (2 bytes) DURATION = Duration of sampling. In seconds (i.e. 300, 900, 1800, 2600)
F19 = (4 bytes) NDATA = Number of single byte data points.
Number of equal width channels dividing the reported frequency width
F20 = (NDATAu bytes) Array of data points.
Each data point is stored as a single byte number
representing the percentage (0..100 % in 0.5 steps)
F21 = (NPAD bytes) Padding. As \0 bytes
"""
def __init__(self, block: BaseBlock) -> None:
self.default = TimedSpectral(block)
self.start = BYTES_65[19].stop
self.stop = self.start + self.ndata
@cached
def namal(self) -> int:
"""F9 = (4 bytes) NAMAL = Amalgamated Results
i.e. ‘number of loops’. Equal to 1 if single measurement.
"""
return bin2int(self.data[BYTES_65[9]])
@cached
def antuid(self) -> int:
"""F10 = (1u bytes) ANTUID Antenna number [ 0- 255]"""
return bin2int(self.data[BYTES_65[10]], False)
@cached
def processing(self) -> Union[str, int]:
"""F11 = (1 byte) PROC = Processing, 0 = single measurement, 1 = average, 2 = peak, 3 = minimum"""
proc = bin2int(self.data[BYTES_65[11]])
return DICT_PROCESSING.get(proc, proc)
@cached
def dtype(self) -> Union[str, int]:
"""F12 = (1 byte) DTYPE = Data Type. 0 = % (undocumented), 1 = dBm, 2 = dBuV/m"""
unit = bin2int(self.data[BYTES_65[12]])
return DICT_UNIT.get(unit, unit)
@cached
def gerror(self) -> int:
"""F13 = (1 byte) GERROR = Global Error Code.Radio or processing global error code"""
return bin2int(self.data[BYTES_65[13]])
@cached
def gflags(self) -> int:
"""F14 = (1 byte) GFLAGS = Global clipping flags etc.Radio or processing global flags."""
return bin2int(self.data[BYTES_65[14]])
@cached
def group_id(self) -> int:
"""F15 = (1 bytes) GROUPID = ID used to group sets of data. 0 = not a member of a group"""
return bin2int(self.data[BYTES_65[15]])
@cached
def npad(self) -> int:
"""F16 = (1 byte) NPAD = Number of bytes of padding. 0-3 (NPAD bytes) Padding = As \0 bytes"""
return bin2int(self.data[BYTES_65[16]])
@cached
def thresh(self) -> int:
"""F17 = (2 bytes) THRESH = Threshold Level in DTYPE"""
return bin2int(self.data[BYTES_65[17]])
@cached
def duration(self) -> int:
"""F18 = (2 bytes) DURATION = Duration of sampling. In seconds (i.e. 300, 900, 1800, 2600)"""
return bin2int(self.data[BYTES_65[18]])
@cached
def ndata(self) -> int:
"""F19 = (4 bytes) NDATA = Number of single byte data points.
Number of equal width channels dividing the reported frequency width
"""
return bin2int(self.data[BYTES_65[19]])
@cached
def levels(self) -> np.ndarray:
"""
self > -> List[float]
:return: A lista de todas as medidas de ocupação do bloco em '%'.
F20 = (NDATAu bytes) Array of data points.
Each data point is stored as a single byte number
representing the percentage (0..100 % in 0.5 steps)
"""
try:
return (
np.frombuffer(self.data[self.start : self.stop], dtype=np.uint8) / 2.0
).astype(np.float32)
except ValueError as e:
raise ValueError(f"{e} while retrieving levels in {self}") from e
class DType66(GetAttr):
"""Data Type 66 – 16 bit IQ Data"""
def __init__(self, block: BaseBlock):
"""This implementation substitues inheritance of the class Block by Composition
The attributes which belong to Block are accessed normally as if it was Inherited
"""
self.default = TimedVersion5(block)
@cached
def description(self) -> str:
start = BYTES_V5[5].stop
stop = start + self.desclen
return bin2str(self.data[start:stop])
@cached
def center_mega(self) -> int:
start = BYTES_V5[5].stop + self.desclen
stop = start + 2
return bin2int(self.data[start:stop], False) + self.center_mili / 1000
@cached
def center_mili(self) -> int:
start = BYTES_V5[5].stop + self.desclen + 2
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def bw(self):
start = BYTES_V5[5].stop + self.desclen + 6
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def sample_rate_numerator(self):
start = BYTES_V5[5].stop + self.desclen + 10
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def sample_rate_denominator(self):
start = BYTES_V5[5].stop + self.desclen + 14
stop = start + 2
return bin2int(self.data[start:stop])
@cached
def input(self) -> int:
start = BYTES_V5[5].stop + self.desclen + 16
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def ngain(self) -> int:
start = BYTES_V5[5].stop + self.desclen + 20
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def gain(self) -> np.ndarray:
start = BYTES_V5[5].stop + self.desclen + 24
stop = start + self.ngain
try:
return np.frombuffer(self.data[start:stop], dtype=np.int32)
except ValueError as e:
raise ValueError(f"{e} while retrieving gain in {self}") from e
@cached
def n_iq_pairs(self) -> int:
start = BYTES_V5[5].stop + self.desclen + 24 + self.ngain
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def iq_array(self) -> np.ndarray:
start = BYTES_V5[5].stop + self.desclen + 28 + self.ngain
stop = start + self.n_iq_pairs
try:
return np.frombuffer(self.data[start:stop], dtype=np.int32)
except ValueError as e:
raise ValueError(f"{e} while retrieving iq_array in {self}") from e
class DType67(GetAttr):
"""Data Type 67 – Spectral Data"""
def __init__(self, block: BaseBlock):
"""This implementation substitues inheritance of the class Block by Composition
The attributes which belong to Block are accessed normally as if it was Inherited
"""
self.default = TimedVersion5(block)
self.start = 68 + self.desclen + 4 * self.ntun + self.nagc
self.stop = self.start + self.ndata
@cached
def description(self) -> str:
start = 24
stop = start + self.desclen
return bin2str(self.data[start:stop])
@cached
def start_mega(self) -> int:
start = 24 + self.desclen
stop = start + 2
return bin2int(self.data[start:stop], False)
@cached
def start_mili(self) -> int:
start = 26 + self.desclen
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def stop_mega(self) -> int:
start = 30 + self.desclen
stop = start + 2
return bin2int(self.data[start:stop], False)
@cached
def stop_mili(self) -> int:
start = 32 + self.desclen
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def bw(self) -> int:
start = 36 + self.desclen
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def start_channel(self) -> int:
start = 40 + self.desclen
stop = start + 2
return bin2int(self.data[start:stop])
@cached
def stop_channel(self) -> int:
start = 42 + self.desclen
stop = start + 2
return bin2int(self.data[start:stop])
@cached
def sample(self) -> int:
start = 44 + self.desclen
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def namal(self) -> int:
start = 48 + self.desclen
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def antuid(self) -> int:
start = 52 + self.desclen
stop = start + 1
return bin2int(self.data[start:stop])
@cached
def processing(self) -> Union[str, int]:
start = 53 + self.desclen
stop = start + 1
proc = bin2int(self.data[start:stop])
return DICT_PROCESSING.get(proc, proc)
@cached
def dtype(self) -> Union[str, int]:
start = 54 + self.desclen
stop = start + 1
unit = bin2int(self.data[start:stop])
return DICT_UNIT.get(unit, unit)
@cached
def offset(self) -> int:
start = 55 + self.desclen
stop = start + 1
return bin2int(self.data[start:stop])
@cached
def minimum(self) -> float:
return self.offset - 127.5
@cached
def thresh(self) -> float:
return self.minimum # redundancy for compatibility with Dtype68
@cached
def gerror(self) -> int:
start = 56 + self.desclen
stop = start + 1
return bin2int(self.data[start:stop])
@cached
def gflags(self) -> int:
start = 57 + self.desclen
stop = start + 1
return bin2int(self.data[start:stop])
@cached
def ntun(self) -> int:
start = 59 + self.desclen # byte 58 reserved alignment
stop = start + 2
return bin2int(self.data[start:stop])
@cached
def nagc(self) -> int:
start = 61 + self.desclen
stop = start + 2
return bin2int(self.data[start:stop])
@cached
def npad(self) -> int:
start = 63 + self.desclen
stop = start + 1
return bin2int(self.data[start:stop])
@cached
def ndata(self) -> int:
start = 64 + self.desclen
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def tunning(self) -> tuple:
start = 68 + self.desclen
stop = start + 4 * self.ntun
return L(self.data[i : i + 2] for i in range(start, stop, 4))
@cached
def agc(self) -> np.ndarray:
start = 68 + self.desclen + 4 * self.ntun
stop = start + self.nagc
return np.frombuffer(self.data[start:stop], np.uint8)
@cached
def raw_data(self) -> np.ndarray:
try:
return np.frombuffer(self.data[self.start : self.stop], dtype=np.uint8)
except ValueError as e:
raise ValueError(f"{e} while retrieving raw_data in {self.filename}") from e
@cached
def levels(self) -> np.ndarray:
return ((self.raw_data / 2) + self.minimum).astype(np.float32)
@cached
def padding(self) -> bytes:
return self.data[self.stop :]
@cached
def frequencies(self) -> np.ndarray:
"""Retorna um numpy array com a faixa de frequências presentes no bloco"""
return np.linspace(
self.start_mega, self.stop_mega, num=self.ndata, dtype=np.float32
)
class DType68(GetAttr):
"""Data Type 68 – Threshold Compressed Data"""
def __init__(self, block: BaseBlock):
"""This implementation substitues inheritance of the class Block by Composition
The attributes which belong to Block are accessed normally as if it was Inherited
"""
self.default = DType67(block)
self.start = 76 + self.desclen + 4 * self.ntun + self.nagc
self.stop = self.start + self.default.ndata
@cached
def thresh(self) -> int:
"""THRESH = Threshold Level in dB. All data below this level will be run length encoded"""
start = 68 + self.desclen
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def ndata(self) -> int:
"""NORIG = Number of original data points. Before compression"""
start = 72 + self.desclen
stop = start + 4
return bin2int(self.data[start:stop])
@cached
def levels(self) -> np.ndarray:
"""Spectrum Data in 'dB' with 0.5 dBm interval"""
return np.frombuffer(self.data[self.start : self.stop], dtype=np.uint8)
@cached
def tunning(self) -> tuple:
start = 76 + self.desclen
stop = start + 4 * self.ntun
return [bin2int(self.data[i : i + 2]) for i in range(start, stop, 4)]
@cached
def agc(self) -> np.ndarray:
start = 76 + self.desclen + 4 * self.ntun
stop = start + self.nagc
try:
return np.frombuffer(self.data[start:stop], np.uint8)
except ValueError as e:
raise ValueError(f"{e} while retrieving agc in {self.filename}") from e
@cached
def padding(self) -> int:
return bin2int(self.data[self.stop :])
class DType69(GetAttr):
"""Data Type 69 – Frequency Channel Occupancy (ITU-R SM.1880)"""
def __init__(self, block: BaseBlock):
"""This implementation substitues inheritance of the class Block by Composition
The attributes which belong to Block are accessed normally as if it was Inherited
"""
self.default = TimedVersion5(block)
self.start = 64 + self.desclen
self.stop = self.start + self.ndata
@cached
def description(self) -> str:
return bin2str(self.data[24 : 24 + self.desclen])
@cached
def start_mega(self) -> int:
return bin2int(self.data[24 + self.desclen : 26 + self.desclen], False)
@cached
def start_mili(self) -> int:
return bin2int(self.data[26 + self.desclen : 30 + self.desclen])
@cached
def stop_mega(self) -> int:
return bin2int(self.data[30 + self.desclen : 32 + self.desclen], False)
@cached
def stop_mili(self) -> int:
return bin2int(self.data[32 + self.desclen : 36 + self.desclen])
@cached
def bw(self) -> int:
return bin2int(self.data[36 + self.desclen : 40 + self.desclen])
@cached
def start_channel(self) -> int:
return bin2int(self.data[40 + self.desclen : 42 + self.desclen])
@cached
def stop_channel(self) -> int:
return bin2int(self.data[42 + self.desclen : 44 + self.desclen])
@cached
def opcount(self) -> int:
return bin2int(self.data[48 + self.desclen : 49 + self.desclen], False)
@cached
def antuid(self) -> int:
return bin2int(self.data[49 + self.desclen : 50 + self.desclen], False)
@cached
def dtype(self) -> int:
return DICT_UNIT.get(
bin2int(self.data[50 + self.desclen : 52 + self.desclen], False), "Unknown"
)
@cached
def gerror(self) -> int:
return bin2int(self.data[52 + self.desclen : 53 + self.desclen])
@cached
def gflags(self) -> int:
return bin2int(self.data[53 + self.desclen : 54 + self.desclen])
@cached
def npad(self) -> int:
return bin2int(self.data[55 + self.desclen : 56 + self.desclen])
@cached
def thresh(self) -> int:
return bin2int(self.data[56 + self.desclen : 58 + self.desclen])
@cached
def duration(self) -> int:
return bin2int(self.data[58 + self.desclen : 60 + self.desclen])
@cached
def ndata(self) -> int:
return bin2int(self.data[60 + self.desclen : 64 + self.desclen])
@cached
def levels(self) -> np.ndarray:
try:
return (
np.frombuffer(self.data[self.start : self.stop], dtype=np.uint8) / 2.0
).astype(np.float32)
except ValueError as e:
raise ValueError(f"{e} while retrieving levels in {self}") from e
@cached
def padding(self) -> int:
return bin2int(self.data[self.stop :])