| import torch | |
| from torch import nn | |
| from torch.nn import functional as F | |
| import pytorch_lightning as pl | |
| import torchmetrics | |
| from torch.optim.lr_scheduler import OneCycleLR | |
| from torchmetrics.functional import accuracy | |
| class ResBlock(nn.Module): | |
| def __init__(self, in_channel, out_channel, stride=1): | |
| super(ResBlock, self).__init__() | |
| self.conv = nn.Sequential( | |
| nn.Conv2d(in_channel, in_channel, kernel_size=3, stride=1, padding=1, bias=False), | |
| nn.BatchNorm2d(in_channel), | |
| nn.ReLU(), | |
| nn.Conv2d(in_channel, out_channel, kernel_size=3, stride=1, padding=1, bias=False), | |
| nn.BatchNorm2d(out_channel), | |
| nn.ReLU(), | |
| ) | |
| def forward(self, x): | |
| return(self.conv(x)) | |
| class ResNet18(pl.LightningModule): | |
| def __init__(self, train_loader_len, criterion, num_classes=10, lr=0.001, max_lr=1.45E-03): | |
| super().__init__() | |
| self.save_hyperparameters(ignore=['criterion']) | |
| self.criterion = criterion | |
| self.train_loader_len = train_loader_len | |
| self.accuracy = torchmetrics.Accuracy(task="multiclass", num_classes=self.hparams.num_classes) | |
| self.prep_layer = nn.Sequential( | |
| nn.Conv2d(3, 64, kernel_size=3, stride=1, padding=1, bias=False), | |
| nn.BatchNorm2d(64), | |
| nn.ReLU() | |
| ) | |
| self.layer_one = nn.Sequential( | |
| nn.Conv2d(64, 128, kernel_size=3, stride=1, padding=1, bias=False), | |
| nn.MaxPool2d(2,2), | |
| nn.BatchNorm2d(128), | |
| nn.ReLU() | |
| ) | |
| self.res_block1 = ResBlock(128, 128) | |
| self.layer_two = nn.Sequential( | |
| nn.Conv2d(128, 256, kernel_size=3, stride=1, padding=1, bias=False), | |
| nn.MaxPool2d(2,2), | |
| nn.BatchNorm2d(256), | |
| nn.ReLU() | |
| ) | |
| self.layer_three = nn.Sequential( | |
| nn.Conv2d(256, 512, kernel_size=3, stride=1, padding=1, bias=False), | |
| nn.MaxPool2d(2,2), | |
| nn.BatchNorm2d(512), | |
| nn.ReLU() | |
| ) | |
| self.res_block2 = ResBlock(512, 512) | |
| self.max_pool = nn.MaxPool2d(4,4) | |
| self.fc = nn.Linear(512, num_classes, bias=False) | |
| def forward(self, x): | |
| x = self.prep_layer(x) | |
| x = self.layer_one(x) | |
| R1 = self.res_block1(x) | |
| x = x + R1 | |
| x = self.layer_two(x) | |
| x = self.layer_three(x) | |
| R2 = self.res_block2(x) | |
| x = x + R2 | |
| x = self.max_pool(x) | |
| x = x.view(x.size(0), -1) | |
| x = self.fc(x) | |
| return(x) | |
| def configure_optimizers(self): | |
| optimizer = torch.optim.Adam(self.parameters(), lr=self.hparams.lr, weight_decay=1e-4) | |
| scheduler = OneCycleLR( | |
| optimizer, | |
| max_lr=self.hparams.max_lr, | |
| epochs=self.trainer.max_epochs, | |
| steps_per_epoch=self.train_loader_len, | |
| pct_start=5/self.trainer.max_epochs, | |
| div_factor=100, | |
| three_phase=False, | |
| ) | |
| if self.hparams.max_lr==1.45E-03: | |
| return(optimizer) | |
| else: | |
| return([optimizer], [scheduler]) | |
| def training_step(self, train_batch, batch_idx): | |
| data, target = train_batch | |
| y_pred = self(data) | |
| loss = self.criterion(y_pred, target) | |
| pred = torch.argmax(y_pred.squeeze(), dim=1) | |
| acc = accuracy(pred, target, task="multiclass", num_classes=self.hparams.num_classes) | |
| self.log('train_loss', loss, prog_bar=True, on_step=False, on_epoch=True) | |
| self.log('train_acc', acc, prog_bar=True, on_step=False, on_epoch=True) | |
| return(loss) | |
| def validation_step(self, batch, batch_idx): | |
| return(self.evaluate(batch, 'val')) | |
| def test_step(self, batch, batch_idx): | |
| return(self.evaluate(batch, 'test')) | |
| def evaluate(self, batch, stage=None): | |
| data, target = batch | |
| y_pred = self(data) | |
| loss = self.criterion(y_pred, target).item() | |
| pred = torch.argmax(y_pred.squeeze(), dim=1) | |
| acc = accuracy(pred, target, task="multiclass", num_classes=self.hparams.num_classes) | |
| if stage: | |
| self.log(f"{stage}_loss", loss, prog_bar=True, on_step=False, on_epoch=True) | |
| self.log(f"{stage}_acc", acc, prog_bar=True, on_step=False, on_epoch=True) | |
| return pred, target |