Inheritance diagram for lsst.pipe.tasks.functors.CompositeFunctor:

Public Member Functions
	__init__ (self, funcs, **kwargs)

	filt (self)

	filt (self, filt)

	update (self, new)

	columns (self)

	multilevelColumns (self, data, **kwargs)

	__call__ (self, data, **kwargs)

	renameCol (cls, col, renameRules)

	from_file (cls, filename, **kwargs)

	from_yaml (cls, translationDefinition, **kwargs)

	name (self)

	noDup (self)

	difference (self, data1, data2, **kwargs)

	fail (self, df)

	shortname (self)

Public Attributes
dict	funcDict = funcs

str	dataset = dataset if dataset is not None else self._defaultDataset

	log = logging.getLogger(type(self).__name__)

	name

Protected Member Functions
	_get_data_columnLevels (self, data, columnIndex=None)

	_get_data_columnLevelNames (self, data, columnIndex=None)

	_colsFromDict (self, colDict, columnIndex=None)

	_func (self, df, dropna=True)

	_get_columnIndex (self, data)

	_get_data (self, data)

	_setLevels (self, df)

	_dropna (self, vals)

Protected Attributes
	_filt = None

	_noDup = noDup

Static Protected Attributes
str	_defaultDataset = 'ref'

tuple	_dfLevels = ('column',)

bool	_defaultNoDup = False

Detailed Description

Perform multiple calculations at once on a catalog.

The role of a `CompositeFunctor` is to group together computations from
multiple functors.
Instead of returning `~pandas.Series` a `CompositeFunctor` returns a
`~pandas.DataFrame`, with the column names being the keys of ``funcDict``.

The `columns` attribute of a `CompositeFunctor` is the union of all columns
in all the component functors.

A `CompositeFunctor` does not use a `_func` method itself; rather, when a
`CompositeFunctor` is called, all its columns are loaded at once, and the
resulting DataFrame is passed to the `_func` method of each component
functor.
This has the advantage of only doing I/O (reading from parquet file) once,
and works because each individual `_func` method of each component functor
does not care if there are *extra* columns in the DataFrame being passed;
only that it must contain *at least* the `columns` it expects.

An important and useful class method is `from_yaml`, which takes as an
argument the path to a YAML file specifying a collection of functors.

Parameters
----------
funcs : `dict` or `list`
    Dictionary or list of functors.
    If a list, then it will be converted into a dictonary according to the
    `.shortname` attribute of each functor.

Definition at line 380 of file functors.py.

Constructor & Destructor Documentation

◆ init()

lsst.pipe.tasks.functors.CompositeFunctor.__init__	(		self,
			funcs,
		**	kwargs )

Definition at line 413 of file functors.py.

    def __init__(self, funcs, **kwargs):
 
        if type(funcs) is dict:
            self.funcDict = funcs
        else:
            self.funcDict = {f.shortname: f for f in funcs}
 
        self._filt = None
 
        super().__init__(**kwargs)
 

Member Function Documentation

◆ call()

lsst.pipe.tasks.functors.CompositeFunctor.__call__	(		self,
			data,
		**	kwargs )

Apply the functor to the data table.

Parameters
----------
data : various
    The data represented as `~lsst.daf.butler.DeferredDatasetHandle`,
    `~lsst.pipe.base.InMemoryDatasetHandle`, or `~pandas.DataFrame`.
    The table or a pointer to a table on disk from which columns can
    be accessed.

