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_create_instance_access_data(self) -> None:
with pytest.raises(ValueError) as error:
InputPort('config', self.storage)
assert str(error.value) == 'The tag name \'config\' is reserved for the central ' \
'configuration of PynPoint.'
with pytest.raises(ValueError) as error:
InputPort('fits_header', self.storage)
assert str(error.value) == 'The tag name \'fits_header\' is reserved for storage of the ' \
'FITS headers.'
port = InputPort('images', self.storage)
assert port[0, 0, 0] == pytest.approx(0.00032486907273264834, rel=self.limit, abs=0.)
data = np.mean(port.get_all())
assert data == pytest.approx(1.1824138000882435e-05, rel=self.limit, abs=0.)
assert len(port[0:2, 0, 0]) == 2
assert port.get_shape() == (5, 11, 11)
assert port.get_attribute('PIXSCALE') == 0.01
assert port.get_attribute('PARANG')[0] == 0.
with pytest.warns(UserWarning):
assert port.get_attribute('none') is None
def test_create_instance_access_non_existing_data(self) -> None:
port = InputPort('test', self.storage)
with pytest.warns(UserWarning):
assert port[0, 0, 0] is None
with pytest.warns(UserWarning):
assert port.get_all() is None
with pytest.warns(UserWarning):
assert port.get_shape() is None
with pytest.warns(UserWarning):
assert port.get_attribute('num_files') is None
with pytest.warns(UserWarning):
assert port.get_all_non_static_attributes() is None
with pytest.raises(ValueError) as error:
OutputPort('config', self.storage)
assert str(error.value) == 'The tag name \'config\' is reserved for the central ' \
'configuration of PynPoint.'
with pytest.raises(ValueError) as error:
OutputPort('fits_header', self.storage)
assert str(error.value) == 'The tag name \'fits_header\' is reserved for storage of the ' \
'FITS headers.'
active_port = OutputPort('test', self.storage, activate_init=True)
deactive_port = OutputPort('test', self.storage, activate_init=False)
control_port = InputPort('test', self.storage)
deactive_port.open_port()
deactive_port.set_all(np.asarray([0, 1, 2, 3]))
deactive_port.flush()
with pytest.warns(UserWarning) as warning:
control_port.get_all()
assert len(warning) == 1
assert warning[0].message.args[0] == 'No data under the tag which is linked by the ' \
'InputPort.'
active_port.set_all(np.asarray([0, 1, 2, 3]))
active_port.flush()
def test_create_instance_no_data_storage(self) -> None:
port = InputPort('test')
with pytest.warns(UserWarning):
assert port[0, 0, 0] is None
with pytest.warns(UserWarning):
assert port.get_all() is None
with pytest.warns(UserWarning):
assert port.get_shape() is None
with pytest.warns(UserWarning):
assert port.get_all_non_static_attributes() is None
with pytest.warns(UserWarning):
assert port.get_all_static_attributes() is None
def test_get_ndim(self) -> None:
with pytest.warns(UserWarning) as warning:
ndim = InputPort('images', None).get_ndim()
assert len(warning) == 1
assert warning[0].message.args[0] == 'InputPort can not load data unless a database is ' \
'connected.'
assert ndim is None
port = InputPort('images', self.storage)
assert port.get_ndim() == 3
def test_get_all_attributes(self) -> None:
port = InputPort('images', self.storage)
assert port.get_all_static_attributes() == {'PIXSCALE': 0.01}
assert port.get_all_non_static_attributes() == ['PARANG', ]
port = OutputPort('images', self.storage)
assert port.del_all_attributes() is None
port = InputPort('images', self.storage)
assert port.get_all_non_static_attributes() is None
def test_create_instance_access_data(self) -> None:
with pytest.raises(ValueError) as error:
InputPort('config', self.storage)
assert str(error.value) == 'The tag name \'config\' is reserved for the central ' \
'configuration of PynPoint.'
with pytest.raises(ValueError) as error:
InputPort('fits_header', self.storage)
assert str(error.value) == 'The tag name \'fits_header\' is reserved for storage of the ' \
'FITS headers.'
port = InputPort('images', self.storage)
assert port[0, 0, 0] == pytest.approx(0.00032486907273264834, rel=self.limit, abs=0.)
data = np.mean(port.get_all())
assert data == pytest.approx(1.1824138000882435e-05, rel=self.limit, abs=0.)
assert len(port[0:2, 0, 0]) == 2
assert port.get_shape() == (5, 11, 11)
assert port.get_attribute('PIXSCALE') == 0.01
assert port.get_attribute('PARANG')[0] == 0.
port = self._m_input_ports[tag]
or by using the returned Port.
Parameters
----------
tag : str
Tag of the new input port.
Returns
-------
pynpoint.core.dataio.InputPort
The new InputPort for the ProcessingModule.
"""
port = InputPort(tag)
if self._m_data_base is not None:
port.set_database_connection(self._m_data_base)
self._m_input_ports[tag] = port
return port
----------
tag : str
The tag of the port. The port can be used in order to get data from the dataset with
the key `tag`.
data_storage_in : pynpoint.core.dataio.DataStorage
It is possible to give the constructor of an InputPort a DataStorage instance which
will link the port to that DataStorage. Usually the DataStorage is set later by calling
:func:`~pynpoint.core.dataio.Port.set_database_connection`.
Returns
-------
NoneType
None
"""
super(InputPort, self).__init__(tag, data_storage_in)
if tag == 'config':
raise ValueError('The tag name \'config\' is reserved for the central configuration '
'of PynPoint.')
if tag == 'fits_header':
raise ValueError('The tag name \'fits_header\' is reserved for storage of the FITS '
'headers.')
port = self._m_input_ports[tag]
or by using the returned Port.
Parameters
----------
tag : str
Tag of the new input port.
Returns
-------
pynpoint.core.dataio.InputPort
The new InputPort for the WritingModule.
"""
port = InputPort(tag)
if self._m_data_base is not None:
port.set_database_connection(self._m_data_base)
self._m_input_ports[tag] = port
return port