add cvae training
This commit is contained in:
+8
-6
@@ -6,22 +6,24 @@ from typing import Any, Callable, List, Optional, Sequence
|
||||
|
||||
class VariationalEncoder(nn.Module):
|
||||
|
||||
def __init__(self, block_setting: CNBlockConfig, kernel_size: int = 7):
|
||||
def __init__(self, block_setting: List[CNBlockConfig], kernel_size: int = 7, input_features: int = 16, device: str = 'cuda'):
|
||||
super(VariationalEncoder, self).__init__()
|
||||
|
||||
self.encoder = ConvNeXt(block_setting=block_setting, kernel_size=kernel_size)
|
||||
self.encoder = ConvNeXt(block_setting=block_setting, kernel_size=kernel_size, input_features=input_features, device=device)
|
||||
|
||||
self.mean_branch = nn.Sequential([CNBlock(block_setting[-1].out_channels, kernel_size=kernel_size) for _ in range(5)])
|
||||
self.var_branch = nn.Sequential([CNBlock(block_setting[-1].out_channels, kernel_size=kernel_size) for _ in range(5)])
|
||||
layers = [CNBlock(block_setting[-1].input_channels, kernel_size=kernel_size) for _ in range(5)]
|
||||
self.mean_branch = nn.Sequential(*layers).to(device)
|
||||
layers = [CNBlock(block_setting[-1].input_channels, kernel_size=kernel_size) for _ in range(5)]
|
||||
self.var_branch = nn.Sequential(*layers).to(device)
|
||||
|
||||
self.flatten_mean = nn.Flatten()
|
||||
self.flatten_var = nn.Flatten()
|
||||
|
||||
|
||||
def call(self, inputs, training):
|
||||
def forward(self, inputs):
|
||||
x = inputs
|
||||
|
||||
x = self.encoder(x, training=training)
|
||||
x = self.encoder(x)
|
||||
mean = self.mean_branch(x)
|
||||
log_var = self.var_branch(x)
|
||||
|
||||
|
||||
@@ -66,6 +66,7 @@ class ConvNeXtBase(nn.Module):
|
||||
kernel_size: int = 7,
|
||||
is_label: bool = False,
|
||||
nr_of_labels: Optional[int] = None,
|
||||
input_features: int = 3,
|
||||
) -> None:
|
||||
super().__init__()
|
||||
|
||||
@@ -85,7 +86,7 @@ class ConvNeXtBase(nn.Module):
|
||||
# Stem
|
||||
firstconv_output_channels = block_setting[0].input_channels
|
||||
self.stem = Conv2dNormActivation(
|
||||
3 if not is_label else nr_of_labels,
|
||||
input_features if not is_label else nr_of_labels,
|
||||
firstconv_output_channels,
|
||||
kernel_size=4,
|
||||
stride=4,
|
||||
@@ -229,6 +230,7 @@ def _convnext(
|
||||
is_label: bool,
|
||||
nr_of_labels: Optional[int],
|
||||
kernel_size: int,
|
||||
input_features: int,
|
||||
device='cuda',
|
||||
**kwargs: Any,
|
||||
) -> ConvNeXtBase:
|
||||
@@ -236,17 +238,17 @@ def _convnext(
|
||||
if reversed:
|
||||
model = ConvNeXtReverseBase(block_setting, stochastic_depth_prob=stochastic_depth_prob, is_label=is_label, nr_of_labels=nr_of_labels, kernel_size=kernel_size, device=device, **kwargs)
|
||||
else:
|
||||
model = ConvNeXtBase(block_setting, stochastic_depth_prob=stochastic_depth_prob, kernel_size=kernel_size, is_label=is_label, nr_of_labels=nr_of_labels, device=device, **kwargs)
|
||||
model = ConvNeXtBase(block_setting, stochastic_depth_prob=stochastic_depth_prob, kernel_size=kernel_size, is_label=is_label, nr_of_labels=nr_of_labels, input_features=input_features, device=device, **kwargs)
|
||||
|
||||
|
||||
return model
|
||||
|
||||
|
||||
class ConvNeXt(nn.Module):
|
||||
def __init__(self, block_setting, stochastic_depth_prob=0.0, reversed=False, is_label=False, nr_of_labels=None, kernel_size=7, device='cuda', **kwargs) -> None:
|
||||
def __init__(self, block_setting, stochastic_depth_prob=0.0, reversed=False, is_label=False, nr_of_labels=None, kernel_size=7, input_features=3, device='cuda', **kwargs) -> None:
|
||||
super().__init__()
|
||||
|
||||
self.model = _convnext(block_setting, stochastic_depth_prob, reversed, is_label, nr_of_labels, kernel_size, device=device, **kwargs)
|
||||
self.model: nn.Module = _convnext(block_setting, stochastic_depth_prob, reversed, is_label, nr_of_labels, kernel_size, input_features=input_features, device=device, **kwargs)
|
||||
|
||||
def forward(self, x: Tensor) -> Tensor:
|
||||
x = self.model(x)
|
||||
|
||||
+39
-194
@@ -1,63 +1,54 @@
|
||||
import os
|
||||
import time
|
||||
import random
|
||||
import tensorflow as tf
|
||||
import torch
|
||||
from torch import nn
|
||||
import numpy as np
|
||||
import scipy as sp
|
||||
# import plotly.graph_objects as go
|
||||
# import plotly.express as ex
|
||||
|
||||
from custom_utils.custom_utils import load_model
|
||||
from utils.checkpoints import load_model
|
||||
from models.base import AEEncoder, VariationalEncoder, Decoder
|
||||
from models.base import AEEncoderNew, VariationalEncoderNew, DecoderNew
|
||||
|
||||
|
||||
class CVAE(tf.keras.Model):
|
||||
class CVAE(nn.Module):
|
||||
|
||||
def __init__(self, config):
|
||||
super(CVAE, self).__init__()
|
||||
self.latent_dim = config['cvae']['latent_dim']
|
||||
self.size = config['resolution']
|
||||
|
||||
if not('only_curvatures' in config.keys() and config['only_curvatures'] == True):
|
||||
self.static_enc : AEEncoder = AEEncoder(config['cond_ae']['latent_dim'], config['cond_ae']['encoder_depths'], config['cond_ae']['encoder_dims'], config['cond_ae']['bottleneck_size'], config['cond_ae']['kernel_size'])
|
||||
self.static_enc : AEEncoder = AEEncoder(config['cond_ae_static']['encoder_block_settings'], config['cond_ae_static']['kernel_size'], config['cond_ae_static']['activation'], device=config['device'])
|
||||
if config['static_model'] != '':
|
||||
load_model(self.static_enc, config['static_model'] + '_enc')
|
||||
|
||||
self.body_enc : AEEncoder = AEEncoder(config['cond_ae_body']['latent_dim'], config['cond_ae_body']['encoder_depths'], config['cond_ae_body']['encoder_dims'], config['cond_ae_body']['bottleneck_size'], config['cond_ae_body']['kernel_size'])
|
||||
self.body_enc : AEEncoder = AEEncoder(config['cond_ae_body']['encoder_block_settings'], config['cond_ae_body']['kernel_size'], config['cond_ae_body']['activation'], device=config['device'])
|
||||
if config['body_model'] != '':
|
||||
load_model(self.body_enc, config['body_model'] + '_enc')
|
||||
|
||||
if config['use_curvature']:
|
||||
self.curv_enc : AEEncoder = AEEncoder(config['cond_ae']['latent_dim'], config['cond_ae']['encoder_depths'], config['cond_ae']['encoder_dims'], config['cond_ae']['bottleneck_size'], config['cond_ae']['kernel_size'])
|
||||
self.curv_enc : AEEncoder = AEEncoder(config['cond_ae_curv']['encoder_block_settings'], config['cond_ae_curv']['kernel_size'], config['cond_ae_curv']['activation'], device=config['device'], is_label=True, nr_of_labels=7)
|
||||
if config['curvature_model'] != '':
|
||||
load_model(self.curv_enc, config['curvature_model'] + '_enc')
|
||||
|
||||
if config['freeze_encoders']:
|
||||
if not('only_curvatures' in config.keys() and config['only_curvatures'] == True):
|
||||
self.static_enc.trainable = False
|
||||
self.body_enc.trainable = False
|
||||
for param in self.static_enc.parameters(): param.requires_grad = False
|
||||
for param in self.body_enc.parameters(): param.requires_grad = False
|
||||
if config['use_curvature']:
|
||||
self.curv_enc.trainable = False
|
||||
for param in self.curv_enc.parameters(): param.requires_grad = False
|
||||
|
||||
self.cond_encoder = tf.keras.Sequential([tf.keras.layers.Dense(512, activation='relu'),
|
||||
tf.keras.layers.Dense(256, activation='relu'),
|
||||
tf.keras.layers.Dense(128, activation='relu'),
|
||||
# tf.keras.layers.Lambda(lambda x: x / 1000),
|
||||
tf.keras.layers.Dense(config['latent_dim'], activation='tanh')])
|
||||
self.encoder = VariationalEncoder(config['cvae']['encoder_block_settings'], config['cvae']['kernel_size'], config['cvae']['input_features'], device=config['device'])
|
||||
self.decoder : Decoder = Decoder(config['cvae']['decoder_block_settings'], kernel_size=config['cvae']['kernel_size'], device=config['device'])
|
||||
|
||||
self.encoder = VariationalEncoder(config['cvae']['latent_dim'], config['cvae']['encoder_depths'], config['cvae']['encoder_dims'], kernel_size=config['cvae']['kernel_size'])
|
||||
self.decoder = Decoder(config['cvae']['decoder_input'], config['cvae']['decoder_depths'], config['cvae']['decoder_dims'], config['cvae']['kernel_size'])
|
||||
self.latent_shape = (config['cvae']['latent_dim'], 8, 4)
|
||||
|
||||
if config['avg_garment_path'] != '':
|
||||
self.avg_garment = np.load(config['avg_garment_path'])
|
||||
self.avg_garment = torch.permute(torch.tensor(np.load(config['avg_garment_path']), device=config['device']), [2,0,1])
|
||||
else:
|
||||
self.avg_garment = None
|
||||
|
||||
self.add_noise = config['cvae']['add_noise']
|
||||
self.noise_amount = config['cvae']['noise_amount']
|
||||
self.use_curvature = config['use_curvature']
|
||||
self.only_curvatures = config['only_curvatures'] if 'only_curvatures' in config.keys() and config['only_curvatures'] == True else None
|
||||
self.use_offsets_over_body = not config['use_skinning']
|
||||
|
||||
if self.use_offsets_over_body:
|
||||
@@ -68,7 +59,7 @@ class CVAE(tf.keras.Model):
|
||||
# @tf.function
|
||||
def sample(self, uv_template, uv_body_posed, uv_curv, uv_mask, noise=None, add_noise=False):
|
||||
if noise is None:
|
||||
eps = tf.random.normal(shape=(uv_template.shape[0], self.latent_dim))
|
||||
eps = tf.random.normal(shape=(uv_template.shape[0], 8, 4, self.latent_dim))
|
||||
else:
|
||||
eps = noise
|
||||
|
||||
@@ -91,195 +82,49 @@ class CVAE(tf.keras.Model):
|
||||
return x
|
||||
|
||||
|
||||
def reparameterize(self, mean, logvar, training, fix_noise=False):
|
||||
def reparameterize(self, mean: torch.Tensor, logvar: torch.Tensor, training: bool = True, fix_noise: bool = False):
|
||||
if training == True:
|
||||
if not fix_noise:
|
||||
eps = tf.random.normal(shape=mean.shape)
|
||||
eps = torch.randn(size=mean.shape).to(mean.device)
|
||||
else:
|
||||
eps = tf.random.normal(shape=(1, mean.shape[1]))
|
||||
eps = tf.tile(eps, [mean.shape[0],1])
|
||||
out = eps * tf.exp(logvar * .5) + mean
|
||||
eps = torch.randn(size=(1, mean.shape[1]))
|
||||
eps = torch.tile(eps, [mean.shape[0],1])
|
||||
out = eps * torch.exp(logvar * 0.5) + mean
|
||||
else:
|
||||
out = tf.exp(logvar * .5) + mean
|
||||
out = torch.exp(logvar * 0.5) + mean
|
||||
return out
|
||||
|
||||
|
||||
def encode_conds(self, uv_template, uv_body_posed, uv_curv, training):
|
||||
if self.only_curvatures:
|
||||
z_cond = self.curv_enc(uv_curv, training=training)
|
||||
def encode_conds(self, uv_template, uv_body_posed, uv_curv):
|
||||
z_body = self.body_enc(uv_body_posed)
|
||||
z_static = self.static_enc(uv_template)
|
||||
if self.use_curvature:
|
||||
z_curv = self.curv_enc(uv_curv)
|
||||
z_cond = torch.concat([z_static, z_body, z_curv], dim=1)
|
||||
else:
|
||||
z_body = self.body_enc(uv_body_posed, training=training)
|
||||
z_static = self.static_enc(uv_template, training=training)
|
||||
if self.use_curvature:
|
||||
z_curv = self.curv_enc(uv_curv, training=training)
|
||||
z_cond = tf.concat([z_static, z_body, z_curv], axis=-1)
|
||||
else:
|
||||
z_cond = tf.concat([z_static, z_body], axis=-1)
|
||||
|
||||
z_cond = self.cond_encoder(z_cond, training=training)
|
||||
z_cond = torch.concat([z_static, z_body], dim=1)
|
||||
|
||||
return z_cond
|
||||
|
||||
|
||||
def call(self, inputs, training):
|
||||
def forward(self, inputs):
|
||||
uv_unposed, uv_template, uv_body_posed, uv_curv, uv_mask = inputs
|
||||
if self.only_curvatures:
|
||||
x = tf.concat([uv_unposed, uv_curv], axis=-1)
|
||||
else:
|
||||
x = tf.concat([uv_unposed, uv_template, uv_body_posed, uv_curv], axis=-1)
|
||||
x = torch.concat([uv_unposed, uv_template, uv_body_posed, uv_curv], dim=1)
|
||||
|
||||
mean, logvar = self.encoder(x, training=training)
|
||||
z = self.reparameterize(mean, logvar, training)
|
||||
z_cond = self.encode_conds(uv_template, uv_body_posed, uv_curv, training=training)
|
||||
mean, logvar = self.encoder(x)
|
||||
z: torch.Tensor = self.reparameterize(mean, logvar)
|
||||
z = z.view((mean.shape[0], *self.latent_shape))
|
||||
z_cond = self.encode_conds(uv_template, uv_body_posed, uv_curv)
|
||||
|
||||
if self.add_noise and training == True:
|
||||
z_cond = z_cond + (tf.random.uniform(z_cond.shape, maxval=self.noise_amount * 2) - self.noise_amount)
|
||||
|
||||
|
||||
z = tf.concat([z, z_cond], axis=-1)
|
||||
x = self.decoder(z, training=training)
|
||||
z = torch.concat([z, z_cond], dim=1)
|
||||
x = self.decoder(z)
|
||||
|
||||
if self.avg_garment is not None:
|
||||
x = x + self.avg_garment[tf.newaxis,:,:,:]
|
||||
x = x + self.avg_garment[None,:,:,:]
|
||||
|
||||
if self.use_offsets_over_body:
|
||||
x = x + uv_body_posed
|
||||
|
||||
x = x * uv_mask[:,:,:,tf.newaxis]
|
||||
|
||||
return x, mean, logvar
|
||||
|
||||
|
||||
|
||||
|
||||
class CVAENew(tf.keras.Model):
|
||||
|
||||
def __init__(self, config):
|
||||
super(CVAENew, self).__init__()
|
||||
self.size = config['resolution']
|
||||
|
||||
if not('only_curvatures' in config.keys() and config['only_curvatures'] == True):
|
||||
self.static_enc : AEEncoderNew = AEEncoderNew(config['cond_ae_static']['encoder_depths'], config['cond_ae_static']['encoder_dims'], config['cond_ae_static']['kernel_size'], config['cond_ae_static']['activation'])
|
||||
if config['static_model'] != '':
|
||||
load_model(self.static_enc, config['static_model'] + '_enc')
|
||||
|
||||
self.body_enc : AEEncoderNew = AEEncoderNew(config['cond_ae_body']['encoder_depths'], config['cond_ae_body']['encoder_dims'], config['cond_ae_body']['kernel_size'])
|
||||
if config['body_model'] != '':
|
||||
load_model(self.body_enc, config['body_model'] + '_enc')
|
||||
|
||||
if config['use_curvature']:
|
||||
self.curv_enc : AEEncoderNew = AEEncoderNew(config['cond_ae_curv']['encoder_depths'], config['cond_ae_curv']['encoder_dims'], config['cond_ae_curv']['kernel_size'], config['cond_ae_curv']['activation'])
|
||||
if config['curvature_model'] != '':
|
||||
load_model(self.curv_enc, config['curvature_model'] + '_enc')
|
||||
|
||||
if config['freeze_encoders']:
|
||||
if not('only_curvatures' in config.keys() and config['only_curvatures'] == True):
|
||||
self.static_enc.trainable = False
|
||||
self.body_enc.trainable = False
|
||||
if config['use_curvature']:
|
||||
self.curv_enc.trainable = False
|
||||
|
||||
self.encoder = VariationalEncoderNew(config['cvae']['encoder_depths'], config['cvae']['encoder_dims'], kernel_size=config['cvae']['kernel_size'])
|
||||
self.decoder = DecoderNew(config['cvae']['decoder_depths'], config['cvae']['decoder_dims'], config['cvae']['kernel_size'])
|
||||
|
||||
self.reshape = tf.keras.layers.Reshape((8, 4, 64))
|
||||
|
||||
if config['avg_garment_path'] != '':
|
||||
self.avg_garment = np.load(config['avg_garment_path'])
|
||||
else:
|
||||
self.avg_garment = None
|
||||
|
||||
self.add_noise = config['cvae']['add_noise']
|
||||
self.noise_amount = config['cvae']['noise_amount']
|
||||
self.use_curvature = config['use_curvature']
|
||||
self.only_curvatures = config['only_curvatures'] if 'only_curvatures' in config.keys() and config['only_curvatures'] == True else None
|
||||
self.use_offsets_over_body = not config['use_skinning']
|
||||
|
||||
if self.use_offsets_over_body:
|
||||
assert self.avg_garment is None, 'When we are not using skinning, we can\'t use avg_garment!'
|
||||
if self.avg_garment is not None:
|
||||
assert not self.use_offsets_over_body, 'When avg_garment is given, we must use skinning!'
|
||||
|
||||
# @tf.function
|
||||
def sample(self, uv_template, uv_body_posed, uv_curv, uv_mask, noise=None, add_noise=False):
|
||||
if noise is None:
|
||||
eps = tf.random.normal(shape=(uv_template.shape[0], self.latent_dim))
|
||||
else:
|
||||
eps = noise
|
||||
|
||||
z_cond = self.encode_conds(uv_template, uv_body_posed, uv_curv, training=False)
|
||||
if add_noise:
|
||||
z_cond = z_cond + (tf.random.uniform(z_cond.shape, maxval=self.noise_amount * 2) - self.noise_amount)
|
||||
|
||||
z = tf.concat([eps, z_cond], axis=-1)
|
||||
|
||||
x = self.decoder(z, training=False)
|
||||
|
||||
if self.avg_garment is not None:
|
||||
x = x + self.avg_garment[tf.newaxis,:,:,:]
|
||||
|
||||
if self.use_offsets_over_body:
|
||||
x = x + uv_body_posed
|
||||
|
||||
x = x * uv_mask[:,:,:,tf.newaxis]
|
||||
|
||||
return x
|
||||
|
||||
|
||||
def reparameterize(self, mean, logvar, training, fix_noise=False):
|
||||
if training == True:
|
||||
if not fix_noise:
|
||||
eps = tf.random.normal(shape=mean.shape)
|
||||
else:
|
||||
eps = tf.random.normal(shape=(1, mean.shape[1]))
|
||||
eps = tf.tile(eps, [mean.shape[0],1])
|
||||
out = eps * tf.exp(logvar * .5) + mean
|
||||
else:
|
||||
out = tf.exp(logvar * .5) + mean
|
||||
return out
|
||||
|
||||
|
||||
def encode_conds(self, uv_template, uv_body_posed, uv_curv, training):
|
||||
if self.only_curvatures:
|
||||
z_cond = self.curv_enc(uv_curv, training=training)
|
||||
else:
|
||||
z_body = self.body_enc(uv_body_posed, training=training)
|
||||
z_static = self.static_enc(uv_template, training=training)
|
||||
if self.use_curvature:
|
||||
z_curv = self.curv_enc(uv_curv, training=training)
|
||||
z_cond = tf.concat([z_static, z_body, z_curv], axis=-1)
|
||||
else:
|
||||
z_cond = tf.concat([z_static, z_body], axis=-1)
|
||||
|
||||
return z_cond
|
||||
|
||||
|
||||
def call(self, inputs, training):
|
||||
uv_unposed, uv_template, uv_body_posed, uv_curv, uv_mask = inputs
|
||||
if self.only_curvatures:
|
||||
x = tf.concat([uv_unposed, uv_curv], axis=-1)
|
||||
else:
|
||||
x = tf.concat([uv_unposed, uv_template, uv_body_posed, uv_curv], axis=-1)
|
||||
|
||||
mean, logvar = self.encoder(x, training=training)
|
||||
z = self.reparameterize(mean, logvar, training)
|
||||
z = self.reshape(z)
|
||||
z_cond = self.encode_conds(uv_template, uv_body_posed, uv_curv, training=training)
|
||||
|
||||
if self.add_noise and training == True:
|
||||
z_cond = z_cond + (tf.random.uniform(z_cond.shape, maxval=self.noise_amount * 2) - self.noise_amount)
|
||||
|
||||
|
||||
z = tf.concat([z, z_cond], axis=-1)
|
||||
x = self.decoder(z, training=training)
|
||||
|
||||
if self.avg_garment is not None:
|
||||
x = x + self.avg_garment[tf.newaxis,:,:,:]
|
||||
|
||||
if self.use_offsets_over_body:
|
||||
x = x + uv_body_posed
|
||||
|
||||
x = x * uv_mask[:,:,:,tf.newaxis]
|
||||
|
||||
return x, mean, logvar
|
||||
x = x * uv_mask[:,None,:,:]
|
||||
|
||||
return x, mean, logvar
|
||||
Reference in New Issue
Block a user