117 lines
3.4 KiB
Python
117 lines
3.4 KiB
Python
import torch
|
|
import torch.nn as nn
|
|
import torch.optim as optim
|
|
from torchvision import models
|
|
from omegaconf import DictConfig, OmegaConf
|
|
from data.dataset import get_dataset
|
|
from utils import get_model
|
|
from model.GanInverter.models.stylegan2.model import Generator
|
|
from model.GanInverter.inference.two_stage_inference import TwoStageInference
|
|
from model import GanAttack
|
|
import time
|
|
|
|
import hydra
|
|
|
|
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
|
|
|
|
|
|
def get_stylegan_generator(cfg):
|
|
|
|
# ensure_checkpoint_exists(ckpt_path)
|
|
ckpt_path=cfg.paths.stylegan
|
|
g_ema = Generator(1024, 512, 8)
|
|
g_ema.load_state_dict(torch.load(ckpt_path)["g_ema"], strict=False)
|
|
g_ema.eval()
|
|
g_ema = g_ema.cuda()
|
|
mean_latent = g_ema.mean_latent(4096)
|
|
|
|
return g_ema, mean_latent
|
|
|
|
def get_stylegan_inverter(cfg):
|
|
|
|
# ensure_checkpoint_exists(ckpt_path)
|
|
path=cfg.paths.inverter_cfg
|
|
inverter=TwoStageInference(opts=path)
|
|
|
|
return inverter.inverse
|
|
|
|
@hydra.main(version_base=None, config_path="./config", config_name="config")
|
|
def main(cfg: DictConfig) -> None:
|
|
model=get_model(cfg)
|
|
train_dataloader,test_dataloader,train_dataset,test_dataset=get_dataset(cfg)
|
|
|
|
classifier=get_model(cfg)
|
|
g_ema, _=get_stylegan_generator(cfg)
|
|
|
|
inverter=get_stylegan_inverter(cfg)
|
|
model=GanAttack(g_ema,inverter,images_resize=cfg.optim.images_resize,prompt=cfg.prompt).to(device)
|
|
|
|
num_epochs = cfg.optim.num_epochs
|
|
criterion = nn.CrossEntropyLoss()
|
|
optimizer = optim.SGD(model.parameters(), lr=cfg.classifier.lr, momentum=cfg.classifier.momentum)
|
|
start_time = time.time()
|
|
|
|
for epoch in range(num_epochs):
|
|
model.train()
|
|
|
|
running_loss = 0
|
|
running_corrects = 0
|
|
|
|
for i, (inputs, labels) in enumerate(train_dataloader):
|
|
inputs = inputs.to(device)
|
|
labels = labels.to(device)
|
|
|
|
optimizer.zero_grad()
|
|
outputs = model(inputs)
|
|
_, preds = torch.max(outputs, 1)
|
|
|
|
loss = criterion(outputs, labels)
|
|
loss.backward()
|
|
optimizer.step()
|
|
|
|
running_loss += loss.item() * inputs.size(0)
|
|
running_corrects += torch.sum(preds == labels.data)
|
|
|
|
epoch_loss = running_loss / len(train_dataset)
|
|
epoch_acc = running_corrects / len(train_dataset) * 100.
|
|
print('[Train #{}] Loss: {:.4f} Acc: {:.4f}% Time: {:.4f}s'.format(epoch, epoch_loss, epoch_acc, time.time() - start_time))
|
|
|
|
|
|
model.eval()
|
|
|
|
with torch.no_grad():
|
|
running_loss = 0.
|
|
running_corrects = 0
|
|
|
|
for inputs, labels in test_dataloader:
|
|
inputs = inputs.to(device)
|
|
labels = labels.to(device)
|
|
|
|
outputs = model(inputs)
|
|
_, preds = torch.max(outputs, 1)
|
|
loss = criterion(outputs, labels)
|
|
|
|
running_loss += loss.item() * inputs.size(0)
|
|
running_corrects += torch.sum(preds == labels.data)
|
|
|
|
epoch_loss = running_loss / len(test_dataset)
|
|
epoch_acc = running_corrects / len(test_dataset) * 100.
|
|
print('[Test #{}] Loss: {:.4f} Acc: {:.4f}% Time: {:.4f}s'.format(epoch, epoch_loss, epoch_acc, time.time() - start_time))
|
|
|
|
save_path = '{}/{}_{}.pth'.format(cfg.paths.classifier, cfg.classifier.model, cfg.dataset)
|
|
torch.save(model.state_dict(), save_path)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |