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_run_module_wrong_tag(self) -> None:
pipeline = Pypeline(self.test_dir, self.test_dir, self.test_dir)
module = FitsReadingModule(name_in='read')
pipeline.add_module(module)
module = FitsWritingModule(name_in='write',
file_name='result.fits',
data_tag='im_list')
pipeline.add_module(module)
module = BadPixelSigmaFilterModule(name_in='badpixel',
image_in_tag='im_list',
image_out_tag='im_out')
pipeline.add_module(module)
with pytest.raises(AttributeError) as error:
pipeline.run_module('badpixel')
assert str(error.value) == 'Pipeline module \'badpixel\' is looking for data under a ' \
'tag which does not exist in the database.'
with pytest.warns(UserWarning) as warning:
pipeline.add_module(module)
assert len(warning) == 1
assert warning[0].message.args[0] == 'Pipeline module names need to be unique. ' \
'Overwriting module \'read2\'.'
module = BadPixelSigmaFilterModule(name_in='badpixel',
image_in_tag='im_arr1',
image_out_tag='im_out')
assert pipeline.add_module(module) is None
module = FitsWritingModule(name_in='write1',
file_name='result.fits',
data_tag='im_arr1')
assert pipeline.add_module(module) is None
module = FitsWritingModule(name_in='write2',
file_name='result.fits',
data_tag='im_arr1',
output_dir=self.test_dir)
assert pipeline.add_module(module) is None
assert pipeline.run() is None
assert pipeline.get_module_names() == ['read1', 'read2', 'badpixel', 'write1', 'write2']
def test_not_overwritten(self) -> None:
module = FitsWritingModule(file_name='test.fits',
name_in='write5',
output_dir=None,
data_tag='images',
data_range=None,
overwrite=False,
subset_size=None)
self.pipeline.add_module(module)
with pytest.warns(UserWarning) as warning:
self.pipeline.run_module('write5')
assert len(warning) == 1
assert warning[0].message.args[0] == 'Filename already present. Use overwrite=True ' \
'to overwrite an existing FITS file.'
def test_fits_writing(self) -> None:
module = FitsWritingModule(file_name='test.fits',
name_in='write1',
output_dir=None,
data_tag='images',
data_range=None,
overwrite=True)
self.pipeline.add_module(module)
self.pipeline.run_module('write1')
def test_data_range(self) -> None:
module = FitsWritingModule(file_name='test.fits',
name_in='write4',
output_dir=None,
data_tag='images',
data_range=(0, 10),
overwrite=True,
subset_size=None)
self.pipeline.add_module(module)
self.pipeline.run_module('write4')
def test_write_fits(self):
writing = FitsWritingModule(name_in="fits_writing",
file_name="test.fits",
data_tag="res_mean")
self.pipeline.add_module(writing)
self.pipeline.run_module("fits_writing")
assert os.path.exists(self.test_dir+"test.fits")
assert warning[0].message.args[0] == 'Pipeline module names need to be unique. ' \
'Overwriting module \'read2\'.'
module = BadPixelSigmaFilterModule(name_in='badpixel',
image_in_tag='im_arr1',
image_out_tag='im_out')
assert pipeline.add_module(module) is None
module = FitsWritingModule(name_in='write1',
file_name='result.fits',
data_tag='im_arr1')
assert pipeline.add_module(module) is None
module = FitsWritingModule(name_in='write2',
file_name='result.fits',
data_tag='im_arr1',
output_dir=self.test_dir)
assert pipeline.add_module(module) is None
assert pipeline.run() is None
assert pipeline.get_module_names() == ['read1', 'read2', 'badpixel', 'write1', 'write2']
os.remove(self.test_dir+'result.fits')
os.remove(self.test_dir+'PynPoint_database.hdf5')
def test_subset_size_data_range(self) -> None:
module = FitsWritingModule(file_name='test.fits',
name_in='write7',
output_dir=None,
data_tag='images',
data_range=(8, 18),
overwrite=True,
subset_size=10)
self.pipeline.add_module(module)
self.pipeline.run_module('write7')
def test_attribute_length(self) -> None:
text = 'long_text_long_text_long_text_long_text_long_text_long_text_long_text_long_text'
self.pipeline.set_attribute('images', 'short', 'value', static=True)
self.pipeline.set_attribute('images', 'longer_than_eight1', 'value', static=True)
self.pipeline.set_attribute('images', 'longer_than_eight2', text, static=True)
module = FitsWritingModule(file_name='test.fits',
name_in='write8',
output_dir=None,
data_tag='images',
data_range=None,
overwrite=True,
subset_size=None)
self.pipeline.add_module(module)
with pytest.warns(UserWarning) as warning:
self.pipeline.run_module('write8')
assert len(warning) == 1
assert warning[0].message.args[0] == 'Key \'hierarch longer_than_eight2\' with value ' \
'\'long_text_long_text_long_text_long_text_long_' \
used to save a subsets of a large dataset. The whole dataset will be exported if set
to None.
overwrite : bool, None
Overwrite an existing FITS file with an identical filename.
subset_size : int, None
Size of the subsets that are created when storing the data. This can be useful if the
dataset contains a large number of images. An increasing index value is appended to
the FITS file names. All images are written to a single FITS file if set to None.
Returns
-------
NoneType
None
"""
super(FitsWritingModule, self).__init__(name_in=name_in, output_dir=output_dir)
if not file_name.endswith('.fits'):
raise ValueError('Output \'file_name\' requires the FITS extension.')
self.m_file_name = file_name
self.m_data_port = self.add_input_port(data_tag)
self.m_range = data_range
self.m_overwrite = overwrite
self.m_subset_size = subset_size