Skip to content

nn_many_to_many_lstm_regressor

ManyToManyLSTM

Bases: Module

A Many-to-Many LSTM model for sequence-to-sequence regression tasks.

This model uses an LSTM layer followed by a fully connected layer and an output layer.

Parameters:

Name Type Description Default
input_size int

The number of input features.

 required
output_size int

The number of output features. Defaults to 1.

 1
lstm_units int

The number of units in the LSTM layer. Defaults to 256.

 256
fc_units int

The number of units in the fully connected layer. Defaults to 256.

 256
activation_fct Module

The activation function to use after the fully connected layer. Defaults to nn.ReLU().

 ReLU()
dropout float

The dropout probability. Defaults to 0.0.

 0.0
bidirectional bool

Whether the LSTM is bidirectional. Defaults to True.

 True
num_layers int

The number of LSTM layers. Defaults to 1.

 1

Examples:

>>> from spotpython.light.regression.nn_many_to_many_lstm_regressor import ManyToManyLSTM
>>> import torch
>>> model = ManyToManyLSTM(input_size=10, output_size=1)
>>> x = torch.randn(16, 10, 10)  # Batch of 16 sequences, each of length 10 with 10 features
>>> lengths = torch.tensor([10] * 16)  # All sequences have length 10
>>> output = model(x, lengths)
>>> print(output.shape)  # Output shape: (16, 10, 1)
Source code in spotpython/light/regression/nn_many_to_many_lstm_regressor.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
class ManyToManyLSTM(nn.Module):
    """A Many-to-Many LSTM model for sequence-to-sequence regression tasks.

    This model uses an LSTM layer followed by a fully connected layer and an output layer.

    Args:
        input_size (int): The number of input features.
        output_size (int): The number of output features. Defaults to 1.
        lstm_units (int): The number of units in the LSTM layer. Defaults to 256.
        fc_units (int): The number of units in the fully connected layer. Defaults to 256.
        activation_fct (nn.Module): The activation function to use after the fully connected layer. Defaults to nn.ReLU().
        dropout (float): The dropout probability. Defaults to 0.0.
        bidirectional (bool): Whether the LSTM is bidirectional. Defaults to True.
        num_layers (int): The number of LSTM layers. Defaults to 1.

    Examples:
        >>> from spotpython.light.regression.nn_many_to_many_lstm_regressor import ManyToManyLSTM
        >>> import torch
        >>> model = ManyToManyLSTM(input_size=10, output_size=1)
        >>> x = torch.randn(16, 10, 10)  # Batch of 16 sequences, each of length 10 with 10 features
        >>> lengths = torch.tensor([10] * 16)  # All sequences have length 10
        >>> output = model(x, lengths)
        >>> print(output.shape)  # Output shape: (16, 10, 1)
    """

    def __init__(
        self,
        input_size,
        output_size=1,
        lstm_units=256,
        fc_units=256,
        activation_fct=nn.ReLU(),
        dropout=0.0,
        bidirectional=True,
        num_layers=1,
    ):
        super(ManyToManyLSTM, self).__init__()
        self.lstm_layer = nn.LSTM(
            input_size=input_size,
            hidden_size=lstm_units,
            num_layers=num_layers,
            batch_first=True,
            bidirectional=bidirectional,
            dropout=dropout if num_layers > 1 else 0.0,
        )
        if bidirectional:
            lstm_units = lstm_units * 2
        self.fc = nn.Linear(lstm_units, fc_units)
        self.dropout = nn.Dropout(dropout)
        self.output_layer = nn.Linear(fc_units, output_size)
        self.activation_fct = activation_fct

    def forward(self, x, lengths) -> torch.Tensor:
        """Forward pass of the ManyToManyLSTM model.

        Args:
            x (torch.Tensor): Input tensor of shape (batch_size, seq_len, input_size).
            lengths (torch.Tensor): Tensor containing the lengths of each sequence in the batch.

        Returns:
            torch.Tensor: Output tensor of shape (batch_size, seq_len, output_size).

        Raises:
            ValueError: If the input tensor is empty or if the lengths tensor is empty.
            RuntimeError: If the lengths tensor does not match the batch size of the input tensor.
        """
        if x.size(0) == 0 or lengths.size(0) == 0:
            raise ValueError("Input tensor or lengths tensor is empty.")
        if x.size(0) != lengths.size(0):
            raise RuntimeError(f"Batch size of input tensor ({x.size(0)}) and lengths tensor ({lengths.size(0)}) must match.")

        x = pack_padded_sequence(x, lengths, batch_first=True, enforce_sorted=False)
        packed_output, _ = self.lstm_layer(x)
        x, _ = pad_packed_sequence(packed_output, batch_first=True)
        x = self.dropout(x)
        x = self.fc(x)
        x = self.activation_fct(x)
        x = self.output_layer(x)
        return x

forward(x, lengths)

Forward pass of the ManyToManyLSTM model.

Parameters:

Name Type Description Default
x Tensor

Input tensor of shape (batch_size, seq_len, input_size).

 required
lengths Tensor

Tensor containing the lengths of each sequence in the batch.

 required

Returns:

Type Description
Tensor

torch.Tensor: Output tensor of shape (batch_size, seq_len, output_size).

Raises:

Type Description
ValueError

If the input tensor is empty or if the lengths tensor is empty.
RuntimeError

If the lengths tensor does not match the batch size of the input tensor.
Source code in spotpython/light/regression/nn_many_to_many_lstm_regressor.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def forward(self, x, lengths) -> torch.Tensor:
    """Forward pass of the ManyToManyLSTM model.

    Args:
        x (torch.Tensor): Input tensor of shape (batch_size, seq_len, input_size).
        lengths (torch.Tensor): Tensor containing the lengths of each sequence in the batch.

    Returns:
        torch.Tensor: Output tensor of shape (batch_size, seq_len, output_size).

    Raises:
        ValueError: If the input tensor is empty or if the lengths tensor is empty.
        RuntimeError: If the lengths tensor does not match the batch size of the input tensor.
    """
    if x.size(0) == 0 or lengths.size(0) == 0:
        raise ValueError("Input tensor or lengths tensor is empty.")
    if x.size(0) != lengths.size(0):
        raise RuntimeError(f"Batch size of input tensor ({x.size(0)}) and lengths tensor ({lengths.size(0)}) must match.")

    x = pack_padded_sequence(x, lengths, batch_first=True, enforce_sorted=False)
    packed_output, _ = self.lstm_layer(x)
    x, _ = pad_packed_sequence(packed_output, batch_first=True)
    x = self.dropout(x)
    x = self.fc(x)
    x = self.activation_fct(x)
    x = self.output_layer(x)
    return x

ManyToManyLSTMRegressor

Bases: LightningModule

A LightningModule for training and evaluating a Many-to-Many LSTM regressor.

Parameters:

Name Type Description Default
_L_in int

The number of input features.

 required
_L_out int

The number of output features.

 required
l1 int

Unused parameter. Defaults to 8.

 8
lstm_units int

The number of units in the LSTM layer. Defaults to 128.

 128
fc_units int

The number of units in the fully connected layer. Defaults to 128.

 128
act_fn Module

The activation function to use after the fully connected layer. Defaults to nn.ReLU().

 ReLU()
dropout_prob float

The dropout probability. Defaults to 0.2.

 0.2
bidirectional bool

Whether the LSTM is bidirectional. Defaults to True.

 True
num_layers int

The number of LSTM layers. Defaults to 2.

 2
optimizer str

The optimizer to use. Defaults to “Adam”.

 'Adam'
lr_mult float

Learning rate multiplier. Defaults to 1.0.

 1.0
patience int

Patience for learning rate scheduler. Defaults to 5.

 5
epochs int

Number of training epochs. Defaults to 100.

 100
batch_size int

Batch size for training. Defaults to 32.

 32
_torchmetric str

The metric to use for evaluation. Defaults to “mean_squared_error”.

 'mean_squared_error'

