utils ¶

def check_estimator(model):
    """Check if a model adheres to `river`'s conventions.
    This will run a series of unit tests. The nature of the unit tests
    depends on the type of model.
    Parameters
    ----------
    model
    """
    for check in yield_checks(model):
        if check.__name__ in model._unit_test_skips():
            continue
        check(copy.deepcopy(model))  # todo change to clone

    for check in yield_deep_checks(model):
        if check.__name__ in model._unit_test_skips():
            continue
        check(copy.deepcopy(model))  # todo change to clone

def deque2rolling_tensor(
    window: Deque,
    device="cpu",
    dtype=torch.float32,
) -> torch.Tensor:
    """
    Convert a dictionary to a rolling tensor.

    Parameters
    ----------
    window
        Rolling window.
    device
        Device.
    dtype
        Dtype.

    Returns
    -------
        torch.Tensor
    """
    output = torch.tensor(window, device=device, dtype=dtype)
    return torch.unsqueeze(output, 1)

def df2tensor(
    X: pd.DataFrame,
    features: SortedSet,
    default_value: float = 0.0,
    device="cpu",
    dtype=torch.float32,
) -> torch.Tensor:
    """
    Convert a dataframe to a tensor.
    Parameters
    ----------
    X
        Dataframe.
    features:
        Set of possible features.
    default_value:
        Value to use for features not present in x.
    device
        Device.
    dtype
        Dtype.

    Returns
    -------
        torch.Tensor
    """
    # Work on a shallow copy to avoid mutating caller's DataFrame
    X_local = X.copy()
    for feature in features:
        if feature not in X_local.columns:
            X_local[feature] = default_value
    cols = list(features)
    # Replace NaNs in selected columns by default_value
    if len(cols) > 0:
        X_local[cols] = X_local[cols].fillna(default_value)
    return torch.tensor(X_local[cols].values, device=device, dtype=dtype)

def dict2tensor(
    x: dict,
    features: SortedSet,
    default_value: float = 0.0,
    device: str = "cpu",
    dtype: torch.dtype = torch.float32,
) -> torch.Tensor:
    """
    Convert a dictionary to a tensor.

    Parameters
    ----------
    x
        Dictionary.
    features:
        Set of possible features.
    default_value:
        Value to use for features not present in x.
    device
        Device.
    dtype
        Dtype.

    Returns
    -------
        torch.Tensor
    """
    row = []
    for feature in features:
        val = x.get(feature, default_value)
        # Replace None/NaN with default_value to prevent NaNs propagating
        if val is None:
            val = default_value
        elif isinstance(val, float) and math.isnan(val):
            val = default_value
        row.append(val)
    return torch.tensor([row], device=device, dtype=dtype)

def float2tensor(
    y: Union[float, int, RegTarget, dict], device="cpu", dtype=torch.float32
) -> torch.Tensor:
    """
    Convert a float to a tensor.

    Parameters
    ----------
    y
        Float.
    device
        Device.
    dtype
        Dtype.

    Returns
    -------
        torch.Tensor
    """
    if isinstance(y, dict):
        return torch.tensor([list(y.values())], device=device, dtype=dtype)
    else:
        return torch.tensor([[y]], device=device, dtype=dtype)

def get_activation_fn(activation_fn: Union[str, Callable]) -> Callable:
    """Returns the requested activation function as a nn.Module class.

    Parameters
    ----------
    activation_fn
        The activation function to fetch. Can be a string or a nn.Module class.

    Returns
    -------
    Callable
        The class of the requested activation function.
    """
    err = ValueError(
        BASE_PARAM_ERROR.format("activation function", activation_fn, "nn.Module")
    )
    if isinstance(activation_fn, str):
        try:
            activation_fn = ACTIVATION_FNS[activation_fn]
        except KeyError:
            raise err
    elif not isinstance(activation_fn(), nn.Module):
        raise err
    return activation_fn

def get_init_fn(init_fn):
    """Returns the requested init function.

    Parameters
    ----------
    init_fn
        The init function to fetch. Must be one of ["xavier_uniform",
        "uniform", "kaiming_uniform"].

    Returns
    -------
    Callable
        The class of the requested activation function.
    """
    init_fn_ = INIT_FNS.get(init_fn, "xavier_uniform")
    if init_fn.startswith("xavier"):

        def result(weight, activation_fn):
            return init_fn_(weight, gain=nn.init.calculate_gain(activation_fn))

    elif init_fn.startswith("kaiming"):

        def result(weight, activation_fn):
            return init_fn_(weight, nonlinearity=activation_fn)

    elif init_fn == "uniform":

        def result(weight, activation_fn):
            return 0

    else:

        def result(weight, activation_fn):
            return init_fn_(weight)

    return result

def get_loss_fn(loss_fn: Union[str, Callable]) -> Callable:
    """Returns the requested loss function as a function.

    Parameters
    ----------
    loss_fn
        The loss function to fetch. Can be a string or a function.

    Returns
    -------
    Callable
        The function of the requested loss function.
    """
    err = ValueError(BASE_PARAM_ERROR.format("loss function", loss_fn, "function"))
    if isinstance(loss_fn, str):
        try:
            loss_fn = LOSS_FNS[loss_fn]
        except KeyError:
            raise err
    elif not callable(loss_fn):
        raise err
    return loss_fn

def get_optim_fn(optim_fn: Union[str, Callable]) -> Callable:
    """Returns the requested optimizer as a nn.Module class.

    Parameters
    ----------
    optim_fn
        The optimizer to fetch. Can be a string or a nn.Module class.


    Returns
    -------
    Callable
        The class of the requested optimizer.
    """
    err = ValueError(BASE_PARAM_ERROR.format("optimizer", optim_fn, "nn.Module"))
    if isinstance(optim_fn, str):
        try:
            optim_fn = OPTIMIZER_FNS[optim_fn]
        except KeyError:
            raise err

    elif not isinstance(
        optim_fn(params=[torch.empty(1)], lr=1e-3), torch.optim.Optimizer
    ):
        raise err
    return optim_fn

def labels2onehot(
    y: Union[base.typing.ClfTarget, pd.Series],
    classes: SortedSet,
    n_classes: Optional[int] = None,
    device="cpu",
    dtype=torch.float32,
) -> torch.Tensor:
    """
    Convert a label or a list of labels to a one-hot encoded tensor.

    Parameters
    ----------
    y
        Label or list of labels.
    classes
        Classes.
    n_classes
        Number of classes.
    device
        Device.
    dtype
        Dtype.

    Returns
    -------
        torch.Tensor
    """

    def get_class_index(label):
        """Retrieve class index with type checking and conversion."""
        if isinstance(label, float):  # Handle float case
            if label.is_integer():  # Convert to int if it's an integer-like float
                label = int(label)
            else:
                raise ValueError(
                    f"Label {label} is a float and cannot be mapped to a class index."
                )
        return classes.index(label)

    if n_classes is None:
        n_classes = len(classes)
    if isinstance(y, pd.Series):
        onehot = torch.zeros(len(y), n_classes, device=device, dtype=dtype)
        pos_idcs = [get_class_index(y_i) for y_i in y]
        for i, pos_idx in enumerate(pos_idcs):
            if isinstance(pos_idx, int) and pos_idx < n_classes:
                onehot[i, pos_idx] = 1
    else:
        onehot = torch.zeros(1, n_classes, device=device, dtype=dtype)
        pos_idx = classes.index(y)
        if isinstance(pos_idx, int) and pos_idx < n_classes:
            onehot[0, pos_idx] = 1

    return onehot