Definition at line 467 of file functors.py.

    def __call__(self, data, **kwargs):
        """Apply the functor to the data table.
 
        Parameters
        ----------
        data : various
            The data represented as `~lsst.daf.butler.DeferredDatasetHandle`,
            `~lsst.pipe.base.InMemoryDatasetHandle`, or `~pandas.DataFrame`.
            The table or a pointer to a table on disk from which columns can
            be accessed.
        """
        if isinstance(data, pd.DataFrame):
            _data = InMemoryDatasetHandle(data, storageClass="DataFrame")
        elif isinstance(data, (DeferredDatasetHandle, InMemoryDatasetHandle)):
            _data = data
        else:
            raise RuntimeError(f"Unexpected type provided for data. Got {get_full_type_name(data)}.")
 
        columnIndex = self._get_columnIndex(_data)
 
        if isinstance(columnIndex, pd.MultiIndex):
            columns = self.multilevelColumns(_data, columnIndex=columnIndex)
            df = _data.get(parameters={"columns": columns})
 
            valDict = {}
            for k, f in self.funcDict.items():
                try:
                    subdf = f._setLevels(
                        df[f.multilevelColumns(_data, returnTuple=True, columnIndex=columnIndex)]
                    )
                    valDict[k] = f._func(subdf)
                except Exception as e:
                    self.log.exception(
                        "Exception in %s (funcs: %s) call: %s",
                        self.name,
                        str(list(self.funcDict.keys())),
                        type(e).__name__,
                    )
                    try:
                        valDict[k] = f.fail(subdf)
                    except NameError:
                        raise e
 
        else:
            df = _data.get(parameters={"columns": self.columns})
 
            valDict = {k: f._func(df) for k, f in self.funcDict.items()}
 
            # Check that output columns are actually columns.
            for name, colVal in valDict.items():
                if len(colVal.shape) != 1:
                    raise RuntimeError("Transformed column '%s' is not the shape of a column. "
                                       "It is shaped %s and type %s." % (name, colVal.shape, type(colVal)))
 
        try:
            valDf = pd.concat(valDict, axis=1)
        except TypeError:
            print([(k, type(v)) for k, v in valDict.items()])
            raise
 
        if kwargs.get('dropna', False):
            valDf = valDf.dropna(how='any')
 
        return valDf
 

◆ _colsFromDict()

lsst.pipe.tasks.functors.Functor._colsFromDict	(	self,
		colDict,
		columnIndex = None )

protectedinherited

Converts dictionary column specficiation to a list of columns.

Definition at line 218 of file functors.py.

    def _colsFromDict(self, colDict, columnIndex=None):
        """Converts dictionary column specficiation to a list of columns."""
        new_colDict = {}
        columnLevels = self._get_data_columnLevels(None, columnIndex=columnIndex)
 
        for i, lev in enumerate(columnLevels):
            if lev in colDict:
                if isinstance(colDict[lev], str):
                    new_colDict[lev] = [colDict[lev]]
                else:
                    new_colDict[lev] = colDict[lev]
            else:
                new_colDict[lev] = columnIndex.levels[i]
 
        levelCols = [new_colDict[lev] for lev in columnLevels]
        cols = list(product(*levelCols))
        colsAvailable = [col for col in cols if col in columnIndex]
        return colsAvailable
 

◆ _dropna()

lsst.pipe.tasks.functors.Functor._dropna	(		self,
			vals )

protectedinherited

Definition at line 345 of file functors.py.

    def _dropna(self, vals):
        return vals.dropna()
 

◆ _func()

lsst.pipe.tasks.functors.Functor._func	(	self,
		df,
		dropna = True )

protectedinherited

Definition at line 291 of file functors.py.

    def _func(self, df, dropna=True):
        raise NotImplementedError('Must define calculation on DataFrame')
 

◆ _get_columnIndex()

lsst.pipe.tasks.functors.Functor._get_columnIndex	(		self,
			data )

protectedinherited

Return columnIndex.

Definition at line 294 of file functors.py.

    def _get_columnIndex(self, data):
        """Return columnIndex."""
 
        if isinstance(data, (DeferredDatasetHandle, InMemoryDatasetHandle)):
            return data.get(component="columns")
        else:
            return None
 

◆ _get_data()

lsst.pipe.tasks.functors.Functor._get_data	(		self,
			data )

protectedinherited

Retrieve DataFrame necessary for calculation.

The data argument can be a `~pandas.DataFrame`, a
`~lsst.daf.butler.DeferredDatasetHandle`, or
an `~lsst.pipe.base.InMemoryDatasetHandle`.

Returns a DataFrame upon which `self._func` can act.

