"""
File: qumphy/models/inception1d.py
Project: 22HLT01 QUMPHY
Contact: oskar.pfeffer@ptb.de
Gitlab: https://gitlab.com/qumphy
Description: InceptionTime-based 1D convolutional neural network.
"""
import torch
import torch.nn as nn
import qumphy.models.basic_conv1d
[docs]
def conv(in_planes, out_planes, kernel_size=3, stride=1):
"""Create a 1D convolutional layer with padding.
Parameters
----------
in_planes : int
Number of input channels.
out_planes : int
Number of output channels.
kernel_size : int
Size of the convolution kernel.
stride : int
Stride of the convolution.
Returns
-------
nn.Conv1d
One-dimensional convolutional layer.
"""
return nn.Conv1d(
in_planes,
out_planes,
kernel_size=kernel_size,
stride=stride,
padding=(kernel_size - 1) // 2,
bias=False,
)
[docs]
def noop(x):
"""Return the input unchanged.
Parameters
----------
x : torch.Tensor
Input tensor.
Returns
-------
torch.Tensor
Same tensor as the input.
"""
return x
[docs]
class InceptionBlock1d(nn.Module):
"""One-dimensional Inception block.
The block applies several convolutions with different kernel sizes and
concatenates their outputs with a max-pooling branch.
Parameters
----------
ni : int
Number of input channels.
nb_filters : int
Number of filters used in each convolution branch.
kss : list
List of kernel sizes used by the convolution branches.
stride : int
Stride used in the bottleneck and pooling branches.
act : str
Activation function name. This parameter is currently not used.
bottleneck_size : int
Number of filters used in the bottleneck convolution. If 0, no
bottleneck convolution is applied.
"""
def __init__(self, ni, nb_filters, kss, stride=1, act="linear", bottleneck_size=32):
super().__init__()
self.bottleneck = (
conv(ni, bottleneck_size, 1, stride) if (bottleneck_size > 0) else noop
)
self.convs = nn.ModuleList(
[
conv(bottleneck_size if (bottleneck_size > 0) else ni, nb_filters, ks)
for ks in kss
]
)
self.conv_bottle = nn.Sequential(
nn.MaxPool1d(3, stride, padding=1), conv(ni, nb_filters, 1)
)
self.bn_relu = nn.Sequential(
nn.BatchNorm1d((len(kss) + 1) * nb_filters), nn.ReLU()
)
[docs]
def forward(self, x):
"""Run a forward pass through the Inception block.
Parameters
----------
x : torch.Tensor
Input tensor of shape (batch_size, channels, sequence_length).
Returns
-------
torch.Tensor
Output tensor after concatenation, batch normalization, and ReLU.
"""
bottled = self.bottleneck(x)
out = self.bn_relu(
torch.cat([c(bottled) for c in self.convs] + [self.conv_bottle(x)], dim=1)
)
return out
[docs]
class Shortcut1d(nn.Module):
"""Residual shortcut connection for 1D convolutional features."""
def __init__(self, ni, nf):
"""Initialize the shortcut connection.
Parameters
----------
ni : int
Number of input channels.
nf : int
Number of output channels.
"""
super().__init__()
self.act_fn = nn.ReLU(True)
self.conv = conv(ni, nf, 1)
self.bn = nn.BatchNorm1d(nf)
[docs]
def forward(self, inp, out):
"""Apply the residual shortcut connection.
Parameters
----------
inp : torch.Tensor
Input tensor used for the shortcut branch.
out : torch.Tensor
Output tensor from the main branch.
Returns
-------
torch.Tensor
Result of adding the shortcut branch to the main branch and
applying ReLU activation.
"""
return self.act_fn(out + self.bn(self.conv(inp)))
[docs]
class InceptionBackbone(nn.Module):
"""Backbone of the 1D Inception architecture.
The backbone stacks multiple Inception blocks and optionally applies
residual shortcut connections every three blocks.
"""
def __init__(
self, input_channels, kss, depth, bottleneck_size, nb_filters, use_residual
):
super().__init__()
"""Initialize the 1D Inception backbone.
Parameters
----------
input_channels : int
Number of input channels.
kss : list
List of kernel sizes used in each Inception block.
depth : int
Number of Inception blocks. Must be divisible by 3.
bottleneck_size : int
Number of filters used in the bottleneck convolutions.
nb_filters : int
Number of filters used in each convolution branch.
use_residual : bool
If True, use residual shortcut connections every three blocks.
"""
super().__init__()
self.depth = depth
assert (depth % 3) == 0
self.use_residual = use_residual
n_ks = len(kss) + 1
self.im = nn.ModuleList(
[
InceptionBlock1d(
input_channels if d == 0 else n_ks * nb_filters,
nb_filters=nb_filters,
kss=kss,
bottleneck_size=bottleneck_size,
)
for d in range(depth)
]
)
self.sk = nn.ModuleList(
[
Shortcut1d(
input_channels if d == 0 else n_ks * nb_filters, n_ks * nb_filters
)
for d in range(depth // 3)
]
)
[docs]
def forward(self, x):
"""Run a forward pass through the Inception backbone.
Parameters
----------
x : torch.Tensor
Input tensor of shape (batch_size, input_channels, sequence_length).
Returns
-------
torch.Tensor
Output features from the stacked Inception blocks.
"""
input_res = x
for d in range(self.depth):
x = self.im[d](x)
if self.use_residual and d % 3 == 2:
x = (self.sk[d // 3])(input_res, x)
input_res = x.clone()
return x
[docs]
class Inception1d(nn.Module):
"""inception time architecture"""
def __init__(
self,
num_classes=2,
input_channels=8,
kss=[39, 19, 9],
depth=6,
bottleneck_size=32,
nb_filters=32,
use_residual=True,
lin_ftrs_head=None,
ps_head=0.5,
bn_head=True,
act_head="relu",
concat_pooling=True,
):
"""Initialize the 1D Inception model.
Parameters
----------
num_classes : int
Number of output classes or output values.
input_channels : int
Number of input channels.
kss : list
List of kernel sizes used in the Inception blocks.
depth : int
Number of Inception blocks. Must be divisible by 3.
bottleneck_size : int
Number of filters used in the bottleneck convolutions.
nb_filters : int
Number of filters used in each convolution branch.
use_residual : bool
If True, use residual shortcut connections.
lin_ftrs_head : list
Hidden layer sizes used in the model head.
ps_head : float or Iterable
Dropout probability or probabilities used in the model head.
bn_head : bool
If True, use batch normalization in the model head.
act_head : str
Activation function used in the model head.
concat_pooling : bool
If True, use concatenated adaptive average and max pooling in the head.
"""
super().__init__()
layers = [
InceptionBackbone(
input_channels=input_channels,
kss=kss,
depth=depth,
bottleneck_size=bottleneck_size,
nb_filters=nb_filters,
use_residual=use_residual,
)
]
n_ks = len(kss) + 1
# head
head = qumphy.models.basic_conv1d.create_head1d(
n_ks * nb_filters,
nc=num_classes,
lin_ftrs=lin_ftrs_head,
ps=ps_head,
bn=bn_head,
act=act_head,
concat_pooling=concat_pooling,
)
layers.append(head)
# layers.append(AdaptiveConcatPool1d())
# layers.append(Flatten())
# layers.append(nn.Linear(2*n_ks*nb_filters, num_classes))
self.layers = nn.Sequential(*layers)
[docs]
def forward(self, x):
"""Run a forward pass through the model.
Parameters
----------
x : torch.Tensor
Input tensor of shape (batch_size, input_channels, sequence_length).
Returns
-------
torch.Tensor
Model output tensor of shape (batch_size, num_classes).
"""
return self.layers(x)
[docs]
def get_layer_groups(self):
"""Get grouped layers for model training.
Returns
-------
tuple or nn.Module
Layer groups containing part of the backbone and the model head.
If the depth is less than or equal to 3, only the model head is returned.
"""
depth = self.layers[0].depth
if depth > 3:
return ((self.layers[0].im[3:], self.layers[0].sk[1:]), self.layers[-1])
else:
return self.layers[-1]
[docs]
def get_output_layer(self):
"""Get the output layer of the model.
Returns
-------
nn.Module
Final output layer of the model head.
"""
return self.layers[-1][-1]
[docs]
def set_output_layer(self, x):
"""Set the output layer of the model.
Parameters
----------
x : nn.Module
New output layer.
Returns
-------
None
The function modifies the output layer in place.
"""
self.layers[-1][-1] = x
[docs]
def inception1d(**kwargs):
"""Construct an Inception1d model.
Parameters
----------
**kwargs
Keyword arguments passed to Inception1d.
Returns
-------
Inception1d
Initialized Inception1d model.
"""
return Inception1d(**kwargs)