250 lines
7.7 KiB
Python
250 lines
7.7 KiB
Python
|
|
|
|
import torch
|
|
import torch.nn as nn
|
|
|
|
from omegaconf import OmegaConf
|
|
from copy import deepcopy
|
|
from modules import devices, lowvram, shared, scripts
|
|
from ldm.modules.diffusionmodules.util import timestep_embedding
|
|
from ldm.modules.diffusionmodules.openaimodel import UNetModel
|
|
|
|
|
|
class TorchHijackForUnet:
|
|
"""
|
|
This is torch, but with cat that resizes tensors to appropriate dimensions if they do not match;
|
|
this makes it possible to create pictures with dimensions that are multiples of 8 rather than 64
|
|
"""
|
|
|
|
def __getattr__(self, item):
|
|
if item == 'cat':
|
|
return self.cat
|
|
|
|
if hasattr(torch, item):
|
|
return getattr(torch, item)
|
|
|
|
raise AttributeError("'{}' object has no attribute '{}'".format(type(self).__name__, item))
|
|
|
|
def cat(self, tensors, *args, **kwargs):
|
|
if len(tensors) == 2:
|
|
a, b = tensors
|
|
if a.shape[-2:] != b.shape[-2:]:
|
|
a = torch.nn.functional.interpolate(a, b.shape[-2:], mode="nearest")
|
|
|
|
tensors = (a, b)
|
|
|
|
return torch.cat(tensors, *args, **kwargs)
|
|
|
|
|
|
th = TorchHijackForUnet()
|
|
|
|
|
|
def align(hint, size):
|
|
b, c, h1, w1 = hint.shape
|
|
h, w = size
|
|
if h != h1 or w != w1:
|
|
hint = th.nn.functional.interpolate(hint, size=size, mode="nearest")
|
|
return hint
|
|
|
|
|
|
def get_node_name(name, parent_name):
|
|
if len(name) <= len(parent_name):
|
|
return False, ''
|
|
p = name[:len(parent_name)]
|
|
if p != parent_name:
|
|
return False, ''
|
|
return True, name[len(parent_name):]
|
|
|
|
|
|
class PlugableAdapter(nn.Module):
|
|
def __init__(self, state_dict, config_path, lowvram=False, base_model=None) -> None:
|
|
super().__init__()
|
|
config = OmegaConf.load(config_path)
|
|
|
|
self.control_model = Adapter(**config.model.params)
|
|
self.control_model.load_state_dict(state_dict)
|
|
self.lowvram = lowvram
|
|
self.control = None
|
|
self.hint_cond = None
|
|
|
|
if not self.lowvram:
|
|
self.control_model.to(devices.get_device_for("controlnet"))
|
|
|
|
def reset(self):
|
|
self.control = None
|
|
self.hint_cond = None
|
|
|
|
def forward(self, hint=None, *args, **kwargs):
|
|
if self.control is not None:
|
|
return deepcopy(self.control)
|
|
|
|
self.hint_cond = hint
|
|
hint_in = hint
|
|
if self.control_model.conv_in.in_channels == 64:
|
|
hint_in = hint_in[0].unsqueeze(0).unsqueeze(0)
|
|
else:
|
|
hint_in = hint_in.unsqueeze(0)
|
|
|
|
self.control = self.control_model(hint_in)
|
|
return deepcopy(self.control)
|
|
|
|
|
|
def conv_nd(dims, *args, **kwargs):
|
|
"""
|
|
Create a 1D, 2D, or 3D convolution module.
|
|
"""
|
|
if dims == 1:
|
|
return nn.Conv1d(*args, **kwargs)
|
|
elif dims == 2:
|
|
return nn.Conv2d(*args, **kwargs)
|
|
elif dims == 3:
|
|
return nn.Conv3d(*args, **kwargs)
|
|
raise ValueError(f"unsupported dimensions: {dims}")
|
|
|
|
def avg_pool_nd(dims, *args, **kwargs):
|
|
"""
|
|
Create a 1D, 2D, or 3D average pooling module.
|
|
"""
|
|
if dims == 1:
|
|
return nn.AvgPool1d(*args, **kwargs)
|
|
elif dims == 2:
|
|
return nn.AvgPool2d(*args, **kwargs)
|
|
elif dims == 3:
|
|
return nn.AvgPool3d(*args, **kwargs)
|
|
raise ValueError(f"unsupported dimensions: {dims}")
|
|
|
|
|
|
class Downsample(nn.Module):
|
|
"""
|
|
A downsampling layer with an optional convolution.
|
|
:param channels: channels in the inputs and outputs.
|
|
:param use_conv: a bool determining if a convolution is applied.
|
|
:param dims: determines if the signal is 1D, 2D, or 3D. If 3D, then
|
|
downsampling occurs in the inner-two dimensions.
|
|
"""
|
|
|
|
def __init__(self, channels, use_conv, dims=2, out_channels=None,padding=1):
|
|
super().__init__()
|
|
self.channels = channels
|
|
self.out_channels = out_channels or channels
|
|
self.use_conv = use_conv
|
|
self.dims = dims
|
|
stride = 2 if dims != 3 else (1, 2, 2)
|
|
if use_conv:
|
|
self.op = conv_nd(
|
|
dims, self.channels, self.out_channels, 3, stride=stride, padding=padding
|
|
)
|
|
else:
|
|
assert self.channels == self.out_channels
|
|
self.op = avg_pool_nd(dims, kernel_size=stride, stride=stride)
|
|
|
|
def forward(self, x):
|
|
assert x.shape[1] == self.channels
|
|
return self.op(x)
|
|
|
|
|
|
class ResnetBlock(nn.Module):
|
|
def __init__(self, in_c, out_c, down, ksize=3, sk=False, use_conv=True):
|
|
super().__init__()
|
|
ps = ksize//2
|
|
if in_c != out_c or sk==False:
|
|
self.in_conv = nn.Conv2d(in_c, out_c, ksize, 1, ps)
|
|
else:
|
|
# print('n_in')
|
|
self.in_conv = None
|
|
self.block1 = nn.Conv2d(out_c, out_c, 3, 1, 1)
|
|
self.act = nn.ReLU()
|
|
self.block2 = nn.Conv2d(out_c, out_c, ksize, 1, ps)
|
|
if sk==False:
|
|
self.skep = nn.Conv2d(in_c, out_c, ksize, 1, ps)
|
|
else:
|
|
# print('n_sk')
|
|
self.skep = None
|
|
|
|
self.down = down
|
|
if self.down == True:
|
|
self.down_opt = Downsample(in_c, use_conv=use_conv)
|
|
|
|
def forward(self, x):
|
|
if self.down == True:
|
|
x = self.down_opt(x)
|
|
if self.in_conv is not None: # edit
|
|
h = self.in_conv(x)
|
|
# x = self.in_conv(x)
|
|
# else:
|
|
# x = x
|
|
|
|
h = self.block1(h)
|
|
h = self.act(h)
|
|
h = self.block2(h)
|
|
if self.skep is not None:
|
|
return h + self.skep(x)
|
|
else:
|
|
return h + x
|
|
|
|
|
|
class ResnetBlock(nn.Module):
|
|
def __init__(self, in_c, out_c, down, ksize=3, sk=False, use_conv=True):
|
|
super().__init__()
|
|
ps = ksize//2
|
|
if in_c != out_c or sk==False:
|
|
self.in_conv = nn.Conv2d(in_c, out_c, ksize, 1, ps)
|
|
else:
|
|
# print('n_in')
|
|
self.in_conv = None
|
|
self.block1 = nn.Conv2d(out_c, out_c, 3, 1, 1)
|
|
self.act = nn.ReLU()
|
|
self.block2 = nn.Conv2d(out_c, out_c, ksize, 1, ps)
|
|
if sk==False:
|
|
self.skep = nn.Conv2d(in_c, out_c, ksize, 1, ps)
|
|
else:
|
|
self.skep = None
|
|
|
|
self.down = down
|
|
if self.down == True:
|
|
self.down_opt = Downsample(in_c, use_conv=use_conv)
|
|
|
|
def forward(self, x):
|
|
if self.down == True:
|
|
x = self.down_opt(x)
|
|
if self.in_conv is not None: # edit
|
|
x = self.in_conv(x)
|
|
|
|
h = self.block1(x)
|
|
h = self.act(h)
|
|
h = self.block2(h)
|
|
if self.skep is not None:
|
|
return h + self.skep(x)
|
|
else:
|
|
return h + x
|
|
|
|
|
|
class Adapter(nn.Module):
|
|
def __init__(self, channels=[320, 640, 1280, 1280], nums_rb=3, cin=64, ksize=3, sk=False, use_conv=True):
|
|
super(Adapter, self).__init__()
|
|
self.unshuffle = nn.PixelUnshuffle(8)
|
|
self.channels = channels
|
|
self.nums_rb = nums_rb
|
|
self.body = []
|
|
for i in range(len(channels)):
|
|
for j in range(nums_rb):
|
|
if (i!=0) and (j==0):
|
|
self.body.append(ResnetBlock(channels[i-1], channels[i], down=True, ksize=ksize, sk=sk, use_conv=use_conv))
|
|
else:
|
|
self.body.append(ResnetBlock(channels[i], channels[i], down=False, ksize=ksize, sk=sk, use_conv=use_conv))
|
|
self.body = nn.ModuleList(self.body)
|
|
self.conv_in = nn.Conv2d(cin, channels[0], 3, 1, 1)
|
|
|
|
def forward(self, x):
|
|
# unshuffle
|
|
x = self.unshuffle(x)
|
|
# extract features
|
|
features = []
|
|
x = self.conv_in(x)
|
|
for i in range(len(self.channels)):
|
|
for j in range(self.nums_rb):
|
|
idx = i*self.nums_rb +j
|
|
x = self.body[idx](x)
|
|
features.append(x)
|
|
|
|
return features |