Definition at line 302 of file functors.py.

    def _get_data(self, data):
        """Retrieve DataFrame necessary for calculation.
 
        The data argument can be a `~pandas.DataFrame`, a
        `~lsst.daf.butler.DeferredDatasetHandle`, or
        an `~lsst.pipe.base.InMemoryDatasetHandle`.
 
        Returns a DataFrame upon which `self._func` can act.
        """
        # We wrap a DataFrame in a handle here to take advantage of the
        # DataFrame delegate DataFrame column wrangling abilities.
        if isinstance(data, pd.DataFrame):
            _data = InMemoryDatasetHandle(data, storageClass="DataFrame")
        elif isinstance(data, (DeferredDatasetHandle, InMemoryDatasetHandle)):
            _data = data
        else:
            raise RuntimeError(f"Unexpected type provided for data. Got {get_full_type_name(data)}.")
 
        # First thing to do: check to see if the data source has a multilevel
        # column index or not.
        columnIndex = self._get_columnIndex(_data)
        is_multiLevel = isinstance(columnIndex, pd.MultiIndex)
 
        # Get proper columns specification for this functor.
        if is_multiLevel:
            columns = self.multilevelColumns(_data, columnIndex=columnIndex)
        else:
            columns = self.columns
 
        # Load in-memory DataFrame with appropriate columns the gen3 way.
        df = _data.get(parameters={"columns": columns})
 
        # Drop unnecessary column levels.
        if is_multiLevel:
            df = self._setLevels(df)
 
        return df
 

◆ _get_data_columnLevelNames()

lsst.pipe.tasks.functors.Functor._get_data_columnLevelNames	(	self,
		data,
		columnIndex = None )

protectedinherited

Gets the content of each of the column levels for a multilevel
table.

Definition at line 204 of file functors.py.

    def _get_data_columnLevelNames(self, data, columnIndex=None):
        """Gets the content of each of the column levels for a multilevel
        table.
        """
        if columnIndex is None:
            columnIndex = data.get(component="columns")
 
        columnLevels = columnIndex.names
        columnLevelNames = {
            level: list(np.unique(np.array([c for c in columnIndex])[:, i]))
            for i, level in enumerate(columnLevels)
        }
        return columnLevelNames
 

◆ _get_data_columnLevels()

lsst.pipe.tasks.functors.Functor._get_data_columnLevels	(	self,
		data,
		columnIndex = None )

protectedinherited

Gets the names of the column index levels.

This should only be called in the context of a multilevel table.

Parameters
----------
data : various
    The data to be read, can be a
    `~lsst.daf.butler.DeferredDatasetHandle` or
    `~lsst.pipe.base.InMemoryDatasetHandle`.
columnIndex (optional): pandas `~pandas.Index` object
    If not passed, then it is read from the
    `~lsst.daf.butler.DeferredDatasetHandle`
    for `~lsst.pipe.base.InMemoryDatasetHandle`.

Definition at line 184 of file functors.py.

    def _get_data_columnLevels(self, data, columnIndex=None):
        """Gets the names of the column index levels.
 
        This should only be called in the context of a multilevel table.
 
        Parameters
        ----------
        data : various
            The data to be read, can be a
            `~lsst.daf.butler.DeferredDatasetHandle` or
            `~lsst.pipe.base.InMemoryDatasetHandle`.
        columnIndex (optional): pandas `~pandas.Index` object
            If not passed, then it is read from the
            `~lsst.daf.butler.DeferredDatasetHandle`
            for `~lsst.pipe.base.InMemoryDatasetHandle`.
        """
        if columnIndex is None:
            columnIndex = data.get(component="columns")
        return columnIndex.names
 

◆ _setLevels()

lsst.pipe.tasks.functors.Functor._setLevels	(		self,
			df )

protectedinherited

Definition at line 340 of file functors.py.

    def _setLevels(self, df):
        levelsToDrop = [n for n in df.columns.names if n not in self._dfLevels]
        df.columns = df.columns.droplevel(levelsToDrop)
        return df
 

◆ columns()

lsst.pipe.tasks.functors.CompositeFunctor.columns ( self )

Columns required to perform calculation.

Reimplemented from lsst.pipe.tasks.functors.Functor.

Definition at line 449 of file functors.py.

    def columns(self):
        return list(set([x for y in [f.columns for f in self.funcDict.values()] for x in y]))
 

◆ difference()

lsst.pipe.tasks.functors.Functor.difference	(		self,
			data1,
			data2,
		**	kwargs )

inherited

Computes difference between functor called on two different
DataFrame/Handle objects.

Definition at line 360 of file functors.py.

    def difference(self, data1, data2, **kwargs):
        """Computes difference between functor called on two different
        DataFrame/Handle objects.
        """
        return self(data1, **kwargs) - self(data2, **kwargs)
 

◆ fail()