Examples:

>>> model = ManyToManyLSTMRegressor(_L_in=10, _L_out=1)
>>> x = torch.randn(16, 10, 10)  # Batch of 16 sequences, each of length 10 with 10 features
>>> lengths = torch.tensor([10] * 16)  # All sequences have length 10
>>> output = model(x, lengths)
>>> print(output.shape)  # Output shape: (16, 10, 1)
Source code in spotpython/light/regression/nn_many_to_many_lstm_regressor.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
class ManyToManyLSTMRegressor(L.LightningModule):
    """A LightningModule for training and evaluating a Many-to-Many LSTM regressor.

    Args:
        _L_in (int): The number of input features.
        _L_out (int): The number of output features.
        l1 (int): Unused parameter. Defaults to 8.
        lstm_units (int): The number of units in the LSTM layer. Defaults to 128.
        fc_units (int): The number of units in the fully connected layer. Defaults to 128.
        act_fn (nn.Module): The activation function to use after the fully connected layer. Defaults to nn.ReLU().
        dropout_prob (float): The dropout probability. Defaults to 0.2.
        bidirectional (bool): Whether the LSTM is bidirectional. Defaults to True.
        num_layers (int): The number of LSTM layers. Defaults to 2.
        optimizer (str): The optimizer to use. Defaults to "Adam".
        lr_mult (float): Learning rate multiplier. Defaults to 1.0.
        patience (int): Patience for learning rate scheduler. Defaults to 5.
        epochs (int): Number of training epochs. Defaults to 100.
        batch_size (int): Batch size for training. Defaults to 32.
        _torchmetric (str): The metric to use for evaluation. Defaults to "mean_squared_error".

    Examples:
        >>> model = ManyToManyLSTMRegressor(_L_in=10, _L_out=1)
        >>> x = torch.randn(16, 10, 10)  # Batch of 16 sequences, each of length 10 with 10 features
        >>> lengths = torch.tensor([10] * 16)  # All sequences have length 10
        >>> output = model(x, lengths)
        >>> print(output.shape)  # Output shape: (16, 10, 1)
    """

    def __init__(
        self,
        _L_in: int,
        _L_out: int,
        l1: int = 8,
        lstm_units: int = 128,
        fc_units: int = 128,
        act_fn: nn.Module = nn.ReLU(),
        dropout_prob: float = 0.2,
        bidirectional: bool = True,
        num_layers: int = 2,
        optimizer: str = "Adam",
        lr_mult: float = 1.0,
        patience: int = 5,
        epochs: int = 100,
        batch_size: int = 32,
        _torchmetric: str = "mean_squared_error",
        *args,
        **kwargs,
    ):
        super().__init__()
        self._L_in = _L_in
        self._L_out = _L_out
        if _torchmetric is None:
            _torchmetric = "mean_squared_error"
        self._torchmetric = _torchmetric
        self.metric = getattr(torchmetrics.functional.regression, _torchmetric)
        self.save_hyperparameters(ignore=["_L_in", "_L_out", "_torchmetric"])
        self.example_input_array = (torch.zeros((batch_size, 10, _L_in)), torch.tensor([10] * batch_size))

        self.layers = ManyToManyLSTM(
            input_size=_L_in,
            output_size=_L_out,
            lstm_units=self.hparams.lstm_units,
            fc_units=self.hparams.fc_units,
            activation_fct=self.hparams.act_fn,
            dropout=self.hparams.dropout_prob,
            bidirectional=self.hparams.bidirectional,
            num_layers=self.hparams.num_layers,
        )

    def forward(self, x, lengths) -> torch.Tensor:
        """Forward pass of the ManyToManyLSTMRegressor.

        Args:
            x (torch.Tensor): Input tensor of shape (batch_size, seq_len, input_size).
            lengths (torch.Tensor): Tensor containing the lengths of each sequence in the batch.

        Returns:
            torch.Tensor: Output tensor of shape (batch_size, seq_len, output_size).
        """
        x = self.layers(x, lengths)
        return x

    def _calculate_loss(self, batch):
        """Calculates the loss for a given batch.

        Args:
            batch (tuple): A tuple containing (x, lengths, y), where:
                - x: Input tensor of shape (batch_size, seq_len, input_size).
                - lengths: Tensor containing the lengths of each sequence in the batch.
                - y: Target tensor of shape (batch_size, seq_len, output_size).

        Returns:
            torch.Tensor: The calculated loss.
        """
        x, lengths, y = batch
        y_hat = self(x, lengths)
        y = y.view_as(y_hat)
        loss = self.metric(y_hat, y)
        return loss

    def training_step(self, batch: tuple, batch_idx) -> torch.Tensor:
        """Performs a single training step.

        Args:
            batch (tuple): A tuple containing (x, lengths, y).
            batch_idx (int): The index of the batch.

        Returns:
            torch.Tensor: The training loss.
        """
        val_loss = self._calculate_loss(batch)
        return val_loss

    def validation_step(self, batch: tuple, batch_idx, prog_bar: bool = False) -> torch.Tensor:
        """Performs a single validation step.

        Args:
            batch (tuple): A tuple containing (x, lengths, y).
            batch_idx (int): The index of the batch.
            prog_bar (bool): Whether to log the loss to the progress bar. Defaults to False.

        Returns:
            torch.Tensor: The validation loss.
        """
        val_loss = self._calculate_loss(batch)
        self.log("val_loss", val_loss, prog_bar=True)
        self.log("hp_metric", val_loss, prog_bar=True)
        return val_loss

    def test_step(self, batch: tuple, batch_idx: int, prog_bar: bool = False) -> torch.Tensor:
        """Performs a single test step.

        Args:
            batch (tuple): A tuple containing (x, lengths, y).
            batch_idx (int): The index of the batch.
            prog_bar (bool): Whether to log the loss to the progress bar. Defaults to False.

        Returns:
            torch.Tensor: The test loss.
        """
        val_loss = self._calculate_loss(batch)
        self.log("val_loss", val_loss, prog_bar=prog_bar)
        self.log("hp_metric", val_loss, prog_bar=prog_bar)
        return val_loss

    def configure_optimizers(self) -> dict:
        """Configures the optimizer and learning rate scheduler.

        Returns:
            dict: A dictionary containing the optimizer and learning rate scheduler configuration.
        """
        optimizer = optimizer_handler(optimizer_name=self.hparams.optimizer, params=self.parameters(), lr_mult=self.hparams.lr_mult)

        num_milestones = 3
        milestones = [int(self.hparams.epochs / (num_milestones + 1) * (i + 1)) for i in range(num_milestones)]
        scheduler = optim.lr_scheduler.MultiStepLR(optimizer, milestones=milestones, gamma=0.1)

        lr_scheduler_config = {
            "scheduler": scheduler,
            "interval": "epoch",
            "frequency": 1,
        }

        return {"optimizer": optimizer, "lr_scheduler": lr_scheduler_config}

