lsst.pipe.tasks.ingestCalibs.CalibsRegisterTask Class Reference

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

Public Member Functions
def	openRegistry (self, directory, create=False, dryrun=False, name="calibRegistry.sqlite3")
def	createTable (self, conn, forceCreateTables=False)
def	addRow (self, conn, info, *args, **kwargs)
def	updateValidityRanges (self, conn, validity, tables=None)
def	fixSubsetValidity (self, conn, table, detectorData, validity)
def	createTable (self, conn, table=None, forceCreateTables=False)
def	check (self, conn, info, table=None)
def	addRow (self, conn, info, dryrun=False, create=False, table=None)

Static Public Attributes
	ConfigClass = CalibsRegisterConfig
string	placeHolder = '?'
dictionary	typemap = {'text': str, 'int': int, 'double': float}

Detailed Description

Task that will generate the calibration registry for the Mapper

Definition at line 102 of file ingestCalibs.py.

Member Function Documentation

addRow() [1/2]

def lsst.pipe.tasks.ingestCalibs.CalibsRegisterTask.addRow	(		self,
			conn,
			info,
		*	args,
		**	kwargs
	)

Add a row to the file table

Definition at line 115 of file ingestCalibs.py.

    def addRow(self, conn, info, *args, **kwargs):
        """Add a row to the file table"""
        info[self.config.validStart] = None
        info[self.config.validEnd] = None
        RegisterTask.addRow(self, conn, info, *args, **kwargs)
 

addRow() [2/2]

def lsst.pipe.tasks.ingest.RegisterTask.addRow (   self,   conn,   info,   dryrun = False,   create = False,   table = None  )

inherited

Add a row to the file table (typically 'raw').

@param conn    Database connection
@param info    File properties to add to database
@param table   Name of table in database

Definition at line 359 of file ingest.py.

    def addRow(self, conn, info, dryrun=False, create=False, table=None):
        """Add a row to the file table (typically 'raw').
 
        @param conn    Database connection
        @param info    File properties to add to database
        @param table   Name of table in database
        """
        with conn:
            if table is None:
                table = self.config.table
            ignoreClause = ""
            if self.config.ignore:
                ignoreClause = " OR IGNORE"
            sql = "INSERT%s INTO %s (%s) VALUES (" % (ignoreClause, table, ",".join(self.config.columns))
            sql += ",".join([self.placeHolder] * len(self.config.columns)) + ")"
            values = [self.typemap[tt](info[col]) for col, tt in self.config.columns.items()]
 
            if dryrun:
                print("Would execute: '%s' with %s" % (sql, ",".join([str(value) for value in values])))
            else:
                conn.cursor().execute(sql, values)
 
            sql = "INSERT OR IGNORE INTO %s_visit VALUES (" % table
            sql += ",".join([self.placeHolder] * len(self.config.visit)) + ")"
            values = [self.typemap[self.config.columns[col]](info[col]) for col in self.config.visit]
 
            if dryrun:
                print("Would execute: '%s' with %s" % (sql, ",".join([str(value) for value in values])))
            else:
                conn.cursor().execute(sql, values)
 
 

check()

def lsst.pipe.tasks.ingest.RegisterTask.check (   self,   conn,   info,   table = None  )

inherited

Check for the presence of a row already

Not sure this is required, given the 'ignore' configuration option.

Definition at line 340 of file ingest.py.

    def check(self, conn, info, table=None):
        """Check for the presence of a row already
 
        Not sure this is required, given the 'ignore' configuration option.
        """
        if table is None:
            table = self.config.table
        if self.config.ignore or len(self.config.unique) == 0:
            return False  # Our entry could already be there, but we don't care
        cursor = conn.cursor()
        sql = "SELECT COUNT(*) FROM %s WHERE " % table
        sql += " AND ".join(["%s = %s" % (col, self.placeHolder) for col in self.config.unique])
        values = [self.typemap[self.config.columns[col]](info[col]) for col in self.config.unique]
 
        cursor.execute(sql, values)
        if cursor.fetchone()[0] > 0:
            return True
        return False
 

createTable() [1/2]

def lsst.pipe.tasks.ingestCalibs.CalibsRegisterTask.createTable	(		self,
			conn,
			forceCreateTables = False
	)

Create the registry tables

