API: preprocessing.base

skyulf.preprocessing.base

Base classes and (X, y) unpack/pack decorators shared by every preprocessing node.

BaseApplier

Bases: ABC

Transforming half of a node pair: applies a fitted artifact to data.

Subclasses implement :meth:apply, taking everything they need from params so the same applier instance can be reused across splits and datasets without carrying state.

Source code in skyulf-core/skyulf/preprocessing/base.py

class BaseApplier(ABC):
    """Transforming half of a node pair: applies a fitted artifact to data.

    Subclasses implement :meth:`apply`, taking everything they need from
    ``params`` so the same applier instance can be reused across splits and
    datasets without carrying state.
    """

    @abstractmethod
    def apply(self, df: pd.DataFrame | SkyulfDataFrame | tuple, params: dict[str, Any]) -> Any:
        """Applies the transformation using fitted parameters.

        The return type is intentionally `Any` because the concrete shape
        depends on the input: passing a `DataFrame` returns a `DataFrame`;
        passing an `(X, y)` tuple returns a tuple; splitters return
        `SplitDataset`. Encoding every case as a union forces callers to
        defensively narrow on every use, which is worse than `Any` here.
        """

apply(df, params) abstractmethod

Applies the transformation using fitted parameters.

The return type is intentionally Any because the concrete shape depends on the input: passing a DataFrame returns a DataFrame; passing an (X, y) tuple returns a tuple; splitters return SplitDataset. Encoding every case as a union forces callers to defensively narrow on every use, which is worse than Any here.

Source code in skyulf-core/skyulf/preprocessing/base.py

@abstractmethod
def apply(self, df: pd.DataFrame | SkyulfDataFrame | tuple, params: dict[str, Any]) -> Any:
    """Applies the transformation using fitted parameters.

    The return type is intentionally `Any` because the concrete shape
    depends on the input: passing a `DataFrame` returns a `DataFrame`;
    passing an `(X, y)` tuple returns a tuple; splitters return
    `SplitDataset`. Encoding every case as a union forces callers to
    defensively narrow on every use, which is worse than `Any` here.
    """

BaseCalculator

Bases: ABC

Learning half of a node pair: derives parameters and returns them as an artifact.

Subclasses implement :meth:fit and hand back a plain mapping that the matching :class:BaseApplier consumes. Calculators never transform data and hold no state between calls.

Source code in skyulf-core/skyulf/preprocessing/base.py

class BaseCalculator(ABC):
    """Learning half of a node pair: derives parameters and returns them as an artifact.

    Subclasses implement :meth:`fit` and hand back a plain mapping that the
    matching :class:`BaseApplier` consumes. Calculators never transform data
    and hold no state between calls.
    """

    @abstractmethod
    def fit(
        self, df: pd.DataFrame | SkyulfDataFrame | tuple, config: dict[str, Any]
    ) -> Mapping[str, Any]:
        """Calculates parameters from the training data.

        Returns a Mapping of fitted parameters (typically a TypedDict
        ``*Artifact`` declared in ``preprocessing._artifacts``). The return
        type is ``Mapping`` rather than ``Dict`` so concrete TypedDict
        subclasses are valid LSP-substitutable returns.
        """

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema | None:
        """Best-effort prediction of the output schema from config alone.

        Override this in concrete Calculators when the output columns/dtypes
        can be derived purely from ``input_schema`` and ``config`` (i.e.
        without seeing data). Examples:

        * Scalers — preserve column names and predict output dtype promotion.
        * Drop columns by name — drop the configured names.
        * One-hot — adds K columns per categorical (K is data-dependent →
          return ``None``).

        Default returns ``None`` to signal "unknown / data-dependent";
        callers should fall back to runtime introspection.
        """
        return None

fit(df, config) abstractmethod

Calculates parameters from the training data.

Returns a Mapping of fitted parameters (typically a TypedDict *Artifact declared in preprocessing._artifacts). The return type is Mapping rather than Dict so concrete TypedDict subclasses are valid LSP-substitutable returns.

Source code in skyulf-core/skyulf/preprocessing/base.py

