lsst.pipe.tasks.functors.CompositeFunctor Class Reference

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)

Public Attributes
	funcDict
	filt
	name

Static Public Attributes
	dataset = None
str	name = "CompositeFunctor"

Protected Attributes
	_filt

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 377 of file functors.py.

Constructor & Destructor Documentation

__init__()

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

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

Definition at line 410 of file functors.py.

    def __init__(self, funcs, **kwargs):
 
        if type(funcs) == 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.

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

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

columns()

lsst.pipe.tasks.functors.CompositeFunctor.columns

(

self

)

Columns required to perform calculation.

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

Definition at line 446 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]))
 

filt() [1/2]

lsst.pipe.tasks.functors.CompositeFunctor.filt

(

self

)

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

Definition at line 422 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 426 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 539 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 548 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 449 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
                ]
            )
        )
 

renameCol()

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

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

update()

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

Update the functor with new functors.

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

_filt

lsst.pipe.tasks.functors.CompositeFunctor._filt

protected

Definition at line 417 of file functors.py.

dataset

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

static

Definition at line 407 of file functors.py.

filt

lsst.pipe.tasks.functors.CompositeFunctor.filt

Definition at line 443 of file functors.py.

funcDict

lsst.pipe.tasks.functors.CompositeFunctor.funcDict

Definition at line 413 of file functors.py.

name [1/2]

str lsst.pipe.tasks.functors.CompositeFunctor.name = "CompositeFunctor"

static

Definition at line 408 of file functors.py.

name [2/2]

lsst.pipe.tasks.functors.CompositeFunctor.name

Definition at line 498 of file functors.py.

---

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

• /j/snowflake/release/lsstsw/stack/lsst-scipipe-8.0.0/Linux64/pipe_tasks/g2a9a014e59+6f9be1b9cd/python/lsst/pipe/tasks/functors.py