Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_read_data(self) -> None:
module = FitsReadingModule(name_in='read',
image_tag='read',
input_dir=self.test_dir+'prep')
self.pipeline.add_module(module)
self.pipeline.run_module('read')
data = self.pipeline.get_data('read')
assert np.sum(data) == pytest.approx(105.54278879805277, rel=self.limit, abs=0.)
assert data.shape == (10, 11, 11)
module = FitsReadingModule(name_in='read_ifs',
image_tag='read_ifs',
input_dir=self.test_dir+'prep_ifs',
ifs_data=True)
self.pipeline.add_module(module)
def test_read_data(self) -> None:
module = FitsReadingModule(name_in='read',
image_tag='images',
input_dir=self.test_dir+'images',
overwrite=True,
check=True)
self.pipeline.add_module(module)
self.pipeline.run_module('read')
data = self.pipeline.get_data('images')
assert np.sum(data) == pytest.approx(105.54278879805277, rel=self.limit, abs=0.)
assert data.shape == (10, 11, 11)
def test_read_data(self) -> None:
module = FitsReadingModule(name_in='read',
image_tag='read',
input_dir=self.test_dir+'images',
overwrite=True,
check=True)
self.pipeline.add_module(module)
self.pipeline.run_module('read')
data = self.pipeline.get_data('read')
assert np.sum(data) == pytest.approx(105.54278879805277, rel=self.limit, abs=0.)
assert data.shape == (10, 11, 11)
attr = self.pipeline.get_attribute('read', 'NDIT', static=False)
assert np.sum(attr) == pytest.approx(10, rel=self.limit, abs=0.)
assert attr.shape == (2, )
def test_remove_module(self) -> None:
pipeline = Pypeline(self.test_dir, self.test_dir, self.test_dir)
module = FitsReadingModule(name_in='read')
pipeline.add_module(module)
module = BadPixelSigmaFilterModule(name_in='badpixel',
image_in_tag='im_arr1',
image_out_tag='im_out')
pipeline.add_module(module)
assert pipeline.get_module_names() == ['read', 'badpixel']
assert pipeline.remove_module('read')
assert pipeline.get_module_names() == ['badpixel']
assert pipeline.remove_module('badpixel')
with pytest.warns(UserWarning) as warning:
def test_read_data(self):
read_science = FitsReadingModule(name_in="read_science",
input_dir=self.test_dir+"adi",
image_tag="im_arr")
self.pipeline.add_module(read_science)
read_dark = FitsReadingModule(name_in="read_dark",
input_dir=self.test_dir+"dark",
image_tag="dark_arr")
self.pipeline.add_module(read_dark)
read_flat = FitsReadingModule(name_in="read_flat",
input_dir=self.test_dir+"flat",
image_tag="flat_arr")
self.pipeline.add_module(read_flat)
def test_fits_read_textfile_exists(self) -> None:
with open(self.test_dir+'filenames.dat', 'w') as file_obj:
file_obj.write(self.test_dir+'fits/images_0.fits\n')
file_obj.write(self.test_dir+'fits/images_2.fits\n')
module = FitsReadingModule(name_in='read10',
input_dir=None,
image_tag='files',
overwrite=True,
check=True,
filenames=self.test_dir+'filenames.dat')
self.pipeline.add_module(module)
with pytest.raises(ValueError) as error:
self.pipeline.run_module('read10')
assert str(error.value) == f'The file {self.test_dir}fits/images_2.fits does not exist. ' \
f'Please check that the path is correct.'
def test_read_data(self) -> None:
module = FitsReadingModule(name_in='read',
image_tag='data',
input_dir=self.test_dir+'data',
overwrite=True,
check=True)
self.pipeline.add_module(module)
self.pipeline.run_module('read')
data = self.pipeline.get_data('data')
assert np.sum(data) == pytest.approx(105.54278879805277, rel=self.limit, abs=0.)
assert data.shape == (10, 11, 11)
def test_read_data(self) -> None:
module = FitsReadingModule(name_in='read1',
image_tag='star',
input_dir=self.test_dir+'star',
overwrite=True,
check=True)
self.pipeline.add_module(module)
module = FitsReadingModule(name_in='read2',
image_tag='waffle',
input_dir=self.test_dir+'waffle',
overwrite=True,
check=True)
self.pipeline.add_module(module)
self.pipeline.run_module('read1')
def test_non_static_not_found(self) -> None:
self.pipeline.set_attribute('config', 'DIT', 'None', static=True)
for i in range(2):
with fits.open(f'{self.test_dir}/fits/images_{i}.fits') as hdu:
header = hdu[0].header
del header['HIERARCH ESO DET DIT']
del header['HIERARCH ESO DET EXP NO']
hdu.writeto(f'{self.test_dir}/fits/images_{i}.fits', overwrite=True)
module = FitsReadingModule(name_in='read6',
input_dir=self.test_dir+'fits',
image_tag='input',
overwrite=True,
check=True)
self.pipeline.add_module(module)
with pytest.warns(UserWarning) as warning:
self.pipeline.run_module('read6')
assert len(warning) == 2
for item in warning:
assert item.message.args[0] == 'Non-static attribute EXP_NO (=ESO DET EXP NO) not ' \
'found in the FITS header.'
directory is used.
image_tag : str
Tag of the read data in the HDF5 database. Non static header information is stored with
the tag: *header_* + image_tag / header_entry_name.
overwrite : bool
Overwrite existing data and header in the central database.
check : bool
Check all the listed non-static attributes or ignore the attributes that are not always
required (e.g. PARANG_START, DITHER_X).
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):