How to use the pynpoint.core.dataio.InputPort function in pynpoint

To help you get started, we’ve selected a few pynpoint examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github PynPoint / PynPoint / tests / test_core / test_inputport.py View on Github external
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
github PynPoint / PynPoint / tests / test_core / test_inputport.py View on Github external
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
github PynPoint / PynPoint / tests / test_core / test_outputport.py View on Github external
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()
github PynPoint / PynPoint / tests / test_core / test_inputport.py View on Github external
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
github PynPoint / PynPoint / tests / test_core / test_inputport.py View on Github external
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
github PynPoint / PynPoint / tests / test_core / test_inputport.py View on Github external
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
github PynPoint / PynPoint / tests / test_core / test_inputport.py View on Github external
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.
github PynPoint / PynPoint / pynpoint / core / processing.py View on Github external
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
github PynPoint / PynPoint / pynpoint / core / dataio.py View on Github external
----------
        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.')
github PynPoint / PynPoint / pynpoint / core / processing.py View on Github external
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