lsst.pipe.tasks.ingestCalibs.CalibsParseTask Class Reference

Inheritance diagram for lsst.pipe.tasks.ingestCalibs.CalibsParseTask:

Public Member Functions
def	getCalibType (self, filename)
def	getDestination (self, butler, info, filename)
def	getInfo (self, filename)
def	getInfoFromMetadata (self, md, info=None)
def	translate_date (self, md)
def	translate_filter (self, md)

Static Public Member Functions
def	getExtensionName (md)

Static Public Attributes
	ConfigClass = ParseConfig
	translator_class = None

Detailed Description

Task that will parse the filename and/or its contents to get the
required information to populate the calibration registry.

Definition at line 17 of file ingestCalibs.py.

Member Function Documentation

getCalibType()

def lsst.pipe.tasks.ingestCalibs.CalibsParseTask.getCalibType	(		self,
			filename
	)

Return a a known calibration dataset type using
the observation type in the header keyword OBSTYPE

@param filename: Input filename

Definition at line 21 of file ingestCalibs.py.

    def getCalibType(self, filename):
        """Return a a known calibration dataset type using
        the observation type in the header keyword OBSTYPE
 
        @param filename: Input filename
        """
        md = readMetadata(filename, self.config.hdu)
        if not md.exists("OBSTYPE"):
            raise RuntimeError("Unable to find the required header keyword OBSTYPE in %s, hdu %d" %
                               (filename, self.config.hdu))
        obstype = md.getScalar("OBSTYPE").strip().lower()
        if "flat" in obstype:
            obstype = "flat"
        elif "zero" in obstype or "bias" in obstype:
            obstype = "bias"
        elif "dark" in obstype:
            obstype = "dark"
        elif "fringe" in obstype:
            obstype = "fringe"
        elif "sky" in obstype:
            obstype = "sky"
        elif "illumcor" in obstype:
            obstype = "illumcor"
        elif "defects" in obstype:
            obstype = "defects"
        elif "qe_curve" in obstype:
            obstype = "qe_curve"
        elif "linearizer" in obstype:
            obstype = "linearizer"
        elif "crosstalk" in obstype:
            obstype = "crosstalk"
        elif "BFK" in obstype:
            obstype = "bfk"
        elif "photodiode" in obstype:
            obstype = 'photodiode'
        return obstype
 

getDestination()

def lsst.pipe.tasks.ingestCalibs.CalibsParseTask.getDestination	(		self,
			butler,
			info,
			filename
	)

Get destination for the file

@param butler      Data butler
@param info        File properties, used as dataId for the butler
@param filename    Input filename
@return Destination filename

Reimplemented from lsst.pipe.tasks.ingest.ParseTask.

Reimplemented in lsst.obs.decam.ingestCalibs.DecamCalibsParseTask.

Definition at line 58 of file ingestCalibs.py.

    def getDestination(self, butler, info, filename):
        """Get destination for the file
 
        @param butler      Data butler
        @param info        File properties, used as dataId for the butler
        @param filename    Input filename
        @return Destination filename
        """
        # 'tempinfo' was added as part of DM-5466 to strip Nones from info.
        # The Butler should handle this behind-the-scenes in the future.
        # Please reference DM-9873 and delete this comment once it is resolved.
        tempinfo = {k: v for (k, v) in info.items() if v is not None}
        calibType = self.getCalibType(filename)
        raw = butler.get(calibType + "_filename", tempinfo)[0]
        # Ensure filename is devoid of cfitsio directions about HDUs
        c = raw.find("[")
        if c > 0:
            raw = raw[:c]
        return raw
 
 

getExtensionName()

def lsst.pipe.tasks.ingest.ParseTask.getExtensionName (   md )

staticinherited

 Get the name of a FITS extension.

Parameters
----------
md : `lsst.daf.base.PropertySet`
    FITS header metadata.

Returns
-------
result : `str` or `None`
    The string from the EXTNAME header card if it exists. None otherwise.

Definition at line 126 of file ingest.py.

    def getExtensionName(md):
        """ Get the name of a FITS extension.
 
        Parameters
        ----------
        md : `lsst.daf.base.PropertySet`
            FITS header metadata.
 
        Returns
        -------
        result : `str` or `None`
            The string from the EXTNAME header card if it exists. None otherwise.
        """
        try:
            ext = md["EXTNAME"]
        except KeyError:
            ext = None
        return ext
 

getInfo()

def lsst.pipe.tasks.ingest.ParseTask.getInfo (   self,   filename  )

inherited

Get information about the image from the filename and its contents

Here, we open the image and parse the header, but one could also look at the filename itself
and derive information from that, or set values from the configuration.

Parameters
----------
filename : `str`
    Name of file to inspect

Returns
-------
phuInfo : `dict`
    File properties
infoList : `list`
    List of file properties for each extension

Reimplemented in lsst.obs.decam.ingestCalibs.DecamCalibsParseTask.