@abstractmethod
def fit(
    self, df: pd.DataFrame | SkyulfDataFrame | tuple, config: dict[str, Any]
) -> Mapping[str, Any]:
    """Calculates parameters from the training data.

    Returns a Mapping of fitted parameters (typically a TypedDict
    ``*Artifact`` declared in ``preprocessing._artifacts``). The return
    type is ``Mapping`` rather than ``Dict`` so concrete TypedDict
    subclasses are valid LSP-substitutable returns.
    """

infer_output_schema(input_schema, config)

Best-effort prediction of the output schema from config alone.

Override this in concrete Calculators when the output columns/dtypes can be derived purely from input_schema and config (i.e. without seeing data). Examples:

• Scalers — preserve column names and predict output dtype promotion.
• Drop columns by name — drop the configured names.
• One-hot — adds K columns per categorical (K is data-dependent → return None).

Default returns None to signal "unknown / data-dependent"; callers should fall back to runtime introspection.

Source code in skyulf-core/skyulf/preprocessing/base.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema | None:
    """Best-effort prediction of the output schema from config alone.

    Override this in concrete Calculators when the output columns/dtypes
    can be derived purely from ``input_schema`` and ``config`` (i.e.
    without seeing data). Examples:

    * Scalers — preserve column names and predict output dtype promotion.
    * Drop columns by name — drop the configured names.
    * One-hot — adds K columns per categorical (K is data-dependent →
      return ``None``).

    Default returns ``None`` to signal "unknown / data-dependent";
    callers should fall back to runtime introspection.
    """
    return None

StatefulTransformer

Fits + applies one pipeline step.

Accepts anything satisfying :class:~skyulf.core.protocols.CalculatorProtocol / :class:~skyulf.core.protocols.ApplierProtocol (structural typing) — a BaseCalculator/BaseApplier subclass, or any duck-typed object exposing matching fit/apply methods, works without subclassing.

Source code in skyulf-core/skyulf/preprocessing/base.py

class StatefulTransformer:
    """Fits + applies one pipeline step.

    Accepts anything satisfying :class:`~skyulf.core.protocols.CalculatorProtocol` /
    :class:`~skyulf.core.protocols.ApplierProtocol` (structural typing) — a
    ``BaseCalculator``/``BaseApplier`` subclass, or any duck-typed object
    exposing matching ``fit``/``apply`` methods, works without subclassing.
    """

    def __init__(
        self,
        calculator: CalculatorProtocol,
        applier: ApplierProtocol,
        node_id: str,
        apply_on_test: bool = True,
        apply_on_validation: bool = True,
    ):
        """Store the node pair and which held-out splits it should be applied to.

        ``apply_on_test``/``apply_on_validation`` turn the transform of those
        splits off without dropping them from the result. ``params`` starts
        empty and is filled by the first :meth:`fit_transform`; the profiling
        counters are zeroed here and overwritten on every run.
        """
        self.calculator = calculator
        self.applier = applier
        self.node_id = node_id
        self.apply_on_test = apply_on_test
        self.apply_on_validation = apply_on_validation
        self.params: dict[str, Any] = {}  # Store params in memory instead of ArtifactStore
        # Profiling metrics
        self.fit_time: float = 0.0
        self.peak_memory_bytes: int = 0
        self.rows_in: int = 0
        self.rows_out: int = 0
        self.train_sample_weight_: Any = None

    def fit_transform(
        self,
        dataset: SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple,
        config: dict[str, Any],
    ) -> SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple:
        """Fit on the training data, apply to the held-out splits, and profile the run.

        Wall time, row counts and peak memory land on ``self`` rather than in
        the return value, and are recorded in a ``finally`` so a failing node
        still reports what it cost. ``tracemalloc`` is borrowed, not owned: if
        the caller is already tracing, only the peak growth since entry can be
        attributed to this step and tracing is left running; otherwise it is
        started here and stopped again on exit.
        """
        validate_weighted_transformer(dataset, self.calculator, self.applier)
        self.rows_in, _ = get_data_stats(dataset)
        tracing_was_active = tracemalloc.is_tracing()
        if not tracing_was_active:
            tracemalloc.start()
            tracemalloc.reset_peak()
        peak_baseline = tracemalloc.get_traced_memory()[1] if tracemalloc.is_tracing() else 0
        # With caller-owned tracing we can only report new global peak growth since entry.
        self.peak_memory_bytes = 0
        self.rows_out = 0
        start = time.time()

        try:
            result = self._fit_transform_inner(dataset, config)
            self.rows_out, _ = get_data_stats(result)
            return result
        finally:
            self.fit_time = time.time() - start
            if tracemalloc.is_tracing():
                _, peak = tracemalloc.get_traced_memory()
                self.peak_memory_bytes = max(0, peak - peak_baseline)
            if not tracing_was_active and tracemalloc.is_tracing():
                tracemalloc.stop()

    def _fit_and_apply_training_data(
        self,
        data: Any,
        config: dict[str, Any],
        *,
        guard_split_output: bool = True,
    ) -> Any:
        """Fit one training input and return its train-time representation."""
        if isinstance(self.calculator, TrainTransformCalculatorProtocol):
            params, transformed = self.calculator.fit_transform_train(data, config)
            self.params = cast(dict[str, Any], params)
            if isinstance(transformed, SplitDataset):
                raise TypeError(
                    "Calculator returned SplitDataset from fit_transform_train, which is not supported."
                )
            return transformed

        self.params = cast(dict[str, Any], self.calculator.fit(data, config))
        if guard_split_output:
            return self._apply_guarded(data, self.params)
        return self.applier.apply(data, self.params)

    def fit_transform_weighted(self, data: Any, config: Any, weights: Any) -> tuple[Any, Any]:
        """Fit on real targets and propagate weights through the actual row selection."""
        if weights is None:
            return self._fit_and_apply_training_data(data, config), None
        validate_weighted_components(self.calculator, self.applier)
        weights = validate_sample_weight(weights, len(unpack_pipeline_input(data)[0]))
        resampling = import_module("skyulf.preprocessing.resampling")

        if type(self.calculator) in (
            resampling.OversamplingCalculator,
            resampling.UndersamplingCalculator,
        ):
            helper = import_module("skyulf.preprocessing._weighted_resampling")

            self.params, transformed, weights = helper.fit_resample_weighted(
                self.calculator, self.applier, data, config, weights
            )
            return transformed, weights
        if has_builtin_row_mapping(self.applier):
            self.params = cast(dict[str, Any], self.calculator.fit(data, config))
            return apply_weighted(
                self.applier, data, self.params | {"_history_training": True}, weights
            )
        if isinstance(self.calculator, TrainTransformCalculatorProtocol):
            validate_training_representation(self.calculator, self.applier)
            transformed = self._fit_and_apply_training_data(data, config)

            return transformed, validate_sample_weight(
                weights, len(unpack_pipeline_input(transformed)[0])
            )
        self.params = cast(dict[str, Any], self.calculator.fit(data, config))
        return apply_weighted(self.applier, data, self.params, weights)

    def _fit_transform_inner(
        self,
        dataset: SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple,
        config: dict[str, Any],
    ) -> SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple:
        # Check for DataFrame-like (Pandas, Polars, Wrapper)
        if (
            hasattr(dataset, "shape")
            and hasattr(dataset, "columns")
            and not isinstance(dataset, tuple)
        ):
            # Fit on the whole dataframe (be careful about leakage!)
            # ty can't narrow a Union through hasattr — cast once for both calls.
            frame = cast(Any, dataset)
            return self._fit_and_apply_training_data(frame, config, guard_split_output=False)

        # If dataset is a tuple (e.g. from FeatureTargetSplitter), pass it through.
        # This allows nodes like TrainTestSplitter to accept (X, y) tuples.
        if isinstance(dataset, tuple):
            return self._fit_and_apply_training_data(dataset, config, guard_split_output=False)

        # 1. Calculate on Train
        new_train, train_weight = self.fit_transform_weighted(
            dataset.train, config, dataset.train_sample_weight
        )

        # 2. Apply fitted params to held-out splits only
        new_test = dataset.test
        if self.apply_on_test:
            new_test = self._apply_guarded(dataset.test, self.params)

        new_val = dataset.validation
        if self.apply_on_validation and dataset.validation is not None:
            new_val = self._apply_guarded(dataset.validation, self.params)

        return SplitDataset(
            train=new_train,
            test=new_test,
            validation=new_val,
            train_sample_weight=train_weight,
            evaluation_coverage=self._evaluation_coverage(dataset, new_test, new_val),
        )

    def _evaluation_coverage(
        self, dataset: SplitDataset, new_test: Any, new_val: Any
    ) -> dict[str, dict[str, Any]]:
        """Carry held-out denominators across nodes without storing row-level data."""
        coverage = deepcopy(dataset.evaluation_coverage)
        for name, result in (("test", new_test), ("validation", new_val)):
            original = getattr(dataset, name)
            if original is None or result is None:
                continue
            before = len(unpack_pipeline_input(original)[0])
            after = len(unpack_pipeline_input(result)[0])
            previous = coverage.get(name, record_coverage(before, before))
            steps = list(previous.get("steps", []))
            if before != after:
                steps.append(
                    {"name": self.node_id, **record_coverage(before, after, step_name=self.node_id)}
                )
            coverage[name] = record_coverage(
                previous["input_rows"],
                after,
                steps,
                reason=previous.get("reason"),
                step_name=self.node_id,
            )
        return coverage

    def _apply_guarded(self, data: Any, params: dict[str, Any]) -> Any:
        """Apply the applier to `data` and raise if it produces a nested SplitDataset."""
        result = self.applier.apply(data, params)
        if isinstance(result, SplitDataset):
            raise TypeError(
                "Applier returned SplitDataset inside StatefulTransformer, which is not supported."
            )
        return result

    def _apply_to_split_dataset(
        self, dataset: SplitDataset, params: dict[str, Any]
    ) -> SplitDataset:
        """Apply the applier to each split (train/test/validation) of a SplitDataset."""
        validate_weighted_transformer(dataset, self.calculator, self.applier)
        if is_sampling_applier(self.applier):
            return dataset
        if dataset.train_sample_weight is None:
            new_train, train_weight = self._apply_guarded(dataset.train, params), None
        else:
            weights = validate_sample_weight(
                dataset.train_sample_weight, len(unpack_pipeline_input(cast(Any, dataset.train))[0])
            )
            new_train, train_weight = apply_weighted(self.applier, dataset.train, params, weights)

        new_test = dataset.test
        if self.apply_on_test:
            new_test = self._apply_guarded(dataset.test, params)

        new_val = dataset.validation
        if self.apply_on_validation and dataset.validation is not None:
            new_val = self._apply_guarded(dataset.validation, params)

        return SplitDataset(
            train=new_train,
            test=new_test,
            validation=new_val,
            train_sample_weight=train_weight,
            evaluation_coverage=self._evaluation_coverage(dataset, new_test, new_val),
        )

    def transform(
        self, dataset: SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple
    ) -> SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple:
        """Apply the params stored by the last fit, without refitting.

        Input detection mirrors :meth:`_fit_transform_inner` so a bare frame or
        an ``(X, y)`` tuple goes straight to the applier, while a
        ``SplitDataset`` is applied split-by-split honouring
        ``apply_on_test``/``apply_on_validation``. Built-in resamplers leave
        ``SplitDataset`` unchanged during replay, with or without weights.
        Calling this before any
        :meth:`fit_transform` hands the applier an empty params dict rather
        than raising.
        """
        # Use stored params
        params = self.params

        # Check for DataFrame-like (Pandas, Polars, Wrapper) input, mirroring
        # `_fit_transform_inner`'s detection above -- an `isinstance(dataset,
        # pd.DataFrame)`-only check here would misroute a raw (unwrapped)
        # `pl.DataFrame` into the SplitDataset branch below, crashing with
        # `AttributeError: 'DataFrame' object has no attribute 'train'`.
        if (
            hasattr(dataset, "shape")
            and hasattr(dataset, "columns")
            and not isinstance(dataset, tuple)
        ):
            return self.applier.apply(cast(Any, dataset), params)

        if isinstance(dataset, tuple):
            return self.applier.apply(dataset, params)

        # 2. Apply
        # ty can't narrow SplitDataset out of the SkyulfDataFrame branch of this
        # Union via isinstance alone (mirrors the `frame = cast(Any, dataset)`
        # note in `_fit_transform_inner` above).
        return self._apply_to_split_dataset(cast(SplitDataset, dataset), params)

