Skip to content
Snippets Groups Projects
Commit 737e7578 authored by Mattia Mancini's avatar Mattia Mancini
Browse files

SSB-47: Removed unused module

parent 9eeaf459
No related branches found
No related tags found
1 merge request!44Merge back holography to master
import logging
import numpy
logger = logging.getLogger(__name__)
def _check_index_type_or_raise_index_error(key):
"""
:param key: the key to be checked
:type key; tuple(str)
:return: None
"""
if not isinstance(key, tuple) and not isinstance(key, str):
raise IndexError("please specify a string or a tuple. Key is " + type(key))
class LazyH5Table():
def __init__(self, uri, h5_file):
"""
Constructs the LazyH5Table
:param uri: location of the data in the H5 hierarchy
:type uri: str
:param h5_file: file descriptor of the h5 file
:type h5_file: h5py.File
"""
self.uri = uri
self.h5_file = h5_file
def __get_data_set(self, key, if_not_present_create=False):
"""
Get the item at the specified index
:param key: index key
:type key: tuple(str)
:param if_not_present_create: if not present create subgroup or dataset
:return: h5py.Dataset
"""
if if_not_present_create and self.uri not in self.h5_file:
self.h5_file.create_group(self.uri)
data_set = self.h5_file[self.uri]
_check_index_type_or_raise_index_error(key)
for index in key:
if if_not_present_create and index not in data_set:
data_set.create_group(index)
data_set = data_set[index]
return data_set
def __getitem__(self, key):
"""
Get the item at the specified index
:param key: index key
:type key: tuple(str)
:return:
"""
data_item = self.__get_data_set(key)
value = data_item
return value
def __setitem__(self, key, value):
"""
Set the item at the specified index
:param key: index key
:param value: value to be set
:return:
"""
data_item = self.__get_data_set(key[:-1], if_not_present_create=True)
data_item[key[-1]] = value
class HolographyDataTable(LazyH5Table):
def __init__(self, uri, h5_file):
super(HolographyDataTable, self).__init__(uri, h5_file)
self._frequencies = set()
self._reference_stations = set()
self._beam_numbers = set()
self.__try_construct_index_array()
def __try_construct_index_array(self):
"""
Try constructing the indexes if they exists
:return:
"""
try:
self.__get_index_arrays()
except KeyError:
logger.debug('uri missing, creating datable at: ' + self.uri)
def __get_index_arrays(self):
"""
Get the a list of indices for each datatable axis.
Note. it is assumed that the @see Holography Data Set is respected.
Hence, the first axis is the frequency
the second axis is the reference station
the third axis is the beam number
:return: None
"""
reference_stations = [station_name
for station_name in self.h5_file[self.uri]]
# I select the first frequency just for convenience to iterate on the second level
first_reference_station = reference_stations[0]
frequencies_string = [frequency
for frequency in self.h5_file[self.uri][first_reference_station]]
# I select the first reference station just for convenience to iterate on the third level
first_frequency = frequencies_string[0]
beam_numbers = {beam_number for beam_number in self.h5_file[self.uri]
[first_reference_station]
[first_frequency]}
self._frequencies = tuple(sorted(map(float, frequencies_string)))
self._reference_stations = tuple(reference_stations)
self._beam_numbers = tuple(sorted(map(int, beam_numbers)))
@property
def frequencies(self):
return self._frequencies
@property
def reference_stations(self):
return self._reference_stations
@property
def beam_numbers(self):
return self._beam_numbers
def __getitem__(self, key):
"""
Get the value in the specified key
:param key: (frequency, reference station name, beam number) to get
:type key: (float, str, int)
:rtype: numpy.ndarray
"""
HolographyDataTable.__raise_if_index_is_invalid(key)
reference_station, frequency, beam_number = key
return super().__getitem__((reference_station,
str(frequency),
str(beam_number)))
@staticmethod
def __raise_if_index_is_invalid(index):
try:
reference_station, frequency, beam_number = index
except ValueError:
raise IndexError('Index error (station_name, frequency, beamlet) required got ' +
repr(index))
if not isinstance(reference_station, str):
raise IndexError('Reference station is expected to be a string. Got : ' +
str(type(reference_station)))
if not isinstance(frequency, float):
raise IndexError('Frequency is expected to be a float. Got : ' +
str(type(frequency)))
if not isinstance(beam_number, int):
raise IndexError('Beam number is expected to be a int. Got : ' +
str(type(beam_number)))
def __setitem__(self, key, value):
"""
Set the value in the specified key
:param key: (frequency, reference station name, beam number) to set
:type key: (float, str, int)
:param value: value to set
:return: None
"""
HolographyDataTable.__raise_if_index_is_invalid(key)
reference_station, frequency, beam_number = key
self._frequencies.add(float(frequency))
self._reference_stations.add(reference_station)
self._beam_numbers.add(int(beam_number))
super().__setitem__((reference_station,
str(frequency),
str(beam_number)), value)
def iter_items(self):
for frequency in self._frequencies:
for station in self._reference_stations:
for beam in self._beam_numbers:
yield (station, float(frequency), int(beam)), \
self.__getitem__((station, frequency, beam))
def keys(self):
key_set = set()
for station in self._reference_stations:
for frequency in self._frequencies:
for beam in self._beam_numbers:
key_set.add((station, frequency, beam))
return key_set
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment