torch

TorchDataset(inputs, outputs=None)

Bases: Dataset[tuple[Tensor, Tensor]]

Dataset for PyTorch.

Initialize the TorchDataset.

Parameters:

Name	Type	Description	Default
inputs	DataFrame	The input data.	required
outputs	DataFrame | None	The output data. Defaults to None.	None

Source code in src/flowcean/torch/dataset.py

def __init__(
    self,
    inputs: pl.DataFrame,
    outputs: pl.DataFrame | None = None,
) -> None:
    """Initialize the TorchDataset.

    Args:
        inputs: The input data.
        outputs: The output data. Defaults to None.
    """
    super().__init__()
    self.inputs = inputs
    self.outputs = outputs

__len__()

Return the length of the dataset.

Source code in src/flowcean/torch/dataset.py

def __len__(self) -> int:
    """Return the length of the dataset."""
    return len(self.inputs)

__getitem__(item)

Return the item at the given index.

Parameters:

Name	Type	Description	Default
item	int	The index of the item to return.	required

Returns:

Type	Description
tuple[Tensor, Tensor]	The inputs and outputs at the given index.

Source code in src/flowcean/torch/dataset.py

def __getitem__(self, item: int) -> tuple[Tensor, Tensor]:
    """Return the item at the given index.

    Args:
        item: The index of the item to return.

    Returns:
        The inputs and outputs at the given index.
    """
    inputs = Tensor(self.inputs.row(item))
    if self.outputs is None:
        outputs = Tensor([])
    else:
        outputs = Tensor(self.outputs.row(item))
    return inputs, outputs

LightningLearner(module, num_workers=None, batch_size=32, max_epochs=100)

Bases: SupervisedLearner

A learner that uses PyTorch Lightning.

Initialize the learner.

Parameters:

Name	Type	Description	Default
module	LightningModule	The PyTorch Lightning module.	required
num_workers	int | None	The number of workers to use for the DataLoader.	None
batch_size	int	The batch size to use for training.	32
max_epochs	int	The maximum number of epochs to train for.	100

Source code in src/flowcean/torch/lightning_learner.py

def __init__(
    self,
    module: lightning.LightningModule,
    num_workers: int | None = None,
    batch_size: int = 32,
    max_epochs: int = 100,
) -> None:
    """Initialize the learner.

    Args:
        module: The PyTorch Lightning module.
        num_workers: The number of workers to use for the DataLoader.
        batch_size: The batch size to use for training.
        max_epochs: The maximum number of epochs to train for.
    """
    self.module = module
    self.num_workers = num_workers or os.cpu_count() or 0
    self.max_epochs = max_epochs
    self.batch_size = batch_size
    self.optimizer = None

MultilayerPerceptron(learning_rate, input_size, output_size, hidden_dimensions=None)

Bases: LightningModule

A multilayer perceptron.

Initialize the model.

Parameters:

Name	Type	Description	Default
learning_rate	float	The learning rate.	required
input_size	int	The size of the input.	required
output_size	int	The size of the output.	required
hidden_dimensions	list[int] | None	The dimensions of the hidden layers.	None

Source code in src/flowcean/torch/lightning_learner.py

def __init__(
    self,
    learning_rate: float,
    input_size: int,
    output_size: int,
    hidden_dimensions: list[int] | None = None,
) -> None:
    """Initialize the model.

    Args:
        learning_rate: The learning rate.
        input_size: The size of the input.
        output_size: The size of the output.
        hidden_dimensions: The dimensions of the hidden layers.
    """
    super().__init__()
    if hidden_dimensions is None:
        hidden_dimensions = []
    self.save_hyperparameters()
    self.learning_rate = learning_rate

    layers: list[Module] = []
    hidden_size = input_size
    for dimension in hidden_dimensions:
        layers.extend(
            (
                torch.nn.Linear(hidden_size, dimension),
                torch.nn.LeakyReLU(),
            ),
        )
        hidden_size = dimension
    layers.append(torch.nn.Linear(hidden_size, output_size))
    self.model = torch.nn.Sequential(*layers)

LinearRegression(input_size, output_size, learning_rate=0.001, loss=None)

Bases: SupervisedIncrementalLearner

Linear regression learner.

Initialize the learner.

Parameters:

Name	Type	Description	Default
input_size	int	The size of the input.	required
output_size	int	The size of the output.	required
learning_rate	float	The learning rate.	0.001
loss	Module | None	The loss function.	None

Source code in src/flowcean/torch/linear_regression.py

def __init__(
    self,
    input_size: int,
    output_size: int,
    learning_rate: float = 1e-3,
    loss: nn.Module | None = None,
) -> None:
    """Initialize the learner.

    Args:
        input_size: The size of the input.
        output_size: The size of the output.
        learning_rate: The learning rate.
        loss: The loss function.
    """
    self.model = nn.Linear(input_size, output_size)
    self.loss = loss or nn.MSELoss()
    self.optimizer = SGD(
        self.model.parameters(),
        lr=learning_rate,
    )

PyTorchModel(module, output_names, batch_size=32, num_workers=1)

Bases: Model

PyTorch model wrapper.

Initialize the model.

Parameters:

Name	Type	Description	Default
module	Module	The PyTorch module.	required
output_names	list[str]	The names of the output columns.	required
batch_size	int	The batch size to use for predictions.	32
num_workers	int	The number of workers to use for the DataLoader.	1

Source code in src/flowcean/torch/model.py

def __init__(
    self,
    module: Module,
    output_names: list[str],
    batch_size: int = 32,
    num_workers: int = 1,
) -> None:
    """Initialize the model.

    Args:
        module: The PyTorch module.
        output_names: The names of the output columns.
        batch_size: The batch size to use for predictions.
        num_workers: The number of workers to use for the DataLoader.
    """
    self.module = module
    self.output_names = output_names
    self.batch_size = batch_size
    self.num_workers = num_workers