__init__(calculator, applier, node_id, apply_on_test=True, apply_on_validation=True)

Store the node pair and which held-out splits it should be applied to.

apply_on_test/apply_on_validation turn the transform of those splits off without dropping them from the result. params starts empty and is filled by the first :meth:fit_transform; the profiling counters are zeroed here and overwritten on every run.

Source code in skyulf-core/skyulf/preprocessing/base.py

def __init__(
    self,
    calculator: CalculatorProtocol,
    applier: ApplierProtocol,
    node_id: str,
    apply_on_test: bool = True,
    apply_on_validation: bool = True,
):
    """Store the node pair and which held-out splits it should be applied to.

    ``apply_on_test``/``apply_on_validation`` turn the transform of those
    splits off without dropping them from the result. ``params`` starts
    empty and is filled by the first :meth:`fit_transform`; the profiling
    counters are zeroed here and overwritten on every run.
    """
    self.calculator = calculator
    self.applier = applier
    self.node_id = node_id
    self.apply_on_test = apply_on_test
    self.apply_on_validation = apply_on_validation
    self.params: dict[str, Any] = {}  # Store params in memory instead of ArtifactStore
    # Profiling metrics
    self.fit_time: float = 0.0
    self.peak_memory_bytes: int = 0
    self.rows_in: int = 0
    self.rows_out: int = 0
    self.train_sample_weight_: Any = None