Definition at line 110 of file ingestCalibs.py.

    def createTable(self, conn, forceCreateTables=False):
        """Create the registry tables"""
        for table in self.config.tables:
            RegisterTask.createTable(self, conn, table=table, forceCreateTables=forceCreateTables)
 

createTable() [2/2]

def lsst.pipe.tasks.ingest.RegisterTask.createTable (   self,   conn,   table = None,   forceCreateTables = False  )

inherited

Create the registry tables

One table (typically 'raw') contains information on all files, and the
other (typically 'raw_visit') contains information on all visits.

@param conn    Database connection
@param table   Name of table to create in database

Definition at line 302 of file ingest.py.

    def createTable(self, conn, table=None, forceCreateTables=False):
        """Create the registry tables
 
        One table (typically 'raw') contains information on all files, and the
        other (typically 'raw_visit') contains information on all visits.
 
        @param conn    Database connection
        @param table   Name of table to create in database
        """
        cursor = conn.cursor()
        if table is None:
            table = self.config.table
        cmd = "SELECT name FROM sqlite_master WHERE type='table' AND name='%s'" % table
        cursor.execute(cmd)
        if cursor.fetchone() and not forceCreateTables:  # Assume if we get an answer the table exists
            self.log.info('Table "%s" exists.  Skipping creation', table)
            return
        else:
            cmd = "drop table if exists %s" % table
            cursor.execute(cmd)
            cmd = "drop table if exists %s_visit" % table
            cursor.execute(cmd)
 
        cmd = "create table %s (id integer primary key autoincrement, " % table
        cmd += ",".join([("%s %s" % (col, colType)) for col, colType in self.config.columns.items()])
        if len(self.config.unique) > 0:
            cmd += ", unique(" + ",".join(self.config.unique) + ")"
        cmd += ")"
        cursor.execute(cmd)
 
        cmd = "create table %s_visit (" % table
        cmd += ",".join([("%s %s" % (col, self.config.columns[col])) for col in self.config.visit])
        cmd += ", unique(" + ",".join(set(self.config.visit).intersection(set(self.config.unique))) + ")"
        cmd += ")"
        cursor.execute(cmd)
 
        conn.commit()
 

fixSubsetValidity()

def lsst.pipe.tasks.ingestCalibs.CalibsRegisterTask.fixSubsetValidity	(		self,
			conn,
			table,
			detectorData,
			validity
	)

Update the validity ranges among selected rows in the registry.

For defects and qe_curve, the products are valid from their start date until
they are superseded by subsequent defect data.
For other calibration products, the validity ranges are checked and
if there are overlaps, a midpoint is used to fix the overlaps,
so that the calibration data with whose date is nearest the date
of the observation is used.

DM generated calibrations contain a CALIB_ID header
keyword. These calibrations likely require the
incrementValidEnd configuration option set to True.  Other
calibrations generate the calibDate via the DATE-OBS header
keyword, and likely require incrementValidEnd=False.

@param conn: Database connection
@param table: Name of table to be selected
@param detectorData: Values identifying a detector (from columns in self.config.detector)
@param validity: Validity range (days)

