Skip to content

tensor_conversion

Functions:

Name Description
deque2rolling_tensor

Convert a dictionary to a rolling tensor.

df2tensor

Convert a dataframe to a tensor.

dict2tensor

Convert a dictionary to a tensor.

float2tensor

Convert a float to a tensor.

labels2onehot

Convert a label or a list of labels to a one-hot encoded tensor.

deque2rolling_tensor

deque2rolling_tensor(
    window: Deque, device="cpu", dtype=float32
) -> Tensor

Convert a dictionary to a rolling tensor.

Parameters:

Name Type Description Default
window Deque

Rolling window.

required
device

Device.

'cpu'
dtype

Dtype.

float32

Returns:

Type Description
torch.Tensor
Source code in deep_river/utils/tensor_conversion.py
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)

df2tensor

df2tensor(
    X: DataFrame,
    features: SortedSet,
    default_value: float = 0,
    device="cpu",
    dtype=float32,
) -> Tensor

Convert a dataframe to a tensor.

Parameters:

Name Type Description Default
X DataFrame

Dataframe.

required
features SortedSet

Set of possible features.

required
default_value float

Value to use for features not present in x.

0
device

Device.

'cpu'
dtype

Dtype.

float32

Returns:

Type Description
torch.Tensor
Source code in deep_river/utils/tensor_conversion.py
def df2tensor(
    X: pd.DataFrame,
    features: SortedSet,
    default_value: float = 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
    """
    for feature in features:
        if feature not in X.columns:
            X[feature] = default_value
    return torch.tensor(X[list(features)].values, device=device, dtype=dtype)

dict2tensor

dict2tensor(
    x: dict,
    features: SortedSet,
    default_value: float = 0,
    device: str = "cpu",
    dtype: dtype = float32,
) -> Tensor

Convert a dictionary to a tensor.

Parameters:

Name Type Description Default
x dict

Dictionary.

required
features SortedSet

Set of possible features.

required
default_value float

Value to use for features not present in x.

0
device str

Device.

'cpu'
dtype dtype

Dtype.

float32

Returns:

Type Description
torch.Tensor
Source code in deep_river/utils/tensor_conversion.py
def dict2tensor(
    x: dict,
    features: SortedSet,
    default_value: float = 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
    """
    return torch.tensor(
        [[x.get(feature, default_value) for feature in features]],
        device=device,
        dtype=dtype,
    )

float2tensor

float2tensor(
    y: Union[float, int, RegTarget, dict],
    device="cpu",
    dtype=float32,
) -> Tensor

Convert a float to a tensor.

Parameters:

Name Type Description Default
y Union[float, int, RegTarget, dict]

Float.

required
device

Device.

'cpu'
dtype

Dtype.

float32

Returns:

Type Description
torch.Tensor
Source code in deep_river/utils/tensor_conversion.py
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)

labels2onehot

labels2onehot(
    y: Union[ClfTarget, Series],
    classes: SortedSet[ClfTarget],
    n_classes: Optional[int] = None,
    device="cpu",
    dtype=float32,
) -> Tensor

Convert a label or a list of labels to a one-hot encoded tensor.

Parameters:

Name Type Description Default
y Union[ClfTarget, Series]

Label or list of labels.

required
classes SortedSet[ClfTarget]

Classes.

required
n_classes Optional[int]

Number of classes.

None
device

Device.

'cpu'
dtype

Dtype.

float32

Returns:

Type Description
torch.Tensor
Source code in deep_river/utils/tensor_conversion.py
def labels2onehot(
    y: Union[base.typing.ClfTarget, pd.Series],
    classes: SortedSet[base.typing.ClfTarget],
    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