Este módulo processa o arquivo bin e extrai os metadados e dados do espectro dos blocos, além de criar estatísticas das medições.
%load_ext autoreload
%load_ext line_profiler
%load_ext cython
%autoreload 2
class CrfsGPS:
"""Class with the GPS Attributes from the CRFS Bin File"""
def __init__(self) -> None:
self._data: L = L()
def __len__(self):
return len(self._data)
def __getitem__(self, key):
return self._latitude[key], self._longitude[key], self._altitude[key], self._num_satellites[key]
def __iter__(self):
return zip(self._latitude, self._longitude, self._altitude, self._num_satellites)
@cached
def _gps_datetime(self):
return self._data.attrgot("gps_datetime")
@cached
def _latitude(self):
return self._data.attrgot("latitude")
@cached
def _longitude(self):
return self._data.attrgot("longitude")
@cached
def _altitude(self):
return self._data.attrgot("altitude")
@cached
def _num_satellites(self):
return self._data.attrgot("num_satellites")
@property
def latitude(self) -> float:
return np.median(self._latitude) if self._latitude else -1
@property
def longitude(self) -> float:
return np.median(self._longitude) if self._longitude else -1
@property
def altitude(self) -> float:
return np.median(self._altitude) if self._altitude else -1
@property
def num_satellites(self) -> float:
return np.median(self._num_satellites) if self._num_satellites else 0
def __repr__(self):
return f"GPS Data - Median of Coordinates: {self.latitude:.5f}:{self.longitude:.5f} Altitude: {self.altitude:.2f} #Satellites: {self.num_satellites:.1f}"
class CrfsSpectrum(GetAttr):
"""Class with the metadata and levels of a spectrum block from a CRFS Bin File"""
def __init__(self, metadata, precision=np.float32):
self.default = metadata
self._data: L = L()
self.precision = precision
def __getitem__(self, key):
return self.timestamp[key], self.levels[key]
def __iter__(self):
return zip(self.timestamp, self.levels)
def __len__(self):
return len(self._data)
def __repr__(self):
return repr(self.default)
def __str__(self):
return f"""Blocks of Type: {self.type}, Thread_id: {self.thread_id}, Start: {self.start_mega} MHz, Stop: {self.stop_mega} MHz"""
@cached
def timestamp(self):
return self._data.attrgot('wallclock_datetime')
@cached
def start_dateidx(self):
return getattr(self._data[0], 'wallclock_datetime').item()
@cached
def stop_dateidx(self):
return getattr(self._data[-1], 'wallclock_datetime').item()
@cached
def levels(self):
"""Return the spectrum levels"""
if self.type in UNCOMPRESSED:
levels = np.empty((len(self._data), self.ndata), dtype=self.precision)
for i, level in enumerate(self._data.attrgot('levels')):
levels[i,:] = level
# levels = np.concatenate(self._data.attrgot('levels')).reshape((-1, self.ndata))
elif self.type in COMPRESSED:
levels = cy_extract_compressed(
list(self._data.attrgot('levels')),
len(self._data),
int(self.ndata),
int(self.thresh),
float(self.minimum),
)
else:
raise ValueError(
"The current block is not of type spectrum or it's not implemented yet"
)
if self.precision != np.float32:
levels = levels.astype(self.precision)
return levels
@cached
def frequencies(self) -> np.ndarray:
return np.linspace(self.start_mega, self.stop_mega, num=self.ndata)
def matrix(self):
"""Returns the matrix formed from the spectrum levels and timestamp"""
index = self.timestamp if len(self.timestamp) == len(self) else None
data = pd.DataFrame(self.levels, index=index, columns=self.frequencies)
data.columns.name = "Frequencies"
data.index.name = "Time"
return data
A função a seguir recebe os bytes lidos do arquivo .bin e mapeia esses bytes em diferentes classes de acordo com o tipo de bloco
A função a seguir recebe os bytes lidos do arquivo .bin e mapeia esses bytes em diferentes classes de acordo com o tipo de bloco
def parse_bin(bin_file: Union[str, Path], precision=np.float32) -> dict:
"""Receives a CRFS binfile and returns a dictionary with the file metadata, a GPS Class and a list with the different Spectrum Classes
A block is a piece of the .bin file with a known start and end and that contains different types of information.
It has several fields: file_type, header, data and footer.
Each field has lengths and information defined in the documentation.
Args:
bin_file (Union[str, Path]): path to the bin file
Returns:
Dictionary with the file metadata, file_version, string info, gps and spectrum blocks.
"""
bin_file = Path(bin_file)
meta = {}
fluxos = {}
gps = CrfsGPS()
with open(bin_file, mode="rb") as file:
# The first block of the file is the header and is 36 bytes long.
header = file.read(BYTES_HEADER)
meta["filename"] = bin_file.name
meta["file_version"] = bin2int(header[:4])
if meta['file_version'] == 21:
meta['method'] = 'Script_CRFSBINv2'
meta["string"] = bin2str(header[4:])
file_size = file.seek(0, 2)
file.seek(36, 0)
while (next_block := file.tell()) < file_size:
block_type, block = create_block(file, next_block)
if block is None:
continue
if block_type in (2, 40):
gps._data.append(block)
elif block_type in VECTOR_BLOCKS:
append_spec_data(block_type,fluxos, block, precision)
else:
meta.update(getattrs(block, KEY_ATTRS.get(block_type)))
meta["gps"] = gps
meta["spectrum"] = L(fluxos.values())
meta['hostname'] = meta['hostname'][:2].upper() + meta['hostname'][2:]
if modtime := getattr(gps, '_gps_datetime'):
modtime = modtime[-1] #.astype(datetime)
else:
modtime = np.datetime64(datetime.fromtimestamp(bin_file.stat().st_mtime))
def set_timestamp(spec)-> None:
"""Create a timestamp from the file modification time backwards by 1s for each measurement"""
timestamp = L()
mtime = modtime
for s in range(len(spec)):
timestamp.append(mtime)
mtime -= np.timedelta64(1, "s")
timestamp.reverse()
setattr(spec, 'timestamp', timestamp)
meta['spectrum'].filter(lambda x: x.type in (4,7)).map(set_timestamp)
return meta
files = get_files('D:\OneDrive - ANATEL\BinFiles\Combo1 (RFLook - arquivos iniciais)')
files
file = r'D:\OneDrive - ANATEL\SpecFiles\Combo3 (CRFS Bin - DataTypes 4, 7, 8, 60-65 e 67-69)\rfeye002292_210208_T203238_CRFSBINv.3.bin'
#file = r'D:\OneDrive - ANATEL\SpecFiles\Combo3 (CRFS Bin - DataTypes 4, 7, 8, 60-65 e 67-69)\rfeye002292_210208_T202215_CRFSBINv.4.bin'
dados = parse_bin(file)
dados
from pathlib import Path
list(Path('D:\OneDrive - ANATEL\SpecFiles').iterdir())
print(dados['gps']._gps_datetime)
%%cython --annotate
cimport cython
ctypedef object CrfsGPS
@cython.boundscheck(False)
@cython.wraparound(False)