Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
The pipeline module.
tags : list(str, )
Tags in the database.
Returns
-------
bool
Module validation.
str
Module name.
"""
if isinstance(module, ReadingModule):
tags.extend(module.get_all_output_tags())
elif isinstance(module, WritingModule):
for tag in module.get_all_input_tags():
if tag not in tags:
return False, module.name
elif isinstance(module, ProcessingModule):
tags.extend(module.get_all_output_tags())
for tag in module.get_all_input_tags():
if tag not in tags:
return False, module.name
else:
return False, None
return True, None
"""
Module for writing a list of tags from the database to a separate HDF5 file.
"""
import os
from typing import Optional
import h5py
from typeguard import typechecked
from pynpoint.core.processing import WritingModule
class Hdf5WritingModule(WritingModule):
"""
Module which exports a part of the PynPoint internal database to a separate HDF5 file. The
datasets of the database can be chosen using the *tag_dictionary*. The module will also export
the static and non-static attributes.
"""
__author__ = 'Markus Bonse, Tomas Stolker'
@typechecked
def __init__(self,
name_in: str,
file_name: str,
output_dir: Optional[str] = None,
tag_dictionary: Optional[dict] = None,
keep_attributes: bool = True,
overwrite: bool = False) -> None:
functions inheriting from this class.
Parameters
----------
name_in : str
The name of the WritingModule.
output_dir : str
Directory where the results will be saved.
Returns
-------
NoneType
None
"""
super(WritingModule, self).__init__(name_in)
assert (os.path.isdir(str(output_dir)) or output_dir is None), 'Output directory for ' \
'writing module does not exist - input requested: %s.' % output_dir
self.m_output_location = output_dir
self._m_input_ports = {}
import os
import warnings
from typing import Optional, Tuple
from astropy.io import fits
from typeguard import typechecked
import numpy as np
from pynpoint.core.processing import WritingModule
from pynpoint.util.module import memory_frames
class FitsWritingModule(WritingModule):
"""
Module for writing a dataset from the central HDF5 database to a FITS file. The static
attributes will be stored as header information. The dataset is selected from the database
by its tag name. :class:`~pynpoint.readwrite.fitswriting.FitsWritingModule` is a
:class:`~pynpoint.core.processing.WritingModule` and uses either the default output directory
of a :class:`~pynpoint.core.pypeline.Pypeline` or a specified location to store the FITS data.
"""
__author__ = 'Markus Bonse, Tomas Stolker'
@typechecked
def __init__(self,
name_in: str,
data_tag: str,
file_name: str,
output_dir: Optional[str] = None,
"""
Modules for writing data as text file.
"""
import os
import numpy as np
from typing import Optional
from typeguard import typechecked
from pynpoint.core.processing import WritingModule
class AttributeWritingModule(WritingModule):
"""
Module for writing a 1D or 2D array of non-static attributes to a text file.
"""
__author__ = 'Tomas Stolker'
@typechecked
def __init__(self,
name_in: str,
data_tag: str,
attribute: str,
file_name: str = 'attributes.dat',
output_dir: Optional[str] = None,
header: Optional[str] = None) -> None:
"""
Parameters
"""
Modules for writing data as text file.
"""
import os
from typing import Optional
import numpy as np
from typeguard import typechecked
from pynpoint.core.processing import WritingModule
class TextWritingModule(WritingModule):
"""
Module for writing a 1D or 2D data set from the central HDF5 database as text file.
TextWritingModule is a :class:`pynpoint.core.processing.WritingModule` and supports
the use of the Pypeline default output directory as well as a specified location.
"""
__author__ = 'Tomas Stolker'
@typechecked
def __init__(self,
name_in: str,
data_tag: str,
file_name: str,
output_dir: Optional[str] = None,
header: Optional[str] = None) -> None:
"""
end of this ordered dictionary. If the input module is a reading or writing module without
a specified input or output location then the Pypeline default location is used. Moreover,
the given module is connected to the Pypeline internal data storage.
Parameters
----------
module : ReadingModule, WritingModule, or ProcessingModule
Input pipeline module.
Returns
-------
NoneType
None
"""
if isinstance(module, WritingModule):
if module.m_output_location is None:
module.m_output_location = self._m_output_place
if isinstance(module, ReadingModule):
if module.m_input_location is None:
module.m_input_location = self._m_input_place
module.connect_database(self.m_data_storage)
if module.name in self._m_modules:
warnings.warn(f'Pipeline module names need to be unique. Overwriting module '
f'\'{module.name}\'.')
self._m_modules[module.name] = module
functions inheriting from this class.
Parameters
----------
name_in : str
The name of the WritingModule.
output_dir : str
Directory where the results will be saved.
Returns
-------
NoneType
None
"""
super(WritingModule, self).__init__(name_in)
assert (os.path.isdir(str(output_dir)) or output_dir is None), 'Output directory for ' \
'writing module does not exist - input requested: %s.' % output_dir
self.m_output_location = output_dir
self._m_input_ports = {}