lsst.pipe.tasks.functors.CompositeFunctor Class Reference

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

Public Member Functions
def	__init__ (self, funcs, **kwargs)
def	filt (self)
def	filt (self, filt)
def	update (self, new)
def	columns (self)
def	multilevelColumns (self, data, **kwargs)
def	__call__ (self, data, **kwargs)
def	renameCol (cls, col, renameRules)
def	from_file (cls, filename, **kwargs)
def	from_yaml (cls, translationDefinition, **kwargs)
def	noDup (self)
def	multilevelColumns (self, data, columnIndex=None, returnTuple=False)
def	__call__ (self, data, dropna=False)
def	difference (self, data1, data2, **kwargs)
def	fail (self, df)
def	name (self)
def	shortname (self)

Public Attributes
	funcDict
	filt

Static Public Attributes
	dataset = None

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

Constructor & Destructor Documentation

__init__()

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

Definition at line 403 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__() [1/2]

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

Apply the functor to the data table

Parameters
----------
data : `lsst.daf.butler.DeferredDatasetHandle`,
       `lsst.pipe.tasks.parquetTable.MultilevelParquetTable`,
       `lsst.pipe.tasks.parquetTable.ParquetTable`,
       or `pandas.DataFrame`.
    The table or a pointer to a table on disk from which columns can
    be accessed

Definition at line 455 of file functors.py.

    def __call__(self, data, **kwargs):
        """Apply the functor to the data table
 
        Parameters
        ----------
        data : `lsst.daf.butler.DeferredDatasetHandle`,
               `lsst.pipe.tasks.parquetTable.MultilevelParquetTable`,
               `lsst.pipe.tasks.parquetTable.ParquetTable`,
               or `pandas.DataFrame`.
            The table or a pointer to a table on disk from which columns can
            be accessed
        """
        columnIndex = self._get_columnIndex(data)
 
        # First, determine whether data has a multilevel index (either gen2 or gen3)
        is_multiLevel = isinstance(data, MultilevelParquetTable) or isinstance(columnIndex, pd.MultiIndex)
 
        # Multilevel index, gen2 or gen3
        if is_multiLevel:
            columns = self.multilevelColumns(data, columnIndex=columnIndex)
 
            if isinstance(data, MultilevelParquetTable):
                # Read data into memory the gen2 way
                df = data.toDataFrame(columns=columns, droplevels=False)
            elif isinstance(data, DeferredDatasetHandle):
                # Read data into memory the gen3 way
                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:
                    try:
                        valDict[k] = f.fail(subdf)
                    except NameError:
                        raise e
 
        else:
            if isinstance(data, DeferredDatasetHandle):
                # input if Gen3 deferLoad=True
                df = data.get(parameters={"columns": self.columns})
            elif isinstance(data, pd.DataFrame):
                # input if Gen3 deferLoad=False
                df = data
            else:
                # Original Gen2 input is type ParquetTable and the fallback
                df = data.toDataFrame(columns=self.columns)
 
            valDict = {k: f._func(df) for k, f in self.funcDict.items()}
 
        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
 

__call__() [2/2]

def lsst.pipe.tasks.functors.Functor.__call__ (   self,   data,   dropna = False  )

inherited

Definition at line 340 of file functors.py.

    def __call__(self, data, dropna=False):
        try:
            df = self._get_data(data)
            vals = self._func(df)
        except Exception:
            vals = self.fail(df)
        if dropna:
            vals = self._dropna(vals)
 
        return vals
 

columns()

def lsst.pipe.tasks.functors.CompositeFunctor.columns

(

self

)

Columns required to perform calculation

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

Definition at line 438 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()

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

inherited

Computes difference between functor called on two different ParquetTable objects

Definition at line 351 of file functors.py.

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

fail()

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

inherited

Definition at line 356 of file functors.py.

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

filt() [1/2]

def lsst.pipe.tasks.functors.CompositeFunctor.filt

(

self

)

Definition at line 415 of file functors.py.

    def filt(self):
        return self._filt
 

filt() [2/2]

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

Definition at line 419 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()

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

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

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

Definition at line 539 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() [1/2]

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

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

multilevelColumns() [2/2]