Definition at line 139 of file ingestCalibs.py.

    def fixSubsetValidity(self, conn, table, detectorData, validity):
        """Update the validity ranges among selected rows in the registry.
 
        For defects and qe_curve, the products are valid from their start date until
        they are superseded by subsequent defect data.
        For other calibration products, the validity ranges are checked and
        if there are overlaps, a midpoint is used to fix the overlaps,
        so that the calibration data with whose date is nearest the date
        of the observation is used.
 
        DM generated calibrations contain a CALIB_ID header
        keyword. These calibrations likely require the
        incrementValidEnd configuration option set to True.  Other
        calibrations generate the calibDate via the DATE-OBS header
        keyword, and likely require incrementValidEnd=False.
 
        @param conn: Database connection
        @param table: Name of table to be selected
        @param detectorData: Values identifying a detector (from columns in self.config.detector)
        @param validity: Validity range (days)
        """
        columns = ", ".join([self.config.calibDate, self.config.validStart, self.config.validEnd])
        sql = "SELECT id, %s FROM %s" % (columns, table)
        sql += " WHERE " + " AND ".join(col + "=?" for col in self.config.detector)
        sql += " ORDER BY " + self.config.calibDate
        cursor = conn.cursor()
        cursor.execute(sql, detectorData)
        rows = cursor.fetchall()
 
        try:
            valids = collections.OrderedDict([(_convertToDate(row[self.config.calibDate]), [None, None]) for
                                              row in rows])
        except Exception:
            det = " ".join("%s=%s" % (k, v) for k, v in zip(self.config.detector, detectorData))
            self.log.warning("Skipped setting the validity overlaps for %s %s: missing calibration dates",
                             table, det)
            return
        dates = list(valids.keys())
        if table in self.config.validityUntilSuperseded:
            # A calib is valid until it is superseded
            for thisDate, nextDate in zip(dates[:-1], dates[1:]):
                valids[thisDate][0] = thisDate
                valids[thisDate][1] = nextDate
            valids[dates[-1]][0] = dates[-1]
            valids[dates[-1]][1] = _convertToDate("2037-12-31")  # End of UNIX time
        else:
            # A calib is valid within the validity range (in days) specified.
            for dd in dates:
                valids[dd] = [dd - datetime.timedelta(validity), dd + datetime.timedelta(validity)]
            # Fix the dates so that they do not overlap, which can cause the butler to find a
            # non-unique calib.
            midpoints = [t1 + (t2 - t1)//2 for t1, t2 in zip(dates[:-1], dates[1:])]
            for i, (date, midpoint) in enumerate(zip(dates[:-1], midpoints)):
                if valids[date][1] > midpoint:
                    nextDate = dates[i + 1]
                    valids[nextDate][0] = midpoint + datetime.timedelta(1)
                    if self.config.incrementValidEnd:
                        valids[date][1] = midpoint + datetime.timedelta(1)
                    else:
                        valids[date][1] = midpoint
            del midpoints
        del dates
        # Update the validity data in the registry
        for row in rows:
            calibDate = _convertToDate(row[self.config.calibDate])
            validStart = valids[calibDate][0].isoformat()
            validEnd = valids[calibDate][1].isoformat()
            sql = "UPDATE %s" % table
            sql += " SET %s=?, %s=?" % (self.config.validStart, self.config.validEnd)
            sql += " WHERE id=?"
            conn.execute(sql, (validStart, validEnd, row["id"]))
 
 

openRegistry()

def lsst.pipe.tasks.ingestCalibs.CalibsRegisterTask.openRegistry	(		self,
			directory,
			create = False,
			dryrun = False,
			name = "calibRegistry.sqlite3"
	)

Open the registry and return the connection handle

Reimplemented from lsst.pipe.tasks.ingest.RegisterTask.

Definition at line 106 of file ingestCalibs.py.

    def openRegistry(self, directory, create=False, dryrun=False, name="calibRegistry.sqlite3"):
        """Open the registry and return the connection handle"""
        return RegisterTask.openRegistry(self, directory, create, dryrun, name)
 

updateValidityRanges()

def lsst.pipe.tasks.ingestCalibs.CalibsRegisterTask.updateValidityRanges	(		self,
			conn,
			validity,
			tables = None
	)

Loop over all tables, filters, and ccdnums,
and update the validity ranges in the registry.

@param conn: Database connection
@param validity: Validity range (days)

Definition at line 121 of file ingestCalibs.py.

    def updateValidityRanges(self, conn, validity, tables=None):
        """Loop over all tables, filters, and ccdnums,
        and update the validity ranges in the registry.
 
        @param conn: Database connection
        @param validity: Validity range (days)
        """
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        if tables is None:
            tables = self.config.tables
        for table in tables:
            sql = "SELECT DISTINCT %s FROM %s" % (", ".join(self.config.detector), table)
            cursor.execute(sql)
            rows = cursor.fetchall()
            for row in rows:
                self.fixSubsetValidity(conn, table, row, validity)
 

Member Data Documentation

ConfigClass

lsst.pipe.tasks.ingestCalibs.CalibsRegisterTask.ConfigClass = CalibsRegisterConfig

static

Definition at line 104 of file ingestCalibs.py.

placeHolder

string lsst.pipe.tasks.ingest.RegisterTask.placeHolder = '?'

staticinherited

Definition at line 283 of file ingest.py.

typemap

dictionary lsst.pipe.tasks.ingest.RegisterTask.typemap = {'text': str, 'int': int, 'double': float}

staticinherited

Definition at line 284 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