fit_transform(dataset, config)

Fit on the training data, apply to the held-out splits, and profile the run.

Wall time, row counts and peak memory land on self rather than in the return value, and are recorded in a finally so a failing node still reports what it cost. tracemalloc is borrowed, not owned: if the caller is already tracing, only the peak growth since entry can be attributed to this step and tracing is left running; otherwise it is started here and stopped again on exit.

Source code in skyulf-core/skyulf/preprocessing/base.py

def fit_transform(
    self,
    dataset: SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple,
    config: dict[str, Any],
) -> SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple:
    """Fit on the training data, apply to the held-out splits, and profile the run.

    Wall time, row counts and peak memory land on ``self`` rather than in
    the return value, and are recorded in a ``finally`` so a failing node
    still reports what it cost. ``tracemalloc`` is borrowed, not owned: if
    the caller is already tracing, only the peak growth since entry can be
    attributed to this step and tracing is left running; otherwise it is
    started here and stopped again on exit.
    """
    validate_weighted_transformer(dataset, self.calculator, self.applier)
    self.rows_in, _ = get_data_stats(dataset)
    tracing_was_active = tracemalloc.is_tracing()
    if not tracing_was_active:
        tracemalloc.start()
        tracemalloc.reset_peak()
    peak_baseline = tracemalloc.get_traced_memory()[1] if tracemalloc.is_tracing() else 0
    # With caller-owned tracing we can only report new global peak growth since entry.
    self.peak_memory_bytes = 0
    self.rows_out = 0
    start = time.time()

    try:
        result = self._fit_transform_inner(dataset, config)
        self.rows_out, _ = get_data_stats(result)
        return result
    finally:
        self.fit_time = time.time() - start
        if tracemalloc.is_tracing():
            _, peak = tracemalloc.get_traced_memory()
            self.peak_memory_bytes = max(0, peak - peak_baseline)
        if not tracing_was_active and tracemalloc.is_tracing():
            tracemalloc.stop()

fit_transform_weighted(data, config, weights)

Fit on real targets and propagate weights through the actual row selection.

Source code in skyulf-core/skyulf/preprocessing/base.py

