Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
config_port : pynpoint.core.dataio.ConfigPort
Configuration port.
image_out_port : pynpoint.core.dataio.OutputPort
Output port of the images to which the static attributes are stored.
check : bool
Print a warning if certain attributes from the configuration file are not present in
the FITS header. If set to `False`, attributes are still written to the dataset but
there will be no warning if a keyword is not found in the FITS header.
Returns
-------
NoneType
None
"""
attributes = get_attributes()
static = []
for key, value in attributes.items():
if value['config'] == 'header' and value['attribute'] == 'static':
static.append(key)
for attr in static:
fitskey = config_port.get_attribute(attr)
if isinstance(fitskey, np.bytes_):
fitskey = str(fitskey.decode('utf-8'))
if fitskey != 'None':
if fitskey in header:
status = image_out_port.check_static_attribute(attr, header[fitskey])
Returns
-------
None
"""
super(FitsReadingModule, self).__init__(name_in, input_dir)
self.m_image_out_port = self.add_output_port(image_tag)
self.m_overwrite = overwrite
self.m_check = check
self.m_static = []
self.m_non_static = []
self.m_attributes = get_attributes()
for key, value in six.iteritems(self.m_attributes):
if value["config"] == "header" and value["attribute"] == "static":
self.m_static.append(key)
for key, value in six.iteritems(self.m_attributes):
if value["attribute"] == "non-static":
self.m_non_static.append(key)
self.m_count = 0
Parameters
----------
header : astropy.io.fits.header.Header
Header information from the FITS file that is read.
config_port : pynpoint.core.dataio.ConfigPort
Configuration port.
image_out_port : pynpoint.core.dataio.OutputPort
Output port of the images to which the non-static attributes are stored.
Returns
-------
NoneType
None
"""
attributes = get_attributes()
nonstatic = []
for key, value in attributes.items():
if value['attribute'] == 'non-static':
nonstatic.append(key)
for attr in nonstatic:
if attributes[attr]['config'] == 'header':
fitskey = config_port.get_attribute(attr)
# if type(fitskey) == np.bytes_:
# fitskey = str(fitskey.decode('utf-8'))
if fitskey != 'None':
if fitskey in header:
image_out_port.append_attribute_data(attr, header[fitskey])
self.m_image_out_port_1 = self.add_output_port(image_out_tag_1)
self.m_image_out_port_2 = self.add_output_port(image_out_tag_2)
self.m_image_out_port_3 = self.add_output_port(image_out_tag_3)
self.m_image_out_port_4 = self.add_output_port(image_out_tag_4)
# Parameters
self.m_im_dir = image_in_dir
self.m_check = check
self.m_overwrite = overwrite
self.m_scheme = scheme
# Arguments
self.m_static = []
self.m_non_static = []
self.m_attributes = get_attributes()
for key, value in six.iteritems(self.m_attributes):
if value["config"] == "header" and value["attribute"] == "static":
self.m_static.append(key)
for key, value in six.iteritems(self.m_attributes):
if value["attribute"] == "non-static":
self.m_non_static.append(key)
self.m_count = 0
def run(self) -> None:
"""
Run method of the module. Reads a list of values from a FITS or ASCII file and writes them
as non-static attribute to a dataset.
Returns
-------
NoneType
None
"""
print('Reading attribute data...', end='')
attributes = get_attributes()
if self.m_attribute not in attributes:
raise ValueError(f'\'{self.m_attribute}\' is not a valid attribute.')
if self.m_file_name.endswith('fits'):
values = fits.getdata(os.path.join(self.m_input_location, self.m_file_name))
else:
values = np.loadtxt(os.path.join(self.m_input_location, self.m_file_name),
dtype=attributes[self.m_attribute]['type'])
if values.ndim != 1:
raise ValueError(f'The input file {self.m_file_name} should contain a 1D list with '
f'attributes.')
status = self.m_data_port.check_non_static_attribute(self.m_attribute, values)
del hdf['config']
config = hdf.create_group('config')
for key in attributes.keys():
if attributes[key]['value'] is not None:
config.attrs[key] = attributes[key]['value']
config.attrs['WORKING_PLACE'] = self._m_working_place
hdf.close()
config_file = os.path.join(self._m_working_place, 'PynPoint_config.ini')
print(f'Configuration: {config_file}\n')
attributes = get_attributes()
attributes['CPU']['value'] = multiprocessing.cpu_count()
if not os.path.isfile(config_file):
warnings.warn('Configuration file not found. Creating PynPoint_config.ini with '
'default values in the working place.')
_create_config(config_file, attributes)
attributes = _read_config(config_file, attributes)
_write_config(attributes)
n_cpu = attributes['CPU']['value']
if 'OMP_NUM_THREADS' in os.environ:
n_thread = os.environ['OMP_NUM_THREADS']