HW02

HW02/README.txt

@@ -0,0 +1,15 @@
+This is homework ResNext realization based on ResNet from pytorch.
+
+Loading of pretrained weights is achieved by method
+load_state_dict of nn.Module.
+
+run
+
+pytest
+
+to execute test
+
+or
+
+python3 setup.py sdist
+

HW02/resnext/resnext.py

@@ -0,0 +1,114 @@
+import torch.nn as nn
+import math
+import torch.utils.model_zoo as model_zoo
+
+
+__all__ = ['ResNext', 'Bottleneck']
+
+
+def conv3x3(in_planes, out_planes, stride=1):
+    """3x3 convolution with padding"""
+    return nn.Conv2d(in_planes, out_planes, kernel_size=3, stride=stride,
+                     padding=1, bias=False)
+
+
+class Bottleneck(nn.Module):
+    expansion = 2
+
+    def __init__(self, inplanes, planes, stride=1, downsample=None):
+        super(Bottleneck, self).__init__()
+        self.conv1 = nn.Conv2d(inplanes, planes, kernel_size=1, bias=False)
+        self.bn1 = nn.BatchNorm2d(planes)
+        self.conv2 = nn.Conv2d(planes, planes, kernel_size=3, stride=stride,
+                               padding=1, bias=False, groups=32)
+        self.bn2 = nn.BatchNorm2d(planes)
+        self.conv3 = nn.Conv2d(planes, planes * Bottleneck.expansion, kernel_size=1, bias=False)
+        self.bn3 = nn.BatchNorm2d(planes * Bottleneck.expansion)
+        self.relu = nn.ReLU(inplace=True)
+        self.downsample = downsample
+        self.stride = stride
+
+    def forward(self, x):
+        residual = x
+
+        out = self.conv1(x)
+        out = self.bn1(out)
+        out = self.relu(out)
+
+        out = self.conv2(out)
+        out = self.bn2(out)
+        out = self.relu(out)
+
+        out = self.conv3(out)
+        out = self.bn3(out)
+
+        if self.downsample is not None:
+            residual = self.downsample(x)
+
+        out += residual
+        out = self.relu(out)
+
+        return out
+
+
+class ResNext(nn.Module):
+
+    def __init__(self, block, layers, num_classes=1000):
+        self.inplanes = 128
+        super(ResNext, self).__init__()
+        self.conv1 = nn.Conv2d(1, self.inplanes, kernel_size=7, stride=2, padding=3,
+                               bias=False)
+        self.bn1 = nn.BatchNorm2d(self.inplanes)
+        self.relu = nn.ReLU(inplace=True)
+        self.maxpool = nn.MaxPool2d(kernel_size=3, stride=2, padding=1)
+        self.layer1 = self._make_layer(block, 128, layers[0])
+        self.layer2 = self._make_layer(block, 256, layers[1], stride=2)
+        self.layer3 = self._make_layer(block, 512, layers[2], stride=2)
+        self.layer4 = self._make_layer(block, 1024, layers[3], stride=2)
+        self.avgpool = nn.AvgPool2d(1, stride=1)
+        self.fc = nn.Linear(1024 * block.expansion, num_classes)
+
+        for m in self.modules():
+            if isinstance(m, nn.Conv2d):
+                n = m.kernel_size[0] * m.kernel_size[1] * m.out_channels
+                m.weight.data.normal_(0, math.sqrt(2. / n))
+            elif isinstance(m, nn.BatchNorm2d):
+                m.weight.data.fill_(1)
+                m.bias.data.zero_()
+
+    def _make_layer(self, block, planes, blocks, stride=1):
+        downsample = None
+        if stride != 1 or self.inplanes != planes * block.expansion:
+            downsample = nn.Sequential(
+                nn.Conv2d(self.inplanes, planes * block.expansion,
+                          kernel_size=1, stride=stride, bias=False),
+                nn.BatchNorm2d(planes * block.expansion),
+            )
+
+        layers = []
+        layers.append(block(self.inplanes, planes, stride, downsample))
+        self.inplanes = planes * block.expansion
+        for i in range(1, blocks):
+            layers.append(block(self.inplanes, planes))
+
+        return nn.Sequential(*layers)
+
+    def forward(self, x):
+        print('inputs in forward: ', x.shape)
+        x = self.conv1(x)
+        x = self.bn1(x)
+        x = self.relu(x)
+        x = self.maxpool(x)
+
+        x = self.layer1(x)
+        x = self.layer2(x)
+        x = self.layer3(x)
+        x = self.layer4(x)
+
+        x = self.avgpool(x)
+        x = x.view(x.size(0), -1)
+        x = self.fc(x)
+
+        return x
+
+

