"""
Modules for reading attributes from a FITS or ASCII file.
"""
import os
import warnings
from typing import Optional
import numpy as np
from astropy.io import fits
from typeguard import typechecked
from pynpoint.core.attributes import get_attributes
from pynpoint.core.processing import ReadingModule
[docs]class AttributeReadingModule(ReadingModule):
"""
Module for reading a list of values from a FITS or ASCII file and appending them as a non-static
attributes to a dataset.
"""
__author__ = 'Tomas Stolker'
@typechecked
def __init__(self,
name_in: str,
data_tag: str,
file_name: str,
attribute: str,
input_dir: Optional[str] = None,
overwrite: bool = False) -> None:
"""
Parameters
----------
name_in : str
Unique name of the module instance.
data_tag : str
Tag of the database entry to which the attribute is written.
file_name : str
Name of the input file with the attribute value. Should be equal in size to the number
of images in ``data_tag``. In case the ``file_name`` is ending with ``.fits``, then a
FITS file is read. Otherwise, a single column of values is expected in an ASCII file.
file_name : str
Name of the input file with a list of values.
attribute : str
Name of the attribute as to be written in the database.
input_dir : str, None
Input directory where the input file is located. If not specified the Pypeline default
directory is used.
overwrite : bool
Overwrite if the attribute is already exists.
Returns
-------
NoneType
None
"""
super().__init__(name_in, input_dir=input_dir)
self.m_data_port = self.add_output_port(data_tag)
self.m_file_name = file_name
self.m_attribute = attribute
self.m_overwrite = overwrite
[docs] @typechecked
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)
if status == 1:
self.m_data_port.add_attribute(self.m_attribute, values, static=False)
elif status == -1 and self.m_overwrite:
self.m_data_port.add_attribute(self.m_attribute, values, static=False)
elif status == -1 and not self.m_overwrite:
warnings.warn(f'The attribute \'{self.m_attribute}\' is already present. Set the '
f'\'overwrite\' parameter to True in order to overwrite the values with '
f'{self.m_file_name}.')
elif status == 0:
warnings.warn(f'The \'{self.m_attribute}\' attribute is already present and '
f'contains the same values as are present in {self.m_file_name}.')
print(' [DONE]')
self.m_data_port.close_port()
[docs]class ParangReadingModule(ReadingModule):
"""
Module for reading a list of parallactic angles from a FITS or ASCII file.
"""
__author__ = 'Tomas Stolker'
@typechecked
def __init__(self,
name_in: str,
data_tag: str,
file_name: str,
input_dir: Optional[str] = None,
overwrite: bool = False) -> None:
"""
Parameters
----------
name_in : str
Unique name of the module instance.
data_tag : str
Tag of the database entry to which the ``PARANG`` attribute is written.
file_name : str
Name of the input file with the parallactic angles (deg). Should be equal in size
to the number of images in ``data_tag``. In case the ``file_name`` is ending with
``.fits``, then a FITS file is read. Otherwise, a single column of values is expected
in an ASCII file.
input_dir : str, None
Input directory where the input file is located. If not specified the Pypeline default
directory is used.
overwrite : bool
Overwrite if the ``PARANG`` attribute already exists.
Returns
-------
NoneType
None
"""
super().__init__(name_in, input_dir=input_dir)
self.m_data_port = self.add_output_port(data_tag)
self.m_file_name = file_name
self.m_overwrite = overwrite
[docs] @typechecked
def run(self) -> None:
"""
Run method of the module. Reads the parallactic angles from a FITS or ASCII file and
writes the values as non-static attribute (``PARANG``) to the database tag.
Returns
-------
NoneType
None
"""
print('Reading parallactic angles...', end='')
if self.m_file_name.endswith('fits'):
parang = fits.getdata(os.path.join(self.m_input_location, self.m_file_name))
else:
parang = np.loadtxt(os.path.join(self.m_input_location, self.m_file_name))
print(' [DONE]')
if parang.ndim != 1:
raise ValueError(f'The input file {self.m_file_name} should contain a 1D data set with '
f'the parallactic angles.')
print(f'Number of angles: {parang.size}')
print(f'Rotation range: {parang[0]:.2f} -> {parang[-1]:.2f} deg')
status = self.m_data_port.check_non_static_attribute('PARANG', parang)
if status == 1:
self.m_data_port.add_attribute('PARANG', parang, static=False)
elif status == -1 and self.m_overwrite:
self.m_data_port.add_attribute('PARANG', parang, static=False)
elif status == -1 and not self.m_overwrite:
warnings.warn(f'The PARANG attribute is already present. Set the \'overwrite\' '
f'parameter to True in order to overwrite the values with '
f'{self.m_file_name}.')
elif status == 0:
warnings.warn(f'The PARANG attribute is already present and contains the same values '
f'as are present in {self.m_file_name}.')
self.m_data_port.close_port()
[docs]class WavelengthReadingModule(ReadingModule):
"""
Module for reading a list of wavelengths from a FITS or ASCII file.
"""
__author__ = 'Tomas Stolker'
@typechecked
def __init__(self,
name_in: str,
data_tag: str,
file_name: str,
input_dir: Optional[str] = None,
overwrite: bool = False) -> None:
"""
Parameters
----------
name_in : str
Unique name of the module instance.
data_tag : str
Tag of the database entry to which the ``WAVELENGTH`` attribute is written.
file_name : str
Name of the input file with the wavelengths (a.u.). Should be equal in size
to the number of images in ``data_tag``. In case the ``file_name`` is ending with
``.fits``, then a FITS file is read. Otherwise, a single column of values is expected
in an ASCII file.
input_dir : str, None
Input directory where the input file is located. If not specified the Pypeline default
directory is used.
overwrite : bool
Overwrite if the ``WAVELENGTH`` attribute already exists.
Returns
-------
NoneType
None
"""
super().__init__(name_in, input_dir=input_dir)
self.m_data_port = self.add_output_port(data_tag)
self.m_file_name = file_name
self.m_overwrite = overwrite
[docs] @typechecked
def run(self) -> None:
"""
Run method of the module. Reads the parallactic angles from a FITS or ASCII file and writes
the values as non-static attribute (``WAVELENGTH``) to the database tag.
Returns
-------
NoneType
None
"""
print('Reading wavelengths...', end='')
if self.m_file_name.endswith('fits'):
wavelength = fits.getdata(os.path.join(self.m_input_location, self.m_file_name))
else:
wavelength = np.loadtxt(os.path.join(self.m_input_location, self.m_file_name))
print(' [DONE]')
if wavelength.ndim != 1:
raise ValueError(f'The input file {self.m_file_name} should contain a 1D data set with '
f'the wavelengths.')
print(f'Number of wavelengths: {wavelength.size}')
print(f'Wavelength range: {wavelength[0]:.2f} - {wavelength[-1]:.2f}')
status = self.m_data_port.check_non_static_attribute('WAVELENGTH', wavelength)
if status == 1:
self.m_data_port.add_attribute('WAVELENGTH', wavelength, static=False)
elif status == -1 and self.m_overwrite:
self.m_data_port.add_attribute('WAVELENGTH', wavelength, static=False)
elif status == -1 and not self.m_overwrite:
warnings.warn(f'The WAVELENGTH attribute is already present. Set the \'overwrite\' '
f'parameter to True in order to overwrite the values with '
f'{self.m_file_name}.')
elif status == 0:
warnings.warn(f'The WAVELENGTH attribute is already present and contains the same '
f'values as are present in {self.m_file_name}.')
self.m_data_port.close_port()