configure_optimizers()

Configures the optimizer and learning rate scheduler.

Returns:

Name Type Description
dict dict

A dictionary containing the optimizer and learning rate scheduler configuration.
Source code in spotpython/light/regression/nn_many_to_many_lstm_regressor.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def configure_optimizers(self) -> dict:
    """Configures the optimizer and learning rate scheduler.

    Returns:
        dict: A dictionary containing the optimizer and learning rate scheduler configuration.
    """
    optimizer = optimizer_handler(optimizer_name=self.hparams.optimizer, params=self.parameters(), lr_mult=self.hparams.lr_mult)

    num_milestones = 3
    milestones = [int(self.hparams.epochs / (num_milestones + 1) * (i + 1)) for i in range(num_milestones)]
    scheduler = optim.lr_scheduler.MultiStepLR(optimizer, milestones=milestones, gamma=0.1)

    lr_scheduler_config = {
        "scheduler": scheduler,
        "interval": "epoch",
        "frequency": 1,
    }

    return {"optimizer": optimizer, "lr_scheduler": lr_scheduler_config}

forward(x, lengths)

Forward pass of the ManyToManyLSTMRegressor.

Parameters:

Name Type Description Default
x Tensor

Input tensor of shape (batch_size, seq_len, input_size).

 required
lengths Tensor