def fit_transform_weighted(self, data: Any, config: Any, weights: Any) -> tuple[Any, Any]:
    """Fit on real targets and propagate weights through the actual row selection."""
    if weights is None:
        return self._fit_and_apply_training_data(data, config), None
    validate_weighted_components(self.calculator, self.applier)
    weights = validate_sample_weight(weights, len(unpack_pipeline_input(data)[0]))
    resampling = import_module("skyulf.preprocessing.resampling")

    if type(self.calculator) in (
        resampling.OversamplingCalculator,
        resampling.UndersamplingCalculator,
    ):
        helper = import_module("skyulf.preprocessing._weighted_resampling")

        self.params, transformed, weights = helper.fit_resample_weighted(
            self.calculator, self.applier, data, config, weights
        )
        return transformed, weights
    if has_builtin_row_mapping(self.applier):
        self.params = cast(dict[str, Any], self.calculator.fit(data, config))
        return apply_weighted(
            self.applier, data, self.params | {"_history_training": True}, weights
        )
    if isinstance(self.calculator, TrainTransformCalculatorProtocol):
        validate_training_representation(self.calculator, self.applier)
        transformed = self._fit_and_apply_training_data(data, config)

        return transformed, validate_sample_weight(
            weights, len(unpack_pipeline_input(transformed)[0])
        )
    self.params = cast(dict[str, Any], self.calculator.fit(data, config))
    return apply_weighted(self.applier, data, self.params, weights)

transform(dataset)

Apply the params stored by the last fit, without refitting.

Input detection mirrors :meth:_fit_transform_inner so a bare frame or an (X, y) tuple goes straight to the applier, while a SplitDataset is applied split-by-split honouring apply_on_test/apply_on_validation. Built-in resamplers leave SplitDataset unchanged during replay, with or without weights. Calling this before any :meth:fit_transform hands the applier an empty params dict rather than raising.

Source code in skyulf-core/skyulf/preprocessing/base.py

def transform(
    self, dataset: SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple
) -> SplitDataset | pd.DataFrame | pl.DataFrame | SkyulfDataFrame | tuple:
    """Apply the params stored by the last fit, without refitting.

    Input detection mirrors :meth:`_fit_transform_inner` so a bare frame or
    an ``(X, y)`` tuple goes straight to the applier, while a
    ``SplitDataset`` is applied split-by-split honouring
    ``apply_on_test``/``apply_on_validation``. Built-in resamplers leave
    ``SplitDataset`` unchanged during replay, with or without weights.
    Calling this before any
    :meth:`fit_transform` hands the applier an empty params dict rather
    than raising.
    """
    # Use stored params
    params = self.params

    # Check for DataFrame-like (Pandas, Polars, Wrapper) input, mirroring
    # `_fit_transform_inner`'s detection above -- an `isinstance(dataset,
    # pd.DataFrame)`-only check here would misroute a raw (unwrapped)
    # `pl.DataFrame` into the SplitDataset branch below, crashing with
    # `AttributeError: 'DataFrame' object has no attribute 'train'`.
    if (
        hasattr(dataset, "shape")
        and hasattr(dataset, "columns")
        and not isinstance(dataset, tuple)
    ):
        return self.applier.apply(cast(Any, dataset), params)

    if isinstance(dataset, tuple):
        return self.applier.apply(dataset, params)

    # 2. Apply
    # ty can't narrow SplitDataset out of the SkyulfDataFrame branch of this
    # Union via isinstance alone (mirrors the `frame = cast(Any, dataset)`
    # note in `_fit_transform_inner` above).
    return self._apply_to_split_dataset(cast(SplitDataset, dataset), params)

TrainTransformCalculatorProtocol

Bases: Protocol

Optional calculator hook for special training-set representations.

Source code in skyulf-core/skyulf/preprocessing/base.py

@runtime_checkable
class TrainTransformCalculatorProtocol(Protocol):
    """Optional calculator hook for special training-set representations."""

    def fit_transform_train(
        self,
        df: pd.DataFrame | SkyulfDataFrame | tuple,
        config: dict[str, Any],
    ) -> tuple[Mapping[str, Any], Any]:
        """Fit on the training data and return ``(artifact, train_representation)``.

        Lets a calculator hand the training split a different representation
        than ``apply`` would produce for held-out splits. The second element
        must not be a ``SplitDataset``.
        """
        ...

