bcg-x-official / sklearndf Goto Github PK

class ColumnTransformerWrapperDF(
    TransformerWrapperDF[ColumnTransformer], metaclass=ABCMeta
):
    """
    DF wrapper for :class:`sklearn.compose.ColumnTransformer`.
    Requires all transformers passed as the ``transformers`` parameter to implement
    :class:`.TransformerDF`.
    """

    __DROP = "drop"
    __PASSTHROUGH = "passthrough"

    __SPECIAL_TRANSFORMERS = (__DROP, __PASSTHROUGH)

    def _validate_delegate_estimator(self) -> None:
        column_transformer: ColumnTransformer = self.native_estimator

        if column_transformer.remainder != ColumnTransformerWrapperDF.__DROP:
            raise ValueError(
                f"unsupported value for arg remainder: ({column_transformer.remainder})"
            )

        non_compliant_transformers: List[str] = [
            type(transformer).__name__
            for _, transformer, _ in column_transformer.transformers
            if not (
                isinstance(transformer, TransformerDF)
                or transformer in ColumnTransformerWrapperDF.__SPECIAL_TRANSFORMERS
            )
        ]
        if non_compliant_transformers:
            from .. import ColumnTransformerDF

            raise ValueError(
                f"{ColumnTransformerDF.__name__} only accepts instances of "
                f"{TransformerDF.__name__} or special values "
                f'"{" and ".join(ColumnTransformerWrapperDF.__SPECIAL_TRANSFORMERS)}" '
                "as valid transformers, but "
                f'also got: {", ".join(non_compliant_transformers)}'
            )

    def _get_features_original(self) -> pd.Series:
        """
        Return the series mapping output column names to original columns names.
        :return: the series with index the column names of the output dataframe and
        values the corresponding input column names.
        """

        return reduce(
            lambda x, y: x.append(y),
            (
                (
                    pd.Series(index=columns, data=columns)
                    if df_transformer == ColumnTransformerWrapperDF.__PASSTHROUGH
                    else df_transformer.feature_names_original_
                )
                for _, df_transformer, columns in self.native_estimator.transformers_
                if (
                    len(columns) > 0
                    and df_transformer != ColumnTransformerWrapperDF.__DROP
                )
            ),
        )

class ColumnTransformerCustomWrapperDF(
    TransformerWrapperDF[ColumnTransformer], metaclass=ABCMeta
):
    """
    DF wrapper for :class:`sklearn.compose.ColumnTransformer`.
    Requires all transformers passed as the ``transformers`` parameter to implement
    :class:`.TransformerDF`.
    """

    __DROP = "drop"
    __PASSTHROUGH = "passthrough"

    __SPECIAL_TRANSFORMERS = (__DROP, __PASSTHROUGH)

    def _validate_delegate_estimator(self) -> None:
        column_transformer: ColumnTransformer = self.native_estimator

        if (
            column_transformer.remainder
            not in ColumnTransformerCustomWrapperDF.__SPECIAL_TRANSFORMERS
        ):
            raise ValueError(
                f"unsupported value for arg remainder: ({column_transformer.remainder})"
            )

        non_compliant_transformers: List[str] = [
            type(transformer).__name__
            for _, transformer, _ in column_transformer.transformers
            if not (
                isinstance(transformer, TransformerDF)
                or transformer
                in ColumnTransformerCustomWrapperDF.__SPECIAL_TRANSFORMERS
            )
        ]
        if non_compliant_transformers:
            authorised_transformers = " and ".join(
                ColumnTransformerCustomWrapperDF.__SPECIAL_TRANSFORMERS
            )
            raise ValueError(
                f"{ColumnTransformerDF.__name__} only accepts instances of "
                f"{TransformerDF.__name__} or special values "
                f'"{authorised_transformers}" as valid transformers, but '
                f'also got: {", ".join(non_compliant_transformers)}'
            )

    def _get_features_original(self) -> pd.Series:
        """
        Return the series mapping output column names to original columns names.
        :return: the series with index the column names of the output dataframe and
        values the corresponding input column names.
        """
        result = reduce(
            lambda x, y: x.append(y),
            (
                (
                    pd.Series(
                        index=self.feature_names_in_[columns],
                        data=self.feature_names_in_[columns],
                    )
                    if df_transformer == ColumnTransformerCustomWrapperDF.__PASSTHROUGH
                    else df_transformer.feature_names_original_
                )
                for _, df_transformer, columns in self.native_estimator.transformers_
                if (
                    len(columns) > 0
                    and df_transformer != ColumnTransformerCustomWrapperDF.__DROP
                )
            ),
        )
        return result

# Instantiation of ColumnCustomTransformerDF
ColumnCustomTransformerDF = make_df_transformer(
    ColumnTransformer,
    name="ColumnTransformerCustomDF",
    base_wrapper=ColumnTransformerCustomWrapperDF,
)

import pandas as pd
import numpy as np
from sklearndf.transformation import StandardScalerDF

# Initialise a random DataFrame
df = pd.DataFrame(np.random.randint(1, 100, size=(10, 2)), columns=["A", "B"])
print(df)

# Instantiation and Fitting of a Standard Scaler
scaler = StandardScalerDF()
scaler.fit(df)
df = scaler.transform(df)
print(df)

# Inverse the Scaling
df = scaler.inverse_transform(df) ## ERROR --> NotFittedError: StandardScalerDF is not fitted