API: preprocessing.cleaning

skyulf.preprocessing.cleaning

Cleaning nodes package.

Importing this package registers all cleaning nodes (text, invalid-value, value, and alias replacement) and re-exports their public classes plus the shared alias-mapping constants for backward compatibility.

AliasReplacementApplier

Bases: BaseApplier

Canonicalise near-duplicate text values in the resolved columns.

The pandas and polars paths must agree value-for-value: each cell is normalised if it is text (lowercased, punctuation and spaces stripped) and looked up in the alias map, and a cell that matches nothing is left exactly as it was. Explicitly selected non-text values are unchanged, including numeric 0/1 and booleans. Cast them to strings first to interpret them as text aliases. punctuation mode is the exception — it strips punctuation only, preserving case and spaces.

Source code in skyulf-core/skyulf/preprocessing/cleaning/alias.py

class AliasReplacementApplier(BaseApplier):
    """Canonicalise near-duplicate text values in the resolved columns.

    The pandas and polars paths must agree value-for-value: each cell is
    normalised if it is text (lowercased, punctuation and spaces stripped) and looked up in
    the alias map, and a cell that matches nothing is left exactly as it was.
    Explicitly selected non-text values are unchanged, including numeric 0/1
    and booleans. Cast them to strings first to interpret them as text aliases.
    ``punctuation`` mode is the exception — it strips punctuation only,
    preserving case and spaces.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect resolved local alias rules without normalizing their saved values."""
        if not local_state_fields(
            raw,
            "alias_replacement",
            {"type", "columns", "alias_type", "custom_map"},
            allow_empty=True,
        ):
            return raw
        _columns(raw["columns"])
        if raw["alias_type"] not in ("boolean", "country", "custom", "punctuation"):
            raise ValueError("Unknown fitted alias type.")
        if type(raw["custom_map"]) is not dict:
            raise ValueError("Fitted custom aliases must be a dictionary.")
        for key, value in raw["custom_map"].items():
            local_scalar(key, "Custom aliases")
            local_scalar(value, "Custom aliases")
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe existing local row rules without promising engine or dtype parity."""
        if engine not in ("pandas", "polars"):
            return None
        AliasReplacementApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Dispatch the replacement to the pandas or polars path; ``y`` passes through."""
        return apply_dual_engine(
            X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

    @staticmethod
    def _apply_polars(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        """Map native text columns without coercing selected non-text columns."""
        valid = resolve_valid_columns(X, params.get("columns", []))
        if not valid:
            return X, _y

        alias_type = params.get("alias_type", "boolean")
        escaped_punct = re.escape(string.punctuation)

        if alias_type == "punctuation":
            # Punctuation mode strips punctuation only — case and spaces are
            # preserved, unlike the alias-mapping modes which normalise fully.
            exprs = [
                pl.col(col).cast(pl.String).str.replace_all(f"[{escaped_punct}]", "").alias(col)
                for col in valid
            ]
            return X.with_columns(exprs), _y

        mapping = _resolve_alias_mapping(alias_type, params.get("custom_map", {}))

        exprs = []
        text_columns = set(_auto_detect_text_columns(X))
        for col in valid:
            if col not in text_columns:
                continue
            if X.schema[col] == pl.Object:
                exprs.append(
                    pl.col(col)
                    .map_elements(
                        lambda value: _normalize_alias_pandas(value, mapping),
                        return_dtype=pl.Object,
                    )
                    .alias(col)
                )
                continue
            clean_expr = (
                pl.col(col)
                .cast(pl.String)
                .str.to_lowercase()
                .str.replace_all(f"[{escaped_punct}]", "")
                .str.replace_all(" ", "")
            )
            # Polars `replace(default=None)` returns null for non-matches, so we
            # coalesce back to the original value.
            mapped_expr = clean_expr.replace_strict(mapping, default=None)
            final_expr = pl.coalesce([mapped_expr, pl.col(col)])
            exprs.append(final_expr.alias(col))
        return X.with_columns(exprs), _y

    @staticmethod
    def _apply_pandas(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        """Apply text aliases without stringifying numbers in mixed object columns."""
        valid = resolve_valid_columns(X, params.get("columns", []))
        if not valid:
            return X, _y

        alias_type = params.get("alias_type", "boolean")

        if alias_type == "punctuation":
            df_out = X.copy()
            for col in valid:
                series = df_out[col]
                stripped = series.astype(str).str.translate(ALIAS_PUNCTUATION_TABLE)
                # astype(str) turns NaN into "nan"; restore the original NaN.
                df_out[col] = stripped.where(series.notna(), series)
            return df_out, _y

        mapping = _resolve_alias_mapping(alias_type, params.get("custom_map", {}))

        df_out = X.copy()
        text_columns = set(_auto_detect_text_columns(X))
        for col in valid:
            if col in text_columns:
                df_out[col] = _replace_pandas_aliases(df_out[col], mapping)
        return df_out, _y

apply(X, _y, params)

Dispatch the replacement to the pandas or polars path; y passes through.

Source code in skyulf-core/skyulf/preprocessing/cleaning/alias.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch the replacement to the pandas or polars path; ``y`` passes through."""
    return apply_dual_engine(
        X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe existing local row rules without promising engine or dtype parity.

Source code in skyulf-core/skyulf/preprocessing/cleaning/alias.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe existing local row rules without promising engine or dtype parity."""
    if engine not in ("pandas", "polars"):
        return None
    AliasReplacementApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect resolved local alias rules without normalizing their saved values.

Source code in skyulf-core/skyulf/preprocessing/cleaning/alias.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect resolved local alias rules without normalizing their saved values."""
    if not local_state_fields(
        raw,
        "alias_replacement",
        {"type", "columns", "alias_type", "custom_map"},
        allow_empty=True,
    ):
        return raw
    _columns(raw["columns"])
    if raw["alias_type"] not in ("boolean", "country", "custom", "punctuation"):
        raise ValueError("Unknown fitted alias type.")
    if type(raw["custom_map"]) is not dict:
        raise ValueError("Fitted custom aliases must be a dictionary.")
    for key, value in raw["custom_map"].items():
        local_scalar(key, "Custom aliases")
        local_scalar(value, "Custom aliases")
    return raw

AliasReplacementCalculator

Bases: BaseCalculator

Resolve alias-replacement config into an artifact; nothing is learned from data.

Source code in skyulf-core/skyulf/preprocessing/cleaning/alias.py

@NodeRegistry.register("AliasReplacement", AliasReplacementApplier)
@node_meta(
    id="AliasReplacement",
    name="Standardize Values",
    category="Cleaning",
    description="Standardize common variations in text values (e.g. Yes/No, Country names).",
    params={"columns": [], "domain": "boolean"},
    learns_from_data=False,
)
class AliasReplacementCalculator(BaseCalculator):
    """Resolve alias-replacement config into an artifact; nothing is learned from data."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Pass the input schema through, since only cell values are rewritten."""
        # Alias normalization replaces values in place; column set is preserved.
        return input_schema

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> AliasReplacementArtifact:  # pylint: disable=arguments-differ
        """Build the artifact, auto-detecting text columns when none are named.

        An explicit empty column selection yields an empty artifact, which
        makes the applier a no-op. Legacy ``alias_type`` spellings are remapped
        and ``custom_map`` keys are normalised exactly like runtime values, so
        a hand-written map still matches.
        """
        if user_picked_no_columns(config):
            return {}

        cols = resolve_columns(X, config, _auto_detect_text_columns)
        alias_type = _resolve_alias_type(config)
        custom_map = _normalize_alias_custom_map(
            config.get("custom_map") or config.get("custom_pairs", {})
        )

        return {
            "type": "alias_replacement",
            "columns": cols,
            "alias_type": alias_type,
            "custom_map": custom_map,
        }

fit(X, _y, config)

Build the artifact, auto-detecting text columns when none are named.

An explicit empty column selection yields an empty artifact, which makes the applier a no-op. Legacy alias_type spellings are remapped and custom_map keys are normalised exactly like runtime values, so a hand-written map still matches.

Source code in skyulf-core/skyulf/preprocessing/cleaning/alias.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> AliasReplacementArtifact:  # pylint: disable=arguments-differ
    """Build the artifact, auto-detecting text columns when none are named.

    An explicit empty column selection yields an empty artifact, which
    makes the applier a no-op. Legacy ``alias_type`` spellings are remapped
    and ``custom_map`` keys are normalised exactly like runtime values, so
    a hand-written map still matches.
    """
    if user_picked_no_columns(config):
        return {}

    cols = resolve_columns(X, config, _auto_detect_text_columns)
    alias_type = _resolve_alias_type(config)
    custom_map = _normalize_alias_custom_map(
        config.get("custom_map") or config.get("custom_pairs", {})
    )

    return {
        "type": "alias_replacement",
        "columns": cols,
        "alias_type": alias_type,
        "custom_map": custom_map,
    }

infer_output_schema(input_schema, config)

Pass the input schema through, since only cell values are rewritten.

Source code in skyulf-core/skyulf/preprocessing/cleaning/alias.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Pass the input schema through, since only cell values are rewritten."""
    # Alias normalization replaces values in place; column set is preserved.
    return input_schema

InvalidValueReplacementApplier

Bases: BaseApplier

Replace values violating the configured rule — and optionally ±inf — with a sentinel.

The pandas and polars paths must agree value-for-value: inf/-inf are replaced first when flagged, then the rule (negative, zero or custom_range, which honours a single bound when only one is given). A configuration without an effective rule or an inf flag is skipped outright, so its values and dtype survive untouched. Active operations require numeric columns on both engines; text must be explicitly converted before this node. Infinity-only cleanup preserves integer values and dtypes: integers cannot contain infinities and must not be widened to a floating-point sentinel. Active integer rules replacing values with None, NaN or pd.NA use nullable integers, retaining their width and exact values in full, singleton and empty requests. Numeric replacements must be exactly representable in the current integer dtype; fractional or out-of-range replacements require an explicit Casting step first. Integral floating range bounds are compared as integers to retain exact ordering.

Source code in skyulf-core/skyulf/preprocessing/cleaning/invalid_value.py

class InvalidValueReplacementApplier(BaseApplier):
    """Replace values violating the configured rule — and optionally ±inf — with a sentinel.

    The pandas and polars paths must agree value-for-value: ``inf``/``-inf``
    are replaced first when flagged, then the rule (``negative``, ``zero`` or
    ``custom_range``, which honours a single bound when only one is given). A
    configuration without an effective rule or an inf flag is skipped outright,
    so its values and dtype survive untouched. Active operations require numeric
    columns on both engines; text must be explicitly converted before this node.
    Infinity-only cleanup preserves integer values and dtypes: integers cannot
    contain infinities and must not be widened to a floating-point sentinel.
    Active integer rules replacing values with None, NaN or pd.NA use nullable integers,
    retaining their width and exact values in full, singleton and empty requests.
    Numeric replacements must be exactly representable in the current integer dtype;
    fractional or out-of-range replacements require an explicit Casting step first.
    Integral floating range bounds are compared as integers to retain exact ordering.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect fixed numeric rules while retaining NumPy sentinels and true no-ops."""
        if not local_state_fields(
            raw,
            "invalid_value_replacement",
            {
                "type",
                "columns",
                "replace_inf",
                "replace_neg_inf",
                "rule",
                "replacement",
                "value",
                "min_value",
                "max_value",
            },
            allow_empty=True,
        ):
            return raw
        _columns(raw["columns"])
        local_boolean(raw["replace_inf"], "replace_inf")
        local_boolean(raw["replace_neg_inf"], "replace_neg_inf")
        if raw["rule"] not in (None, "negative", "negative_to_nan", "zero", "custom_range"):
            raise ValueError("Unknown fitted invalid-value rule.")
        for name in ("min_value", "max_value"):
            value = raw[name]
            if value is not None and (
                isinstance(value, (bool, np.bool_)) or not isinstance(value, Real)
            ):
                raise ValueError("Fitted invalid-value bounds must be real numbers or None.")
        local_scalar(raw["replacement"], "Invalid-value replacements")
        local_scalar(raw["value"], "Invalid-value replacements")
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe row dependency while leaving actual schema parity to the probe."""
        if engine not in ("pandas", "polars"):
            return None
        InvalidValueReplacementApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Dispatch the rule to the pandas or polars path; ``y`` passes through."""
        if _has_numeric_operation(params):
            _validate_numeric_columns(X, resolve_valid_columns(X, params.get("columns", [])))
        return apply_dual_engine(
            X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

    @staticmethod
    def _apply_polars(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        valid = resolve_valid_columns(X, params.get("columns", []))
        if not valid or not _has_numeric_operation(params):
            return X, _y

        exprs = [_invalid_column_polars(col, X[col].dtype, params) for col in valid]
        return X.with_columns(exprs), _y

    @staticmethod
    def _apply_pandas_column(
        df_out: Any,
        col: str,
        replace_inf: bool,
        replace_neg_inf: bool,
        rule: Any,
        final_replacement: Any,
        min_value: Any,
        max_value: Any,
    ) -> None:
        """Normalize numeric data, replace infinities, and apply a rule in-place."""
        to_replace = []
        if replace_inf:
            to_replace.append(np.inf)
        if replace_neg_inf:
            to_replace.append(-np.inf)
        df_out[col] = pd.to_numeric(df_out[col], errors="coerce")
        if to_replace:
            df_out[col] = df_out[col].replace(to_replace, final_replacement)
        if pd.api.types.is_integer_dtype(df_out[col]) and _has_numeric_rule(
            rule, min_value, max_value
        ):
            final_replacement = _integer_rule_replacement(final_replacement, df_out[col].dtype)
            min_value = _integer_comparison_bound(min_value)
            max_value = _integer_comparison_bound(max_value)
        mask = _invalid_rule_pandas_mask(df_out[col], rule, min_value, max_value)
        if mask is not None:
            if pd.api.types.is_integer_dtype(df_out[col]) and _is_null_replacement(
                final_replacement
            ):
                df_out[col] = df_out[col].convert_dtypes()
            df_out.loc[mask, col] = final_replacement

    @staticmethod
    def _apply_pandas(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        valid = resolve_valid_columns(X, params.get("columns", []))
        if not valid or not _has_numeric_operation(params):
            return X, _y

        replace_inf = params.get("replace_inf", False)
        replace_neg_inf = params.get("replace_neg_inf", False)
        rule = params.get("rule")
        final_replacement = _resolve_invalid_replacement(params)
        min_value = params.get("min_value")
        max_value = params.get("max_value")

        df_out = X.copy()
        for col in valid:
            InvalidValueReplacementApplier._apply_pandas_column(
                df_out,
                col,
                replace_inf,
                replace_neg_inf,
                rule,
                final_replacement,
                min_value,
                max_value,
            )
        return df_out, _y

apply(X, _y, params)

Dispatch the rule to the pandas or polars path; y passes through.

Source code in skyulf-core/skyulf/preprocessing/cleaning/invalid_value.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch the rule to the pandas or polars path; ``y`` passes through."""
    if _has_numeric_operation(params):
        _validate_numeric_columns(X, resolve_valid_columns(X, params.get("columns", [])))
    return apply_dual_engine(
        X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe row dependency while leaving actual schema parity to the probe.

Source code in skyulf-core/skyulf/preprocessing/cleaning/invalid_value.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe row dependency while leaving actual schema parity to the probe."""
    if engine not in ("pandas", "polars"):
        return None
    InvalidValueReplacementApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect fixed numeric rules while retaining NumPy sentinels and true no-ops.

Source code in skyulf-core/skyulf/preprocessing/cleaning/invalid_value.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect fixed numeric rules while retaining NumPy sentinels and true no-ops."""
    if not local_state_fields(
        raw,
        "invalid_value_replacement",
        {
            "type",
            "columns",
            "replace_inf",
            "replace_neg_inf",
            "rule",
            "replacement",
            "value",
            "min_value",
            "max_value",
        },
        allow_empty=True,
    ):
        return raw
    _columns(raw["columns"])
    local_boolean(raw["replace_inf"], "replace_inf")
    local_boolean(raw["replace_neg_inf"], "replace_neg_inf")
    if raw["rule"] not in (None, "negative", "negative_to_nan", "zero", "custom_range"):
        raise ValueError("Unknown fitted invalid-value rule.")
    for name in ("min_value", "max_value"):
        value = raw[name]
        if value is not None and (
            isinstance(value, (bool, np.bool_)) or not isinstance(value, Real)
        ):
            raise ValueError("Fitted invalid-value bounds must be real numbers or None.")
    local_scalar(raw["replacement"], "Invalid-value replacements")
    local_scalar(raw["value"], "Invalid-value replacements")
    return raw

InvalidValueReplacementCalculator

Bases: BaseCalculator

Resolve invalid-value config into an artifact; nothing is learned from data.

Source code in skyulf-core/skyulf/preprocessing/cleaning/invalid_value.py

@NodeRegistry.register("InvalidValueReplacement", InvalidValueReplacementApplier)
@node_meta(
    id="InvalidValueReplacement",
    name="Replace Invalid Values",
    category="Cleaning",
    description="Replace specified values with nan.",
    params={"columns": []},
    learns_from_data=False,
)
class InvalidValueReplacementCalculator(BaseCalculator):
    """Resolve invalid-value config into an artifact; nothing is learned from data."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Pass the input schema through: sentinels are rewritten in place."""
        # Replaces invalid sentinel values with NaN in place; columns preserved.
        return input_schema

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> InvalidValueReplacementArtifact:  # pylint: disable=arguments-differ
        """Build the artifact, auto-detecting numeric columns when none are named.

        An explicit empty column selection yields an empty artifact, which
        makes the applier a no-op. UI convenience modes are canonicalised here
        rather than in each engine: ``percentage_bounds``/``age_bounds`` become
        a ``custom_range`` carrying their default bounds unless the user
        overrode them, and ``zero_to_nan`` becomes ``zero``.
        Active rules reject nonnumeric selected columns with ``ValueError``;
        the applier repeats this check when inference data changes dtype.
        """
        if user_picked_no_columns(config):
            return {}
        cols = resolve_columns(X, config, _numeric_columns)
        rule, min_value, max_value = _normalize_rule(
            config.get("rule") or config.get("mode"),
            config.get("min_value"),
            config.get("max_value"),
        )
        params: InvalidValueReplacementArtifact = {
            "type": "invalid_value_replacement",
            "columns": cols,
            "replace_inf": config.get("replace_inf", False),
            "replace_neg_inf": config.get("replace_neg_inf", False),
            "rule": rule,
            "replacement": config.get("replacement", np.nan),
            "value": config.get("value"),
            "min_value": min_value,
            "max_value": max_value,
        }
        if _has_numeric_operation(params):
            _validate_numeric_columns(X, cols)
        return params

fit(X, _y, config)

Build the artifact, auto-detecting numeric columns when none are named.

An explicit empty column selection yields an empty artifact, which makes the applier a no-op. UI convenience modes are canonicalised here rather than in each engine: percentage_bounds/age_bounds become a custom_range carrying their default bounds unless the user overrode them, and zero_to_nan becomes zero. Active rules reject nonnumeric selected columns with ValueError; the applier repeats this check when inference data changes dtype.

Source code in skyulf-core/skyulf/preprocessing/cleaning/invalid_value.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> InvalidValueReplacementArtifact:  # pylint: disable=arguments-differ
    """Build the artifact, auto-detecting numeric columns when none are named.

    An explicit empty column selection yields an empty artifact, which
    makes the applier a no-op. UI convenience modes are canonicalised here
    rather than in each engine: ``percentage_bounds``/``age_bounds`` become
    a ``custom_range`` carrying their default bounds unless the user
    overrode them, and ``zero_to_nan`` becomes ``zero``.
    Active rules reject nonnumeric selected columns with ``ValueError``;
    the applier repeats this check when inference data changes dtype.
    """
    if user_picked_no_columns(config):
        return {}
    cols = resolve_columns(X, config, _numeric_columns)
    rule, min_value, max_value = _normalize_rule(
        config.get("rule") or config.get("mode"),
        config.get("min_value"),
        config.get("max_value"),
    )
    params: InvalidValueReplacementArtifact = {
        "type": "invalid_value_replacement",
        "columns": cols,
        "replace_inf": config.get("replace_inf", False),
        "replace_neg_inf": config.get("replace_neg_inf", False),
        "rule": rule,
        "replacement": config.get("replacement", np.nan),
        "value": config.get("value"),
        "min_value": min_value,
        "max_value": max_value,
    }
    if _has_numeric_operation(params):
        _validate_numeric_columns(X, cols)
    return params

infer_output_schema(input_schema, config)

Pass the input schema through: sentinels are rewritten in place.

Source code in skyulf-core/skyulf/preprocessing/cleaning/invalid_value.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Pass the input schema through: sentinels are rewritten in place."""
    # Replaces invalid sentinel values with NaN in place; columns preserved.
    return input_schema

TextCleaningApplier

Bases: BaseApplier

Apply an ordered chain of text operations to the resolved columns.

The pandas and polars paths must agree value-for-value. Each column is coerced to string first, then every entry in operations is applied in list order so later operations see earlier results. The pandas coercion masks nulls back afterwards, otherwise astype(str) would turn them into the literal "nan". Unrecognized operation names raise ValueError, and an empty column set or empty operation list is a no-op.

Source code in skyulf-core/skyulf/preprocessing/cleaning/text.py

class TextCleaningApplier(BaseApplier):
    """Apply an ordered chain of text operations to the resolved columns.

    The pandas and polars paths must agree value-for-value. Each column is
    coerced to string first, then every entry in ``operations`` is applied in
    list order so later operations see earlier results. The pandas coercion
    masks nulls back afterwards, otherwise ``astype(str)`` would turn them into
    the literal ``"nan"``. Unrecognized operation names raise ``ValueError``, and
    an empty column set or empty operation list is a no-op.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect saved operations without rewriting their order, defaults or no-ops."""
        if not local_state_fields(
            raw, "text_cleaning", {"type", "columns", "operations"}, allow_empty=True
        ):
            return raw
        _columns(raw["columns"])
        operations = raw["operations"]
        if operations is None:
            return raw
        if type(operations) not in (list, tuple):
            raise ValueError("Fitted text operations must be a list or tuple.")
        _validate_text_operations(operations)
        for operation in operations:
            _validate_inference_text_operation(operation)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe local row operations while retaining engine-specific regex behavior."""
        if engine not in ("pandas", "polars"):
            return None
        TextCleaningApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Dispatch the operation chain to the pandas or polars path; ``y`` passes through."""
        return apply_dual_engine(
            X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

    @staticmethod
    def _apply_polars(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        cols = params.get("columns", [])
        operations = params.get("operations", [])
        valid = resolve_valid_columns(X, cols)
        if not valid or not operations:
            return X, _y
        _validate_text_operations(operations)

        exprs = []
        for col in valid:
            expr = pl.col(col).cast(pl.String)
            for op in operations:
                handler = _TEXT_OPS_POLARS[op["op"]]
                expr = handler(expr, op)
            exprs.append(expr.alias(col))
        return X.with_columns(exprs), _y

    @staticmethod
    def _apply_pandas(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        cols = params.get("columns", [])
        operations = params.get("operations", [])
        valid = resolve_valid_columns(X, cols)
        if not valid or not operations:
            return X, _y
        _validate_text_operations(operations)

        df_out = X.copy()
        for col in valid:
            if not pd.api.types.is_string_dtype(df_out[col]):
                null_mask = df_out[col].isna()
                df_out[col] = df_out[col].astype(str).mask(null_mask)
            series = df_out[col]
            for op in operations:
                handler = _TEXT_OPS_PANDAS[op["op"]]
                series = handler(series, op)
            df_out[col] = series
        return df_out, _y

apply(X, _y, params)

Dispatch the operation chain to the pandas or polars path; y passes through.

Source code in skyulf-core/skyulf/preprocessing/cleaning/text.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch the operation chain to the pandas or polars path; ``y`` passes through."""
    return apply_dual_engine(
        X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe local row operations while retaining engine-specific regex behavior.

Source code in skyulf-core/skyulf/preprocessing/cleaning/text.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe local row operations while retaining engine-specific regex behavior."""
    if engine not in ("pandas", "polars"):
        return None
    TextCleaningApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect saved operations without rewriting their order, defaults or no-ops.

Source code in skyulf-core/skyulf/preprocessing/cleaning/text.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect saved operations without rewriting their order, defaults or no-ops."""
    if not local_state_fields(
        raw, "text_cleaning", {"type", "columns", "operations"}, allow_empty=True
    ):
        return raw
    _columns(raw["columns"])
    operations = raw["operations"]
    if operations is None:
        return raw
    if type(operations) not in (list, tuple):
        raise ValueError("Fitted text operations must be a list or tuple.")
    _validate_text_operations(operations)
    for operation in operations:
        _validate_inference_text_operation(operation)
    return raw

TextCleaningCalculator

Bases: BaseCalculator

Resolve text-cleaning config into an artifact; nothing is learned from data.

Source code in skyulf-core/skyulf/preprocessing/cleaning/text.py

@NodeRegistry.register("TextCleaning", TextCleaningApplier)
@node_meta(
    id="TextCleaning",
    name="Text Cleaning",
    category="Cleaning",
    description="Clean text data (trim, case conversion, remove special chars).",
    params={"columns": [], "operations": []},
    learns_from_data=False,
)
class TextCleaningCalculator(BaseCalculator):
    """Resolve text-cleaning config into an artifact; nothing is learned from data."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Pass the input schema through: string values are rewritten in place."""
        # Text cleaning rewrites string values in place; column set is preserved.
        return input_schema

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> TextCleaningArtifact:  # pylint: disable=arguments-differ
        """Build the artifact, auto-detecting text columns when none are named.

        Two cases short-circuit to an empty artifact: an explicit
        ``columns: []`` from the UI, and auto-detection finding no text columns
        at all. Both make the applier a no-op rather than coercing numeric
        columns to text.
        """
        if user_picked_no_columns(config):
            return {}
        cols = resolve_columns(X, config, _auto_detect_text_columns)
        if not cols:
            return {}
        return {
            "type": "text_cleaning",
            "columns": cols,
            "operations": config.get("operations", []),
        }

fit(X, _y, config)

Build the artifact, auto-detecting text columns when none are named.

Two cases short-circuit to an empty artifact: an explicit columns: [] from the UI, and auto-detection finding no text columns at all. Both make the applier a no-op rather than coercing numeric columns to text.

Source code in skyulf-core/skyulf/preprocessing/cleaning/text.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> TextCleaningArtifact:  # pylint: disable=arguments-differ
    """Build the artifact, auto-detecting text columns when none are named.

    Two cases short-circuit to an empty artifact: an explicit
    ``columns: []`` from the UI, and auto-detection finding no text columns
    at all. Both make the applier a no-op rather than coercing numeric
    columns to text.
    """
    if user_picked_no_columns(config):
        return {}
    cols = resolve_columns(X, config, _auto_detect_text_columns)
    if not cols:
        return {}
    return {
        "type": "text_cleaning",
        "columns": cols,
        "operations": config.get("operations", []),
    }

infer_output_schema(input_schema, config)

Pass the input schema through: string values are rewritten in place.

Source code in skyulf-core/skyulf/preprocessing/cleaning/text.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Pass the input schema through: string values are rewritten in place."""
    # Text cleaning rewrites string values in place; column set is preserved.
    return input_schema

ValueReplacementApplier

Bases: BaseApplier

Replace configured values in the resolved columns, leaving other cells alone.

The pandas and polars paths must agree value-for-value. A mapping wins over to_replace/value when both are configured, and may be flat (one map applied to every column) or nested ({column: {old: new}}). Mapping and to_replace keys are coerced to the column's dtype before lookup. A list of keys can share one replacement or have a same-length list of replacements. A column selection resolving to nothing is a no-op.

Numeric rules on integer columns must fit the existing integer dtype; fractional, infinite and out-of-range replacements require an explicit Casting step. Missing replacements use native nullable integers. These checks use configured rules before matching rows, including empty requests. Nonnumeric replacements retain their native engine behavior.

Source code in skyulf-core/skyulf/preprocessing/cleaning/value_replacement.py

class ValueReplacementApplier(BaseApplier):
    """Replace configured values in the resolved columns, leaving other cells alone.

    The pandas and polars paths must agree value-for-value. A ``mapping`` wins
    over ``to_replace``/``value`` when both are configured, and may be flat
    (one map applied to every column) or nested (``{column: {old: new}}``).
    Mapping and ``to_replace`` keys are coerced to the column's dtype before
    lookup. A list of keys can share one replacement or have a same-length
    list of replacements. A column selection resolving to nothing is a no-op.

    Numeric rules on integer columns must fit the existing integer dtype;
    fractional, infinite and out-of-range replacements require an explicit
    Casting step. Missing replacements use native nullable integers. These
    checks use configured rules before matching rows, including empty requests.
    Nonnumeric replacements retain their native engine behavior.
    """

    @staticmethod
    def validate_inference_state(raw: dict) -> dict:
        """Inspect scalar, list, tuple and dictionary state without coercing saved rules."""
        local_state_fields(
            raw, "value_replacement", {"type", "columns", "mapping", "to_replace", "value"}
        )
        _columns(raw["columns"])
        _validate_replacement_rules(raw)
        return raw

    @staticmethod
    def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
        """Describe row dependency without claiming chunk-invariant dtype inference."""
        if engine not in ("pandas", "polars"):
            return None
        ValueReplacementApplier.validate_inference_state(state)
        return ExecutionCapability(engine, "apply", "local", "preserve", "row")

    @apply_method
    def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
        """Dispatch the replacement to the pandas or polars path; ``y`` passes through."""
        return apply_dual_engine(
            X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
        )

    @staticmethod
    def _apply_polars(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        valid = resolve_valid_columns(X, params.get("columns", []))
        if not valid:
            return X, _y
        exprs = _value_replacement_exprs_polars(
            X,
            valid,
            params.get("mapping"),
            params.get("to_replace"),
            params.get("value"),
        )
        return (X.with_columns(exprs) if exprs else X), _y

    @staticmethod
    def _apply_pandas(X: Any, _y: Any, params: dict[str, Any]) -> tuple[Any, Any]:
        valid = resolve_valid_columns(X, params.get("columns", []))
        if not valid:
            return X, _y
        df_out = X.copy()
        df_out = _apply_value_replacement_pandas(
            df_out,
            valid,
            params.get("mapping"),
            params.get("to_replace"),
            params.get("value"),
        )
        return df_out, _y

apply(X, _y, params)

Dispatch the replacement to the pandas or polars path; y passes through.

Source code in skyulf-core/skyulf/preprocessing/cleaning/value_replacement.py

@apply_method
def apply(self, X: Any, _y: Any, params: dict[str, Any]) -> Any:  # pylint: disable=arguments-differ
    """Dispatch the replacement to the pandas or polars path; ``y`` passes through."""
    return apply_dual_engine(
        X, params, {"polars": self._apply_polars, "pandas": self._apply_pandas}
    )

inference_capability(state, *, engine) staticmethod

Describe row dependency without claiming chunk-invariant dtype inference.

Source code in skyulf-core/skyulf/preprocessing/cleaning/value_replacement.py

@staticmethod
def inference_capability(state: dict, *, engine: str) -> ExecutionCapability | None:
    """Describe row dependency without claiming chunk-invariant dtype inference."""
    if engine not in ("pandas", "polars"):
        return None
    ValueReplacementApplier.validate_inference_state(state)
    return ExecutionCapability(engine, "apply", "local", "preserve", "row")

validate_inference_state(raw) staticmethod

Inspect scalar, list, tuple and dictionary state without coercing saved rules.

Source code in skyulf-core/skyulf/preprocessing/cleaning/value_replacement.py

@staticmethod
def validate_inference_state(raw: dict) -> dict:
    """Inspect scalar, list, tuple and dictionary state without coercing saved rules."""
    local_state_fields(
        raw, "value_replacement", {"type", "columns", "mapping", "to_replace", "value"}
    )
    _columns(raw["columns"])
    _validate_replacement_rules(raw)
    return raw

ValueReplacementCalculator

Bases: BaseCalculator

Resolve value-replacement config into an artifact; nothing is learned from data.

Source code in skyulf-core/skyulf/preprocessing/cleaning/value_replacement.py

@NodeRegistry.register("ValueReplacement", ValueReplacementApplier)
@node_meta(
    id="ValueReplacement",
    name="Replace Values",
    category="Cleaning",
    description="Replace specified values with new values.",
    params={"columns": [], "mapping": {}},
    learns_from_data=False,
)
class ValueReplacementCalculator(BaseCalculator):
    """Resolve value-replacement config into an artifact; nothing is learned from data."""

    def infer_output_schema(
        self, input_schema: SkyulfSchema, config: dict[str, Any]
    ) -> SkyulfSchema:
        """Preserve columns and report only output dtypes guaranteed by the active rules.

        Integer null rules expose their native nullable type. Data-dependent or
        unsupported replacements omit the affected dtype; consumers can still
        validate column references and display the type as unknown.
        """
        dtypes = dict(input_schema.dtypes)
        for column in config.get("columns") or []:
            if column not in dtypes:
                continue
            try:
                label = _replacement_output_dtype(
                    dtypes[column], _schema_replacement_rules(config, column)
                )
            except (TypeError, ValueError, OverflowError):
                label = None
            if label is None:
                dtypes.pop(column)
            else:
                dtypes[column] = label
        if dtypes == input_schema.dtypes:
            return input_schema
        return SkyulfSchema.from_columns(input_schema.columns, dtypes)

    @fit_method
    def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> ValueReplacementArtifact:  # pylint: disable=arguments-differ
        """Build the artifact, folding the UI's ``replacements`` pairs into a flat mapping.

        A non-empty ``replacements`` list overrides any configured ``mapping``.
        ``to_replace``/``value`` are carried through untouched for the applier
        to fall back on when no mapping survived.
        """
        cols = resolve_columns(X, config)
        return {
            "type": "value_replacement",
            "columns": cols,
            "mapping": _configured_mapping(config),
            "to_replace": config.get("to_replace"),
            "value": config.get("value"),
        }

fit(X, _y, config)

Build the artifact, folding the UI's replacements pairs into a flat mapping.

A non-empty replacements list overrides any configured mapping. to_replace/value are carried through untouched for the applier to fall back on when no mapping survived.

Source code in skyulf-core/skyulf/preprocessing/cleaning/value_replacement.py

@fit_method
def fit(self, X: Any, _y: Any, config: dict[str, Any]) -> ValueReplacementArtifact:  # pylint: disable=arguments-differ
    """Build the artifact, folding the UI's ``replacements`` pairs into a flat mapping.

    A non-empty ``replacements`` list overrides any configured ``mapping``.
    ``to_replace``/``value`` are carried through untouched for the applier
    to fall back on when no mapping survived.
    """
    cols = resolve_columns(X, config)
    return {
        "type": "value_replacement",
        "columns": cols,
        "mapping": _configured_mapping(config),
        "to_replace": config.get("to_replace"),
        "value": config.get("value"),
    }

infer_output_schema(input_schema, config)

Preserve columns and report only output dtypes guaranteed by the active rules.

Integer null rules expose their native nullable type. Data-dependent or unsupported replacements omit the affected dtype; consumers can still validate column references and display the type as unknown.

Source code in skyulf-core/skyulf/preprocessing/cleaning/value_replacement.py

def infer_output_schema(
    self, input_schema: SkyulfSchema, config: dict[str, Any]
) -> SkyulfSchema:
    """Preserve columns and report only output dtypes guaranteed by the active rules.

    Integer null rules expose their native nullable type. Data-dependent or
    unsupported replacements omit the affected dtype; consumers can still
    validate column references and display the type as unknown.
    """
    dtypes = dict(input_schema.dtypes)
    for column in config.get("columns") or []:
        if column not in dtypes:
            continue
        try:
            label = _replacement_output_dtype(
                dtypes[column], _schema_replacement_rules(config, column)
            )
        except (TypeError, ValueError, OverflowError):
            label = None
        if label is None:
            dtypes.pop(column)
        else:
            dtypes[column] = label
    if dtypes == input_schema.dtypes:
        return input_schema
    return SkyulfSchema.from_columns(input_schema.columns, dtypes)