HW02/resnext/test/test_resnext.py

@@ -0,0 +1,34 @@
+
+from torch.utils.data import Dataset
+
+import sys
+sys.path.append('.')
+from resnext.resnext import *
+from resnext.trainer import Trainer
+import torch
+
+torch.manual_seed(10)
+
+print('forward pass on random data test')
+
+class simple_dataset(Dataset):
+    def __init__(self):
+        pass
+
+    def __len__(self):
+        return 4
+
+    def __getitem__(self, idx):
+        return torch.randn(1, 28, 28), 1
+
+
+def test_forward_pass():
+    net = ResNext(Bottleneck, [3, 4, 6, 3], num_classes=2)
+    trainer = Trainer(simple_dataset(), simple_dataset())
+
+    trainer.train(net)
+
+    assert True
+
+if __name__ == "__main__":
+    test_forward_pass()

HW02/resnext/trainer.py

@@ -0,0 +1,77 @@
+
+import torch
+import torch.optim as optim
+from tensorboardX import SummaryWriter
+
+class Trainer:
+    def __init__(self, trainset, testset):
+        #self.trainset = torchvision.datasets.CIFAR10(root='./data', train=True,
+        #                                download=True, transform=transform)
+        self.trainloader = torch.utils.data.DataLoader(trainset, batch_size=4,
+                                                  shuffle=False, num_workers=2)
+
+        #self.testset = torchvision.datasets.CIFAR10(root='./data', train=False,
+        #                                       download=True, transform=transform)
+        self.testloader = torch.utils.data.DataLoader(testset, batch_size=4,
+                                                 shuffle=False, num_workers=2)
+
+
+    def compute_loss_accuracy(self, dataloader, net, criterion):
+        count = 0
+        full_loss = 0
+
+        correct_predictions = 0
+        all_predictions = 0
+        with torch.no_grad():
+            for data in dataloader:
+                count += 1
+                inputs, labels = data
+
+                outputs = net(inputs)
+
+                loss = criterion(outputs, labels)
+                full_loss += loss.item()
+
+                _, predicted = torch.max(outputs, 1)
+                # print(predicted, labels, predicted == labels, torch.sum(predicted == labels))
+
+                all_predictions += labels.shape[0]
+                correct_predictions += torch.sum(predicted == labels).item()
+
+        full_loss /= count
+
+        return (full_loss, correct_predictions / all_predictions)
+
+    def train(self, net, should_log=True):
+        criterion = torch.nn.CrossEntropyLoss()
+        optimizer = optim.SGD(net.parameters(), lr=0.001, momentum=0.9)
+
+        sw = SummaryWriter('logs')
+
+        for epoch in range(1):  # loop over the dataset multiple times
+
+            #running_loss = 0.0
+            for i, data in enumerate(self.trainloader, 0):
+                # get the inputs
+                inputs, labels = data
+                print('inputs in train: ', inputs.shape)
+
+                # zero the parameter gradients
+                optimizer.zero_grad()
+
+                # forward + backward + optimize
+                outputs = net(inputs)
+                loss = criterion(outputs, labels)
+                loss.backward()
+                optimizer.step()
+
+                # print statistics
+                #running_loss += loss.item()
+                if should_log:
+
+                    loss, acc = self.compute_loss_accuracy(self.trainloader, net, criterion)
+                    sw.add_scalar('trainloss', loss)
+
+                    loss, acc = self.compute_loss_accuracy(self.testloader, net, criterion)
+
+                    sw.add_scalar('testloss', loss)

HW02/setup.py

@@ -0,0 +1,10 @@
+from distutils.core import setup
+
+setup(
+    name='ResNextStudy',
+    version='0.1.0',
+    author='Student Studentson',
+    packages=['resnext'],
+    description='Homework ResNext based on ResNet.',
+    long_description=open('README.txt').read(),
+)