lsst.pipe.tasks.functors.Functor.fail	(		self,
			df )

inherited

Definition at line 366 of file functors.py.

    def fail(self, df):
        return pd.Series(np.full(len(df), np.nan), index=df.index)
 

◆ filt() [1/2]

lsst.pipe.tasks.functors.CompositeFunctor.filt ( self )

Reimplemented from lsst.pipe.tasks.functors.Functor.

Definition at line 425 of file functors.py.

    def filt(self):
        return self._filt
 

◆ filt() [2/2]

lsst.pipe.tasks.functors.CompositeFunctor.filt	(		self,
			filt )

Reimplemented from lsst.pipe.tasks.functors.Functor.

Definition at line 429 of file functors.py.

    def filt(self, filt):
        if filt is not None:
            for _, f in self.funcDict.items():
                f.filt = filt
        self._filt = filt
 

◆ from_file()

lsst.pipe.tasks.functors.CompositeFunctor.from_file	(		cls,
			filename,
		**	kwargs )

Definition at line 542 of file functors.py.

    def from_file(cls, filename, **kwargs):
        # Allow environment variables in the filename.
        filename = os.path.expandvars(filename)
        with open(filename) as f:
            translationDefinition = yaml.safe_load(f)
 
        return cls.from_yaml(translationDefinition, **kwargs)
 

◆ from_yaml()

lsst.pipe.tasks.functors.CompositeFunctor.from_yaml	(		cls,
			translationDefinition,
		**	kwargs )

Definition at line 551 of file functors.py.

    def from_yaml(cls, translationDefinition, **kwargs):
        funcs = {}
        for func, val in translationDefinition['funcs'].items():
            funcs[func] = init_fromDict(val, name=func)
 
        if 'flag_rename_rules' in translationDefinition:
            renameRules = translationDefinition['flag_rename_rules']
        else:
            renameRules = None
 
        if 'calexpFlags' in translationDefinition:
            for flag in translationDefinition['calexpFlags']:
                funcs[cls.renameCol(flag, renameRules)] = Column(flag, dataset='calexp')
 
        if 'refFlags' in translationDefinition:
            for flag in translationDefinition['refFlags']:
                funcs[cls.renameCol(flag, renameRules)] = Column(flag, dataset='ref')
 
        if 'forcedFlags' in translationDefinition:
            for flag in translationDefinition['forcedFlags']:
                funcs[cls.renameCol(flag, renameRules)] = Column(flag, dataset='forced_src')
 
        if 'flags' in translationDefinition:
            for flag in translationDefinition['flags']:
                funcs[cls.renameCol(flag, renameRules)] = Column(flag, dataset='meas')
 
        return cls(funcs, **kwargs)
 
 

◆ multilevelColumns()

lsst.pipe.tasks.functors.CompositeFunctor.multilevelColumns	(		self,
			data,
		**	columnIndex )

Returns columns needed by functor from multilevel dataset.

To access tables with multilevel column structure, the
`~lsst.daf.butler.DeferredDatasetHandle` or
`~lsst.pipe.base.InMemoryDatasetHandle` needs to be passed
either a list of tuples or a dictionary.

Parameters
----------
data : various
    The data as either `~lsst.daf.butler.DeferredDatasetHandle`, or
    `~lsst.pipe.base.InMemoryDatasetHandle`.
columnIndex (optional): pandas `~pandas.Index` object
    Either passed or read in from
    `~lsst.daf.butler.DeferredDatasetHandle`.
`returnTuple` : `bool`
    If true, then return a list of tuples rather than the column
    dictionary specification.
    This is set to `True` by `CompositeFunctor` in order to be able to
    combine columns from the various component functors.

Reimplemented from lsst.pipe.tasks.functors.Functor.

Definition at line 452 of file functors.py.

    def multilevelColumns(self, data, **kwargs):
        # Get the union of columns for all component functors.
        # Note the need to have `returnTuple=True` here.
        return list(
            set(
                [
                    x
                    for y in [
                        f.multilevelColumns(data, returnTuple=True, **kwargs) for f in self.funcDict.values()
                    ]
                    for x in y
                ]
            )
        )
 

◆ name()

lsst.pipe.tasks.functors.Functor.name ( self )

inherited

Full name of functor (suitable for figure labels).

