SwaV
Example implementation of the SwaV architecture.
This example can be run from the command line with:
python lightly/examples/pytorch/swav.py
# Note: The model and training settings do not follow the reference settings
# from the paper. The settings are chosen such that the example can easily be
# run on a small dataset with a single GPU.
import torch
from torch import nn
import torchvision
from lightly.data import LightlyDataset
from lightly.data import SwaVCollateFunction
from lightly.loss import SwaVLoss
from lightly.models.modules import SwaVProjectionHead
from lightly.models.modules import SwaVPrototypes
class SwaV(nn.Module):
def __init__(self, backbone):
super().__init__()
self.backbone = backbone
self.projection_head = SwaVProjectionHead(512, 512, 128)
self.prototypes = SwaVPrototypes(128, n_prototypes=512)
def forward(self, x):
x = self.backbone(x).flatten(start_dim=1)
x = self.projection_head(x)
x = nn.functional.normalize(x, dim=1, p=2)
p = self.prototypes(x)
return p
resnet = torchvision.models.resnet18()
backbone = nn.Sequential(*list(resnet.children())[:-1])
model = SwaV(backbone)
device = "cuda" if torch.cuda.is_available() else "cpu"
model.to(device)
# we ignore object detection annotations by setting target_transform to return 0
pascal_voc = torchvision.datasets.VOCDetection(
"datasets/pascal_voc", download=True, target_transform=lambda t: 0
)
dataset = LightlyDataset.from_torch_dataset(pascal_voc)
# or create a dataset from a folder containing images or videos:
# dataset = LightlyDataset("path/to/folder")
collate_fn = SwaVCollateFunction()
dataloader = torch.utils.data.DataLoader(
dataset,
batch_size=128,
collate_fn=collate_fn,
shuffle=True,
drop_last=True,
num_workers=8,
)
criterion = SwaVLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
print("Starting Training")
for epoch in range(10):
total_loss = 0
for batch, _, _ in dataloader:
model.prototypes.normalize()
multi_crop_features = [model(x.to(device)) for x in batch]
high_resolution = multi_crop_features[:2]
low_resolution = multi_crop_features[2:]
loss = criterion(high_resolution, low_resolution)
total_loss += loss.detach()
loss.backward()
optimizer.step()
optimizer.zero_grad()
avg_loss = total_loss / len(dataloader)
print(f"epoch: {epoch:>02}, loss: {avg_loss:.5f}")
This example can be run from the command line with:
python lightly/examples/pytorch_lightning/swav.py
# Note: The model and training settings do not follow the reference settings
# from the paper. The settings are chosen such that the example can easily be
# run on a small dataset with a single GPU.
import torch
from torch import nn
import torchvision
import pytorch_lightning as pl
from lightly.data import LightlyDataset
from lightly.data import SwaVCollateFunction
from lightly.loss import SwaVLoss
from lightly.models.modules import SwaVProjectionHead
from lightly.models.modules import SwaVPrototypes
class SwaV(pl.LightningModule):
def __init__(self):
super().__init__()
resnet = torchvision.models.resnet18()
self.backbone = nn.Sequential(*list(resnet.children())[:-1])
self.projection_head = SwaVProjectionHead(512, 512, 128)
self.prototypes = SwaVPrototypes(128, n_prototypes=512)
self.criterion = SwaVLoss()
def forward(self, x):
x = self.backbone(x).flatten(start_dim=1)
x = self.projection_head(x)
x = nn.functional.normalize(x, dim=1, p=2)
p = self.prototypes(x)
return p
def training_step(self, batch, batch_idx):
self.prototypes.normalize()
crops, _, _ = batch
multi_crop_features = [self.forward(x.to(self.device)) for x in crops]
high_resolution = multi_crop_features[:2]
low_resolution = multi_crop_features[2:]
loss = self.criterion(high_resolution, low_resolution)
return loss
def configure_optimizers(self):
optim = torch.optim.Adam(self.parameters(), lr=0.001)
return optim
model = SwaV()
# we ignore object detection annotations by setting target_transform to return 0
pascal_voc = torchvision.datasets.VOCDetection(
"datasets/pascal_voc", download=True, target_transform=lambda t: 0
)
dataset = LightlyDataset.from_torch_dataset(pascal_voc)
# or create a dataset from a folder containing images or videos:
# dataset = LightlyDataset("path/to/folder")
collate_fn = SwaVCollateFunction()
dataloader = torch.utils.data.DataLoader(
dataset,
batch_size=128,
collate_fn=collate_fn,
shuffle=True,
drop_last=True,
num_workers=8,
)
gpus = 1 if torch.cuda.is_available() else 0
trainer = pl.Trainer(max_epochs=10, gpus=gpus)
trainer.fit(model=model, train_dataloaders=dataloader)
This example runs on multiple gpus using Distributed Data Parallel (DDP) training with Pytorch Lightning. At least one GPU must be available on the system. The example can be run from the command line with:
python lightly/examples/pytorch_lightning_distributed/swav.py
The model differs in the following ways from the non-distributed implementation:
Distributed Data Parallel is enabled
Synchronized Batch Norm is used in place of standard Batch Norm
Distributed Sinkhorn is used in the loss calculation
Note that Synchronized Batch Norm and distributed Sinkhorn are optional and the model can also be trained without them. Without Synchronized Batch Norm and distributed Sinkhorn the batch norm and loss for each GPU are only calculated based on the features on that specific GPU.
# Note: The model and training settings do not follow the reference settings
# from the paper. The settings are chosen such that the example can easily be
# run on a small dataset with a single GPU.
import torch
from torch import nn
import torchvision
import pytorch_lightning as pl
from lightly.data import LightlyDataset
from lightly.data import SwaVCollateFunction
from lightly.loss import SwaVLoss
from lightly.models.modules import SwaVProjectionHead
from lightly.models.modules import SwaVPrototypes
class SwaV(pl.LightningModule):
def __init__(self):
super().__init__()
resnet = torchvision.models.resnet18()
self.backbone = nn.Sequential(*list(resnet.children())[:-1])
self.projection_head = SwaVProjectionHead(512, 512, 128)
self.prototypes = SwaVPrototypes(128, n_prototypes=512)
# enable sinkhorn_gather_distributed to gather features from all gpus
# while running the sinkhorn algorithm in the loss calculation
self.criterion = SwaVLoss(sinkhorn_gather_distributed=True)
def forward(self, x):
x = self.backbone(x).flatten(start_dim=1)
x = self.projection_head(x)
x = nn.functional.normalize(x, dim=1, p=2)
p = self.prototypes(x)
return p
def training_step(self, batch, batch_idx):
self.prototypes.normalize()
crops, _, _ = batch
multi_crop_features = [self.forward(x.to(self.device)) for x in crops]
high_resolution = multi_crop_features[:2]
low_resolution = multi_crop_features[2:]
loss = self.criterion(high_resolution, low_resolution)
return loss
def configure_optimizers(self):
optim = torch.optim.Adam(self.parameters(), lr=0.001)
return optim
model = SwaV()
# we ignore object detection annotations by setting target_transform to return 0
pascal_voc = torchvision.datasets.VOCDetection(
"datasets/pascal_voc", download=True, target_transform=lambda t: 0
)
dataset = LightlyDataset.from_torch_dataset(pascal_voc)
# or create a dataset from a folder containing images or videos:
# dataset = LightlyDataset("path/to/folder")
collate_fn = SwaVCollateFunction()
dataloader = torch.utils.data.DataLoader(
dataset,
batch_size=128,
collate_fn=collate_fn,
shuffle=True,
drop_last=True,
num_workers=8,
)
gpus = torch.cuda.device_count()
# train with DDP and use Synchronized Batch Norm for a more accurate batch norm
# calculation
trainer = pl.Trainer(
max_epochs=10,
gpus=gpus,
strategy='ddp',
sync_batchnorm=True,
)
trainer.fit(model=model, train_dataloaders=dataloader)