fit_transform_train(df, config)

Fit on the training data and return (artifact, train_representation).

Lets a calculator hand the training split a different representation than apply would produce for held-out splits. The second element must not be a SplitDataset.

Source code in skyulf-core/skyulf/preprocessing/base.py

def fit_transform_train(
    self,
    df: pd.DataFrame | SkyulfDataFrame | tuple,
    config: dict[str, Any],
) -> tuple[Mapping[str, Any], Any]:
    """Fit on the training data and return ``(artifact, train_representation)``.

    Lets a calculator hand the training split a different representation
    than ``apply`` would produce for held-out splits. The second element
    must not be a ``SplitDataset``.
    """
    ...

apply_method(fn)

Decorator that handles unpack/pack boilerplate around an Applier's apply.

The decorated method is written with signature (self, X, y, params) instead of (self, df, params). The wrapper:

1. Calls unpack_pipeline_input(df) to get (X, y, is_tuple).
2. Invokes the user's method with the unpacked X and y.
3. If the method returns a 2-tuple (X_out, y_out), that pair is packed; otherwise the result is treated as X_out and the original y is reused.
4. Calls pack_pipeline_output to restore the original input shape.

Useful for ~50 Appliers that share the same boilerplate. Skip it for splitters (which return SplitDataset directly) or analyzers that don't transform the frame.

Source code in skyulf-core/skyulf/preprocessing/base.py

def apply_method(fn: Callable[..., Any]) -> Callable[..., Any]:
    """Decorator that handles unpack/pack boilerplate around an Applier's `apply`.

    The decorated method is written with signature ``(self, X, y, params)``
    instead of ``(self, df, params)``. The wrapper:

    1. Calls ``unpack_pipeline_input(df)`` to get ``(X, y, is_tuple)``.
    2. Invokes the user's method with the unpacked ``X`` and ``y``.
    3. If the method returns a 2-tuple ``(X_out, y_out)``, that pair is
       packed; otherwise the result is treated as ``X_out`` and the
       original ``y`` is reused.
    4. Calls ``pack_pipeline_output`` to restore the original input shape.

    Useful for ~50 Appliers that share the same boilerplate. Skip it for
    splitters (which return ``SplitDataset`` directly) or analyzers that
    don't transform the frame.
    """

    @functools.wraps(fn)
    def wrapper(self: Any, df: Any, params: dict[str, Any]) -> Any:
        X, y, is_tuple = unpack_pipeline_input(df)
        result = fn(self, X, y, params)
        if isinstance(result, tuple) and len(result) == 2:
            X_out, y_out = result
        else:
            X_out, y_out = result, y
        return pack_pipeline_output(X_out, y_out, is_tuple)

    return wrapper

fit_method(fn)

Decorator that handles unpack boilerplate around a Calculator's fit.

The decorated method is written as (self, X, y, config) and may ignore y for X-only fits. No packing is done — fit returns a params dict, not a frame.

The type parameter T preserves the specific TypedDict return type (see preprocessing._artifacts) so callers see the concrete shape.

Source code in skyulf-core/skyulf/preprocessing/base.py

def fit_method[T: Mapping[str, Any]](fn: Callable[..., T]) -> Callable[..., T]:
    """Decorator that handles unpack boilerplate around a Calculator's `fit`.

    The decorated method is written as ``(self, X, y, config)`` and may
    ignore ``y`` for X-only fits. No packing is done — `fit` returns a
    params dict, not a frame.

    The type parameter ``T`` preserves the specific TypedDict return type
    (see ``preprocessing._artifacts``) so callers see the concrete shape.
    """

    @functools.wraps(fn)
    def wrapper(self: Any, df: Any, config: dict[str, Any]) -> T:
        X, y, _ = unpack_pipeline_input(df)
        return fn(self, X, y, config)

    return wrapper  # type: ignore[return-value]