Definition at line 370 of file functors.py.

    def name(self):
        """Full name of functor (suitable for figure labels)."""
        return NotImplementedError
 

◆ noDup()

lsst.pipe.tasks.functors.Functor.noDup ( self )

inherited

Do not explode by band if used on object table.

Definition at line 170 of file functors.py.

    def noDup(self):
        """Do not explode by band if used on object table."""
        if self._noDup is not None:
            return self._noDup
        else:
            return self._defaultNoDup
 

◆ renameCol()

lsst.pipe.tasks.functors.CompositeFunctor.renameCol	(	cls,
		col,
		renameRules )

Definition at line 533 of file functors.py.

    def renameCol(cls, col, renameRules):
        if renameRules is None:
            return col
        for old, new in renameRules:
            if col.startswith(old):
                col = col.replace(old, new)
        return col
 

◆ shortname()

lsst.pipe.tasks.functors.Functor.shortname ( self )

inherited

Short name of functor (suitable for column name/dict key).

Reimplemented in lsst.pipe.tasks.functors.Color, and lsst.pipe.tasks.functors.MagDiff.

Definition at line 375 of file functors.py.

    def shortname(self):
        """Short name of functor (suitable for column name/dict key)."""
        return self.name
 
 

◆ update()

lsst.pipe.tasks.functors.CompositeFunctor.update	(		self,
			new )

Update the functor with new functors.

Definition at line 435 of file functors.py.

    def update(self, new):
        """Update the functor with new functors."""
        if isinstance(new, dict):
            self.funcDict.update(new)
        elif isinstance(new, CompositeFunctor):
            self.funcDict.update(new.funcDict)
        else:
            raise TypeError('Can only update with dictionary or CompositeFunctor.')
 
        # Make sure new functors have the same 'filt' set.
        if self.filt is not None:
            self.filt = self.filt
 

Member Data Documentation

◆ _defaultDataset

str lsst.pipe.tasks.functors.Functor._defaultDataset = 'ref'

staticprotectedinherited

Definition at line 159 of file functors.py.

◆ _defaultNoDup

bool lsst.pipe.tasks.functors.Functor._defaultNoDup = False

staticprotectedinherited

Definition at line 161 of file functors.py.

◆ _dfLevels

tuple lsst.pipe.tasks.functors.Functor._dfLevels = ('column',)

staticprotectedinherited

Definition at line 160 of file functors.py.

◆ _filt

lsst.pipe.tasks.functors.CompositeFunctor._filt = None

protected

Definition at line 420 of file functors.py.

◆ _noDup

lsst.pipe.tasks.functors.Functor._noDup = noDup

protectedinherited

Definition at line 166 of file functors.py.

◆ dataset

str lsst.pipe.tasks.functors.Functor.dataset = dataset if dataset is not None else self._defaultDataset

inherited

Definition at line 165 of file functors.py.

◆ funcDict

dict lsst.pipe.tasks.functors.CompositeFunctor.funcDict = funcs

Definition at line 416 of file functors.py.

◆ log

lsst.pipe.tasks.functors.Functor.log = logging.getLogger(type(self).__name__)

inherited

Definition at line 167 of file functors.py.

◆ name

lsst.pipe.tasks.functors.Functor.name

inherited

Definition at line 353 of file functors.py.

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

/j/snowflake/release/lsstsw/stack/lsst-scipipe-12.1.0/Linux/pipe_tasks/g99af87f6a8+b0f4ad7b8d/python/lsst/pipe/tasks/functors.py

Public Member Functions

Public Attributes

Protected Member Functions

Protected Attributes

Static Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

Member Function Documentation

◆ __call__()

◆ _colsFromDict()

◆ _dropna()

◆ _func()

◆ _get_columnIndex()

◆ _get_data()

◆ _get_data_columnLevelNames()

◆ _get_data_columnLevels()

◆ _setLevels()

◆ columns()

◆ difference()

◆ fail()

◆ filt() [1/2]

◆ filt() [2/2]

◆ from_file()

◆ from_yaml()

◆ multilevelColumns()

◆ name()

◆ noDup()

◆ renameCol()

◆ shortname()

◆ update()

Member Data Documentation

◆ _defaultDataset

◆ _defaultNoDup

◆ _dfLevels

◆ _filt

◆ _noDup

◆ dataset

◆ funcDict

◆ log

◆ name

◆ init()

◆ call()