_actions.py

Go to the documentation of this file.

from __future__ import annotations
 
__all__ = ("SingleColumnAction", "MultiColumnAction", "CoordColumn", "MagColumnDN", "SumColumns", "AddColumn",
           "DivideColumns", "SubtractColumns", "MultiplyColumns", "MagColumnNanoJansky",)
 
from typing import Iterable
 
import numpy as np
import pandas as pd
from astropy import units
 
from ..configurableActions import ConfigurableActionStructField, ConfigurableActionField
from ._baseDataFrameActions import DataFrameAction
from ._evalColumnExpression import makeColumnExpressionAction
 
from lsst.pex.config import Field
 
 
class SingleColumnAction(DataFrameAction):
    column = Field(doc="Column to load for this action", dtype=str, optional=False)
 
    @property
    def columns(self) -> Iterable[str]:
        return (self.columncolumn, )
 
    def __call__(self, df, **kwargs):
        return df[self.columncolumn]
 
 
class MultiColumnAction(DataFrameAction):
    actions = ConfigurableActionStructField(doc="Configurable actions to use in a joint action")
 
    @property
    def columns(self) -> Iterable[str]:
        yield from (column for action in self.actionsactions for column in action.columns)
 
 
class CoordColumn(SingleColumnAction):
    inRadians = Field(doc="Return the column in radians if true", default=True, dtype=bool)
 
    def __call__(self, df):
        col = super().__call__(df)
        return col * 180 / np.pi if self.inRadiansinRadians else col
 
 
class MagColumnDN(SingleColumnAction):
    coadd_zeropoint = Field(doc="Magnitude zero point", dtype=float, default=27)
 
    def __call__(self, df: pd.DataFrame, **kwargs):
        if not (fluxMag0 := kwargs.get('fluxMag0')):
            fluxMag0 = 1/np.power(10, -0.4*self.coadd_zeropointcoadd_zeropoint)
 
        with np.warnings.catch_warnings():
            np.warnings.filterwarnings('ignore', r'invalid value encountered')
            np.warnings.filterwarnings('ignore', r'divide by zero')
            return -2.5 * np.log10(df[self.columncolumn] / fluxMag0)
 
 
class MagColumnNanoJansky(SingleColumnAction):
 
    def __call__(self, df: pd.DataFrame, **kwargs):
 
        with np.warnings.catch_warnings():
            np.warnings.filterwarnings('ignore', r'invalid value encountered')
            np.warnings.filterwarnings('ignore', r'divide by zero')
            return -2.5 * np.log10((df[self.columncolumn] * 1e-9) / 3631.0)
 
 
class NanoJansky(SingleColumnAction):
    ab_flux_scale = Field(doc="Scaling of ab flux", dtype=float, default=(0*units.ABmag).to_value(units.nJy))
    coadd_zeropoint = Field(doc="Magnitude zero point", dtype=float, default=27)
 
    def __call__(self, df, **kwargs):
        dataNumber = super().__call__(df, **kwargs)
        if not (fluxMag0 := kwargs.get('fluxMag0')):
            fluxMag0 = 1/np.power(10, -0.4*self.coadd_zeropointcoadd_zeropoint)
        return self.ab_flux_scaleab_flux_scale * dataNumber / fluxMag0
 
    def setDefaults(self):
        super().setDefaults()
        self.cachecachecache = True  # cache this action for future calls
 
 
class NanoJanskyErr(SingleColumnAction):
    flux_mag_err = Field(doc="Error in the magnitude zeropoint", dtype=float, default=0)
    flux_action = ConfigurableActionField(doc="Action to use if flux is not provided to the call method",
                                          default=NanoJansky, dtype=DataFrameAction)
 
    @property
    def columns(self):
        yield from zip((self.columncolumn,), self.flux_actionflux_action.columns)
 
    def __call__(self, df, flux_column=None, flux_mag_err=None, **kwargs):
        if flux_column is None:
            flux_column = self.flux_actionflux_action(df, **kwargs)
        if flux_mag_err is None:
            flux_mag_err = self.flux_mag_errflux_mag_err
 
 
_docs = """This is a `DataFrameAction` that is designed to add two columns
together and return the result.
"""
SumColumns = makeColumnExpressionAction("SumColumns", "colA+colB",
                                        exprDefaults={"colA": SingleColumnAction,
                                                      "colB": SingleColumnAction},
                                        docstring=_docs)
 
_docs = """This is a `MultiColumnAction` that is designed to subtract two columns
together and return the result.
"""
SubtractColumns = makeColumnExpressionAction("SubtractColumns", "colA-colB",
                                             exprDefaults={"colA": SingleColumnAction,
                                                           "colB": SingleColumnAction},
                                             docstring=_docs)
 
_docs = """This is a `MultiColumnAction` that is designed to multiply two columns
together and return the result.
"""
MultiplyColumns = makeColumnExpressionAction("MultiplyColumns", "colA*colB",
                                             exprDefaults={"colA": SingleColumnAction,
                                                           "colB": SingleColumnAction},
                                             docstring=_docs)
 
_docs = """This is a `MultiColumnAction` that is designed to multiply two columns
together and return the result.
"""
DivideColumns = makeColumnExpressionAction("DivideColumns", "colA/colB",
                                           exprDefaults={"colA": SingleColumnAction,
                                                         "colB": SingleColumnAction},
                                           docstring=_docs)
 
 
class AddColumn(DataFrameAction):
    aggregator = ConfigurableActionField(doc="This is an instance of a Dataframe action that will be used "
                                             "to create a new column", dtype=DataFrameAction)
    newColumn = Field(doc="Name of the new column to add", dtype=str)
 
    @property
    def columns(self) -> Iterable[str]:
        yield from self.aggregatoraggregator.columns
 
    def __call__(self, df, **kwargs) -> pd.DataFrame:
        # do your calculation and and
        df[self.newColumnnewColumn] = self.aggregatoraggregator(df, kwargs)
        return df