Tensor containing the lengths of each sequence in the batch.

 required

Returns:

Type Description
Tensor

torch.Tensor: Output tensor of shape (batch_size, seq_len, output_size).
Source code in spotpython/light/regression/nn_many_to_many_lstm_regressor.py 
160
161
162
163
164
165
166
167
168
169
170
171
def forward(self, x, lengths) -> torch.Tensor:
    """Forward pass of the ManyToManyLSTMRegressor.

    Args:
        x (torch.Tensor): Input tensor of shape (batch_size, seq_len, input_size).
        lengths (torch.Tensor): Tensor containing the lengths of each sequence in the batch.

    Returns:
        torch.Tensor: Output tensor of shape (batch_size, seq_len, output_size).
    """
    x = self.layers(x, lengths)
    return x

test_step(batch, batch_idx, prog_bar=False)

Performs a single test step.

Parameters:

Name Type Description Default
batch tuple

A tuple containing (x, lengths, y).

 required
batch_idx int

The index of the batch.

 required
prog_bar bool

Whether to log the loss to the progress bar. Defaults to False.

 False

Returns:

Type Description
Tensor

torch.Tensor: The test loss.
Source code in spotpython/light/regression/nn_many_to_many_lstm_regressor.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def test_step(self, batch: tuple, batch_idx: int, prog_bar: bool = False) -> torch.Tensor:
    """Performs a single test step.

    Args:
        batch (tuple): A tuple containing (x, lengths, y).
        batch_idx (int): The index of the batch.
        prog_bar (bool): Whether to log the loss to the progress bar. Defaults to False.

    Returns:
        torch.Tensor: The test loss.
    """
    val_loss = self._calculate_loss(batch)
    self.log("val_loss", val_loss, prog_bar=prog_bar)
    self.log("hp_metric", val_loss, prog_bar=prog_bar)
    return val_loss

training_step(batch, batch_idx)

Performs a single training step.

Parameters:

Name Type Description Default
batch tuple

A tuple containing (x, lengths, y).

 required
batch_idx int

The index of the batch.

 required

Returns:

Type Description
Tensor

torch.Tensor: The training loss.
Source code in spotpython/light/regression/nn_many_to_many_lstm_regressor.py 
191
192
193
194
195
196
197
198
199
200
201
202
def training_step(self, batch: tuple, batch_idx) -> torch.Tensor:
    """Performs a single training step.

    Args:
        batch (tuple): A tuple containing (x, lengths, y).
        batch_idx (int): The index of the batch.

    Returns:
        torch.Tensor: The training loss.
    """
    val_loss = self._calculate_loss(batch)
    return val_loss

validation_step(batch, batch_idx, prog_bar=False)

Performs a single validation step.

Parameters:

Name Type Description Default
batch tuple

A tuple containing (x, lengths, y).

 required
batch_idx int

The index of the batch.

 required
prog_bar bool

Whether to log the loss to the progress bar. Defaults to False.

 False

Returns:

Type Description
Tensor

torch.Tensor: The validation loss.
Source code in spotpython/light/regression/nn_many_to_many_lstm_regressor.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def validation_step(self, batch: tuple, batch_idx, prog_bar: bool = False) -> torch.Tensor:
    """Performs a single validation step.

    Args:
        batch (tuple): A tuple containing (x, lengths, y).
        batch_idx (int): The index of the batch.
        prog_bar (bool): Whether to log the loss to the progress bar. Defaults to False.

    Returns:
        torch.Tensor: The validation loss.
    """
    val_loss = self._calculate_loss(batch)
    self.log("val_loss", val_loss, prog_bar=True)
    self.log("hp_metric", val_loss, prog_bar=True)
    return val_loss