def lsst.pipe.tasks.functors.Functor.multilevelColumns (   self,   data,   columnIndex = None,   returnTuple = False  )

inherited

Returns columns needed by functor from multilevel dataset

To access tables with multilevel column structure, the `MultilevelParquetTable`
or `DeferredDatasetHandle` need to be passed either a list of tuples or a
dictionary.

Parameters
----------
data : `MultilevelParquetTable` or `DeferredDatasetHandle`

columnIndex (optional): pandas `Index` object
    either passed or read in from `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.

Definition at line 229 of file functors.py.

    def multilevelColumns(self, data, columnIndex=None, returnTuple=False):
        """Returns columns needed by functor from multilevel dataset
 
        To access tables with multilevel column structure, the `MultilevelParquetTable`
        or `DeferredDatasetHandle` need to be passed either a list of tuples or a
        dictionary.
 
        Parameters
        ----------
        data : `MultilevelParquetTable` or `DeferredDatasetHandle`
 
        columnIndex (optional): pandas `Index` object
            either passed or read in from `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.
 
        """
        if isinstance(data, DeferredDatasetHandle) and columnIndex is None:
            columnIndex = data.get(component="columns")
 
        # Confirm that the dataset has the column levels the functor is expecting it to have.
        columnLevels = self._get_data_columnLevels(data, columnIndex)
 
        columnDict = {'column': self.columns,
                      'dataset': self.dataset}
        if self.filt is None:
            columnLevelNames = self._get_data_columnLevelNames(data, columnIndex)
            if "band" in columnLevels:
                if self.dataset == "ref":
                    columnDict["band"] = columnLevelNames["band"][0]
                else:
                    raise ValueError(f"'filt' not set for functor {self.name}"
                                     f"(dataset {self.dataset}) "
                                     "and ParquetTable "
                                     "contains multiple filters in column index. "
                                     "Set 'filt' or set 'dataset' to 'ref'.")
        else:
            columnDict['band'] = self.filt
 
        if isinstance(data, MultilevelParquetTable):
            return data._colsFromDict(columnDict)
        elif isinstance(data, DeferredDatasetHandle):
            if returnTuple:
                return self._colsFromDict(columnDict, columnIndex=columnIndex)
            else:
                return columnDict
 

name()

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

inherited

Full name of functor (suitable for figure labels)

Reimplemented in lsst.pipe.tasks.functors.Ratio, lsst.pipe.tasks.functors.LocalDipoleDiffFluxErr, lsst.pipe.tasks.functors.LocalDipoleDiffFlux, lsst.pipe.tasks.functors.LocalDipoleMeanFluxErr, lsst.pipe.tasks.functors.LocalDipoleMeanFlux, lsst.pipe.tasks.functors.LocalMagnitudeErr, lsst.pipe.tasks.functors.LocalMagnitude, lsst.pipe.tasks.functors.LocalNanojanskyErr, lsst.pipe.tasks.functors.LocalNanojansky, lsst.pipe.tasks.functors.Photometry, lsst.pipe.tasks.functors.ConvertPixelSqToArcsecondsSq, lsst.pipe.tasks.functors.ConvertPixelToArcseconds, lsst.pipe.tasks.functors.Color, lsst.pipe.tasks.functors.MagDiff, lsst.pipe.tasks.functors.MagErr, lsst.pipe.tasks.functors.Mag, lsst.pipe.tasks.functors.Column, and lsst.pipe.tasks.functors.CustomFunctor.

Definition at line 360 of file functors.py.

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

noDup()

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

inherited

Definition at line 148 of file functors.py.

    def noDup(self):
        if self._noDup is not None:
            return self._noDup
        else:
            return self._defaultNoDup
 

renameCol()

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

Definition at line 521 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()

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

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

update()

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

Definition at line 425 of file functors.py.

    def update(self, new):
        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

dataset

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

static

Definition at line 401 of file functors.py.

filt

lsst.pipe.tasks.functors.CompositeFunctor.filt

Definition at line 435 of file functors.py.

funcDict

lsst.pipe.tasks.functors.CompositeFunctor.funcDict

Definition at line 406 of file functors.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-147-g0e635eb1+1acddb5be5/python/lsst/pipe/tasks/functors.py