Definition at line 80 of file ingest.py.

    def getInfo(self, filename):
        """Get information about the image from the filename and its contents
 
        Here, we open the image and parse the header, but one could also look at the filename itself
        and derive information from that, or set values from the configuration.
 
        Parameters
        ----------
        filename : `str`
            Name of file to inspect
 
        Returns
        -------
        phuInfo : `dict`
            File properties
        infoList : `list`
            List of file properties for each extension
        """
        md = readMetadata(filename, self.config.hdu)
        fix_header(md, translator_class=self.translator_class)
        phuInfo = self.getInfoFromMetadata(md)
        if len(self.config.extnames) == 0:
            # No extensions to worry about
            return phuInfo, [phuInfo]
        # Look in the provided extensions
        extnames = set(self.config.extnames)
        extnum = 0
        infoList = []
        while len(extnames) > 0:
            extnum += 1
            try:
                md = readMetadata(filename, extnum)
                fix_header(md, translator_class=self.translator_class)
            except Exception as e:
                self.log.warning("Error reading %s extensions %s: %s", filename, extnames, e)
                break
            ext = self.getExtensionName(md)
            if ext in extnames:
                hduInfo = self.getInfoFromMetadata(md, info=phuInfo.copy())
                # We need the HDU number when registering MEF files.
                hduInfo["hdu"] = extnum
                infoList.append(hduInfo)
                extnames.discard(ext)
        return phuInfo, infoList
 

getInfoFromMetadata()

def lsst.pipe.tasks.ingest.ParseTask.getInfoFromMetadata (   self,   md,   info = None  )

inherited

Attempt to pull the desired information out of the header

This is done through two mechanisms:
* translation: a property is set directly from the relevant header keyword
* translator: a property is set with the result of calling a method

The translator methods receive the header metadata and should return the
appropriate value, or None if the value cannot be determined.

@param md      FITS header
@param info    File properties, to be supplemented
@return info

Definition at line 145 of file ingest.py.

    def getInfoFromMetadata(self, md, info=None):
        """Attempt to pull the desired information out of the header
 
        This is done through two mechanisms:
        * translation: a property is set directly from the relevant header keyword
        * translator: a property is set with the result of calling a method
 
        The translator methods receive the header metadata and should return the
        appropriate value, or None if the value cannot be determined.
 
        @param md      FITS header
        @param info    File properties, to be supplemented
        @return info
        """
        if info is None:
            info = {}
        for p, h in self.config.translation.items():
            value = md.get(h, None)
            if value is not None:
                if isinstance(value, str):
                    value = value.strip()
                info[p] = value
            elif p in self.config.defaults:
                info[p] = self.config.defaults[p]
            else:
                self.log.warning("Unable to find value for %s (derived from %s)", p, h)
        for p, t in self.config.translators.items():
            func = getattr(self, t)
            try:
                value = func(md)
            except Exception as e:
                self.log.warning("%s failed to translate %s: %s", t, p, e)
                value = None
            if value is not None:
                info[p] = value
        return info
 

translate_date()

def lsst.pipe.tasks.ingest.ParseTask.translate_date (   self,   md  )

inherited

Convert a full DATE-OBS to a mere date

Besides being an example of a translator, this is also generally useful.
It will only be used if listed as a translator in the configuration.

Reimplemented in lsst.obs.decam.ingestCalibs.DecamCalibsParseTask.

Definition at line 182 of file ingest.py.

    def translate_date(self, md):
        """Convert a full DATE-OBS to a mere date
 
        Besides being an example of a translator, this is also generally useful.
        It will only be used if listed as a translator in the configuration.
        """
        date = md.getScalar("DATE-OBS").strip()
        c = date.find("T")
        if c > 0:
            date = date[:c]
        return date
 

translate_filter()

def lsst.pipe.tasks.ingest.ParseTask.translate_filter (   self,   md  )

inherited

Translate a full filter description into a mere filter name

Besides being an example of a translator, this is also generally useful.
It will only be used if listed as a translator in the configuration.

Reimplemented in lsst.obs.decam.ingestCalibs.DecamCalibsParseTask.

Definition at line 194 of file ingest.py.

    def translate_filter(self, md):
        """Translate a full filter description into a mere filter name
 
        Besides being an example of a translator, this is also generally useful.
        It will only be used if listed as a translator in the configuration.
        """
        filterName = md.getScalar("FILTER").strip()
        filterName = filterName.strip()
        c = filterName.find(" ")
        if c > 0:
            filterName = filterName[:c]
        return filterName
 

Member Data Documentation

ConfigClass

lsst.pipe.tasks.ingest.ParseTask.ConfigClass = ParseConfig

staticinherited

Definition at line 70 of file ingest.py.

translator_class

lsst.pipe.tasks.ingest.ParseTask.translator_class = None

staticinherited

Definition at line 71 of file ingest.py.

---

The documentation for this class was generated from the following file:

• /j/snowflake/release/lsstsw/stack/lsst-scipipe-0.7.0/Linux64/pipe_tasks/21.0.0-172-gfb10e10a+18fedfabac/python/lsst/pipe/tasks/ingestCalibs.py