mirror of
https://github.com/vladmandic/automatic
synced 2026-09-19 01:04:32 +02:00
implement complete face module
This commit is contained in:
Vladimir Mandic
@@ -1,273 +0,0 @@
|
||||
import os
|
||||
import cv2
|
||||
import torch
|
||||
import numpy as np
|
||||
import gradio as gr
|
||||
import diffusers
|
||||
import huggingface_hub as hf
|
||||
from PIL import Image
|
||||
from modules import scripts, processing, shared, devices, images
|
||||
|
||||
|
||||
debug = shared.log.trace if os.environ.get('SD_FACEID_DEBUG', None) is not None else lambda *args, **kwargs: None
|
||||
MODELS = {
|
||||
'FaceID Base': 'h94/IP-Adapter-FaceID/ip-adapter-faceid_sd15.bin',
|
||||
'FaceID Plus': 'h94/IP-Adapter-FaceID/ip-adapter-faceid-plus_sd15.bin',
|
||||
'FaceID Plus v2': 'h94/IP-Adapter-FaceID/ip-adapter-faceid-plusv2_sd15.bin',
|
||||
'FaceID XL': 'h94/IP-Adapter-FaceID/ip-adapter-faceid_sdxl.bin'
|
||||
}
|
||||
app = None
|
||||
ip_model = None
|
||||
ip_model_name = None
|
||||
ip_model_tokens = None
|
||||
ip_model_rank = None
|
||||
swapper = None
|
||||
|
||||
|
||||
def dependencies():
|
||||
from installer import installed, install
|
||||
packages = [
|
||||
('insightface', 'insightface'),
|
||||
('git+https://github.com/tencent-ailab/IP-Adapter.git', 'ip_adapter'),
|
||||
]
|
||||
for pkg in packages:
|
||||
if not installed(pkg[1], reload=False, quiet=True):
|
||||
install(pkg[0], pkg[1], ignore=True)
|
||||
|
||||
|
||||
def face_id(p: processing.StableDiffusionProcessing, faces, image, model, override, tokens, rank, cache, scale, structure):
|
||||
global ip_model, ip_model_name, ip_model_tokens, ip_model_rank # pylint: disable=global-statement
|
||||
from insightface.utils import face_align
|
||||
from ip_adapter.ip_adapter_faceid import IPAdapterFaceID, IPAdapterFaceIDPlus, IPAdapterFaceIDXL
|
||||
|
||||
face_embeds = torch.from_numpy(faces[0].normed_embedding).unsqueeze(0)
|
||||
face_image = face_align.norm_crop(image, landmark=faces[0].kps, image_size=224) # you can also segment the face
|
||||
|
||||
ip_ckpt = MODELS[model]
|
||||
folder, filename = os.path.split(ip_ckpt)
|
||||
basename, _ext = os.path.splitext(filename)
|
||||
model_path = hf.hf_hub_download(repo_id=folder, filename=filename, cache_dir=shared.opts.diffusers_dir)
|
||||
if model_path is None:
|
||||
shared.log.error(f'FaceID download failed: model={model} file={ip_ckpt}')
|
||||
return None
|
||||
|
||||
processing.process_init(p)
|
||||
if override:
|
||||
shared.sd_model.scheduler = diffusers.DDIMScheduler(
|
||||
num_train_timesteps=1000,
|
||||
beta_start=0.00085,
|
||||
beta_end=0.012,
|
||||
beta_schedule="scaled_linear",
|
||||
clip_sample=False,
|
||||
set_alpha_to_one=False,
|
||||
steps_offset=1,
|
||||
)
|
||||
shortcut = None
|
||||
if ip_model is None or ip_model_name != model or ip_model_tokens != tokens or ip_model_rank != rank or not cache:
|
||||
shared.log.debug(f'FaceID load: model={model} file={ip_ckpt} tokens={tokens} rank={rank}')
|
||||
if 'Plus' in model:
|
||||
image_encoder_path = "laion/CLIP-ViT-H-14-laion2B-s32B-b79K"
|
||||
ip_model = IPAdapterFaceIDPlus(
|
||||
sd_pipe=shared.sd_model,
|
||||
image_encoder_path=image_encoder_path,
|
||||
ip_ckpt=model_path,
|
||||
lora_rank=rank,
|
||||
num_tokens=tokens,
|
||||
device=devices.device,
|
||||
torch_dtype=devices.dtype,
|
||||
)
|
||||
shortcut = 'v2' in model
|
||||
elif 'XL' in model:
|
||||
ip_model = IPAdapterFaceIDXL(
|
||||
sd_pipe=shared.sd_model,
|
||||
ip_ckpt=model_path,
|
||||
lora_rank=rank,
|
||||
num_tokens=tokens,
|
||||
device=devices.device,
|
||||
torch_dtype=devices.dtype,
|
||||
)
|
||||
else:
|
||||
ip_model = IPAdapterFaceID(
|
||||
sd_pipe=shared.sd_model,
|
||||
ip_ckpt=model_path,
|
||||
lora_rank=rank,
|
||||
num_tokens=tokens,
|
||||
device=devices.device,
|
||||
torch_dtype=devices.dtype,
|
||||
)
|
||||
ip_model_name = model
|
||||
ip_model_tokens = tokens
|
||||
ip_model_rank = rank
|
||||
else:
|
||||
shared.log.debug(f'FaceID cached: model={model} file={ip_ckpt} tokens={tokens} rank={rank}')
|
||||
|
||||
# main generate dict
|
||||
ip_model_dict = {
|
||||
'num_samples': p.batch_size,
|
||||
'width': p.width,
|
||||
'height': p.height,
|
||||
'num_inference_steps': p.steps,
|
||||
'scale': scale,
|
||||
'guidance_scale': p.cfg_scale,
|
||||
'faceid_embeds': face_embeds.shape,
|
||||
}
|
||||
|
||||
# optional generate dict
|
||||
if shortcut is not None:
|
||||
ip_model_dict['shortcut'] = shortcut
|
||||
if 'Plus' in model:
|
||||
ip_model_dict['s_scale'] = structure
|
||||
ip_model_dict['face_image'] = face_image.shape
|
||||
shared.log.debug(f'FaceID args: {ip_model_dict}')
|
||||
if 'Plus' in model:
|
||||
ip_model_dict['face_image'] = face_image
|
||||
ip_model_dict['faceid_embeds'] = face_embeds
|
||||
|
||||
# run generate
|
||||
processed_images = []
|
||||
ip_model.set_scale(scale)
|
||||
for i in range(p.n_iter):
|
||||
ip_model_dict.update(
|
||||
{
|
||||
'prompt': p.all_prompts[i],
|
||||
'negative_prompt': p.all_negative_prompts[i],
|
||||
'seed': int(p.all_seeds[i]),
|
||||
}
|
||||
)
|
||||
debug(f'FaceID: {ip_model_dict}')
|
||||
res = ip_model.generate(**ip_model_dict)
|
||||
if isinstance(res, list):
|
||||
processed_images += res
|
||||
ip_model.set_scale(0)
|
||||
|
||||
if not cache:
|
||||
ip_model = None
|
||||
ip_model_name = None
|
||||
devices.torch_gc()
|
||||
|
||||
p.extra_generation_params["IP Adapter"] = f'{basename}:{scale}'
|
||||
return processed_images
|
||||
|
||||
|
||||
def face_swap(p: processing.StableDiffusionProcessing, image, source_face):
|
||||
import insightface.model_zoo
|
||||
global swapper # pylint: disable=global-statement
|
||||
if swapper is None:
|
||||
model_path = hf.hf_hub_download(repo_id='ezioruan/inswapper_128.onnx', filename='inswapper_128.onnx', cache_dir=shared.opts.diffusers_dir)
|
||||
router = insightface.model_zoo.model_zoo.ModelRouter(model_path)
|
||||
swapper = router.get_model()
|
||||
|
||||
np_image = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
|
||||
faces = app.get(np_image)
|
||||
|
||||
res = np_image.copy()
|
||||
for target_face in faces:
|
||||
res = swapper.get(res, target_face, source_face, paste_back=True) # pylint: disable=too-many-function-args, unexpected-keyword-arg
|
||||
|
||||
p.extra_generation_params["FaceSwap"] = f'{len(faces)}'
|
||||
np_image = cv2.cvtColor(res, cv2.COLOR_BGR2RGB)
|
||||
return Image.fromarray(np_image)
|
||||
|
||||
|
||||
class Script(scripts.Script):
|
||||
def title(self):
|
||||
return 'FaceID'
|
||||
|
||||
def show(self, is_img2img):
|
||||
return True if shared.backend == shared.Backend.DIFFUSERS else False
|
||||
|
||||
# return signature is array of gradio components
|
||||
def ui(self, _is_img2img):
|
||||
with gr.Row():
|
||||
mode = gr.CheckboxGroup(label='Mode', choices=['FaceID', 'FaceSwap'], value=['FaceID'])
|
||||
model = gr.Dropdown(choices=list(MODELS), label='FaceID Model', value='FaceID Base')
|
||||
with gr.Row(visible=True):
|
||||
override = gr.Checkbox(label='Override sampler', value=True)
|
||||
cache = gr.Checkbox(label='Cache model', value=True)
|
||||
with gr.Row(visible=True):
|
||||
scale = gr.Slider(label='Strength', minimum=0.0, maximum=2.0, step=0.01, value=1.0)
|
||||
structure = gr.Slider(label='Structure', minimum=0.0, maximum=1.0, step=0.01, value=1.0)
|
||||
with gr.Row(visible=False):
|
||||
rank = gr.Slider(label='Rank', minimum=4, maximum=256, step=4, value=128)
|
||||
tokens = gr.Slider(label='Tokens', minimum=1, maximum=16, step=1, value=4)
|
||||
with gr.Row():
|
||||
image = gr.Image(image_mode='RGB', label='Image', source='upload', type='pil', width=512)
|
||||
return [mode, model, scale, image, override, rank, tokens, structure, cache]
|
||||
|
||||
def run(self, p: processing.StableDiffusionProcessing, mode, model, scale, image, override, rank, tokens, structure, cache): # pylint: disable=arguments-differ, unused-argument
|
||||
if len(mode) == 0:
|
||||
return None
|
||||
dependencies()
|
||||
try:
|
||||
import onnxruntime
|
||||
from insightface.app import FaceAnalysis
|
||||
except Exception as e:
|
||||
shared.log.error(f'FaceID: {e}')
|
||||
return None
|
||||
if image is None:
|
||||
shared.log.error('FaceID: no init_images')
|
||||
return None
|
||||
if shared.sd_model_type != 'sd' and shared.sd_model_type != 'sdxl':
|
||||
shared.log.error('FaceID: base model not supported')
|
||||
return None
|
||||
|
||||
global app # pylint: disable=global-statement
|
||||
if app is None:
|
||||
shared.log.debug(f"ONNX: device={onnxruntime.get_device()} providers={onnxruntime.get_available_providers()}")
|
||||
app = FaceAnalysis(name="buffalo_l", providers=['CUDAExecutionProvider', 'CPUExecutionProvider'])
|
||||
onnxruntime.set_default_logger_severity(3)
|
||||
app.prepare(ctx_id=0, det_thresh=0.5, det_size=(640, 640))
|
||||
|
||||
if isinstance(image, str):
|
||||
from modules.api.api import decode_base64_to_image
|
||||
image = decode_base64_to_image(image).convert("RGB")
|
||||
|
||||
np_image = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
|
||||
faces = app.get(np_image)
|
||||
if len(faces) == 0:
|
||||
shared.log.error('FaceID: no faces found')
|
||||
return None
|
||||
for i, face in enumerate(faces):
|
||||
shared.log.debug(f'FaceID face: i={i+1} score={face.det_score:.2f} gender={"female" if face.gender==0 else "male"} age={face.age} bbox={face.bbox}')
|
||||
p.extra_generation_params[f"FaceID {i+1}"] = f'{face.det_score:.2f} {"female" if face.gender==0 else "male"} {face.age}y'
|
||||
|
||||
processed_images = []
|
||||
if 'FaceID' in mode:
|
||||
processed_images = face_id(p, faces, np_image, model, override, tokens, rank, cache, scale, structure) # run faceid pipeline
|
||||
processed = processing.Processed(
|
||||
p,
|
||||
images_list=processed_images,
|
||||
seed=p.seed,
|
||||
subseed=p.subseed,
|
||||
index_of_first_image=0,
|
||||
)
|
||||
if 'FaceSwap' not in mode:
|
||||
if shared.opts.samples_save and not p.do_not_save_samples:
|
||||
for i, image in enumerate(processed.images):
|
||||
info = processing.create_infotext(p, index=i)
|
||||
images.save_image(image, path=p.outpath_samples, seed=p.all_seeds[i], prompt=p.all_prompts[i], info=info, p=p)
|
||||
else:
|
||||
if shared.opts.save_images_before_face_restoration and not p.do_not_save_samples:
|
||||
for i, image in enumerate(processed.images):
|
||||
info = processing.create_infotext(p, index=i)
|
||||
images.save_image(image, path=p.outpath_samples, seed=p.all_seeds[i], prompt=p.all_prompts[i], info=info, p=p, suffix="-before-face-swap")
|
||||
|
||||
else:
|
||||
processed = processing.process_images(p) # run normal pipeline
|
||||
processed_images = processed.images
|
||||
|
||||
if 'FaceSwap' in mode: # replace faces as postprocess
|
||||
processed.images = []
|
||||
for batch_image in processed_images:
|
||||
swapped_image = face_swap(p, batch_image, source_face=faces[0])
|
||||
processed.images.append(swapped_image)
|
||||
|
||||
if shared.opts.samples_save and not p.do_not_save_samples:
|
||||
for i, image in enumerate(processed.images):
|
||||
info = processing.create_infotext(p, index=i)
|
||||
images.save_image(image, path=p.outpath_samples, seed=p.all_seeds[i], prompt=p.all_prompts[i], info=info, p=p)
|
||||
|
||||
processed.info = processed.infotext(p, 0)
|
||||
processed.infotexts = [processed.info]
|
||||
|
||||
return processed
|
||||
@@ -1,118 +0,0 @@
|
||||
import os
|
||||
import gradio as gr
|
||||
import huggingface_hub as hf
|
||||
from PIL import Image
|
||||
from modules import shared, processing, sd_models, scripts
|
||||
|
||||
|
||||
class Script(scripts.Script):
|
||||
def title(self):
|
||||
return 'PhotoMaker'
|
||||
|
||||
def show(self, is_img2img):
|
||||
return True if shared.backend == shared.Backend.DIFFUSERS else False
|
||||
|
||||
def load_images(self, files):
|
||||
init_images = []
|
||||
for file in files or []:
|
||||
try:
|
||||
if isinstance(file, str):
|
||||
from modules.api.api import decode_base64_to_image
|
||||
image = decode_base64_to_image(file)
|
||||
elif isinstance(file, Image.Image):
|
||||
image = file
|
||||
elif isinstance(file, dict) and 'name' in file:
|
||||
image = Image.open(file['name']) # _TemporaryFileWrapper from gr.Files
|
||||
elif hasattr(file, 'name'):
|
||||
image = Image.open(file.name) # _TemporaryFileWrapper from gr.Files
|
||||
else:
|
||||
raise ValueError(f'PhotoMaker unknown input: {file}')
|
||||
init_images.append(image)
|
||||
except Exception as e:
|
||||
shared.log.warning(f'PhotoMaker failed to load image: {e}')
|
||||
return init_images
|
||||
|
||||
def ui(self, _is_img2img):
|
||||
with gr.Row():
|
||||
trigger = gr.Text(label='Trigger word', value="person")
|
||||
strength = gr.Slider(label='Strength', minimum=0.0, maximum=2.0, step=0.01, value=1.0)
|
||||
start = gr.Slider(label='Start', minimum=0.0, maximum=1.0, step=0.01, value=0.5)
|
||||
with gr.Row():
|
||||
files = gr.File(label='Input images', file_count='multiple', file_types=['image'], type='file', interactive=True, height=100)
|
||||
with gr.Row():
|
||||
gallery = gr.Gallery(show_label=False, value=[])
|
||||
with gr.Row():
|
||||
gr.HTML('<a href="https://github.com/TencentARC/PhotoMaker>PhotoMaker</a>"')
|
||||
files.change(fn=self.load_images, inputs=[files], outputs=[gallery])
|
||||
return [trigger, strength, start, gallery]
|
||||
|
||||
# Run pipeline
|
||||
def run(self, p: processing.StableDiffusionProcessing, trigger, strength, start, images): # pylint: disable=arguments-differ
|
||||
from scripts.photomaker_model import PhotoMakerStableDiffusionXLPipeline # pylint: disable=no-name-in-module
|
||||
|
||||
# prepare pipeline
|
||||
input_images = self.load_images(images)
|
||||
if len(input_images) == 0:
|
||||
shared.log.warning('PhotoMaker: no input images')
|
||||
return None
|
||||
|
||||
c = shared.sd_model.__class__.__name__ if shared.sd_model is not None else ''
|
||||
if c != 'StableDiffusionXLPipeline':
|
||||
shared.log.warning(f'PhotoMaker invalid base model: current={c} required=StableDiffusionXLPipeline')
|
||||
return None
|
||||
|
||||
# validate prompt
|
||||
trigger_ids = shared.sd_model.tokenizer.encode(trigger) + shared.sd_model.tokenizer_2.encode(trigger)
|
||||
prompt_ids1 = shared.sd_model.tokenizer.encode(p.prompt)
|
||||
prompt_ids2 = shared.sd_model.tokenizer_2.encode(p.prompt)
|
||||
for t in trigger_ids:
|
||||
if prompt_ids1.count(t) != 1:
|
||||
shared.log.error(f'PhotoMaker: trigger word not matched in prompt: {trigger} ids={trigger_ids} prompt={p.prompt} ids={prompt_ids1}')
|
||||
return None
|
||||
if prompt_ids2.count(t) != 1:
|
||||
shared.log.error(f'PhotoMaker: trigger word not matched in prompt: {trigger} ids={trigger_ids} prompt={p.prompt} ids={prompt_ids1}')
|
||||
return None
|
||||
|
||||
# create new pipeline
|
||||
orig_pipeline = shared.sd_model # backup current pipeline definition
|
||||
shared.sd_model = PhotoMakerStableDiffusionXLPipeline(
|
||||
vae = shared.sd_model.vae,
|
||||
text_encoder=shared.sd_model.text_encoder,
|
||||
text_encoder_2=shared.sd_model.text_encoder_2,
|
||||
tokenizer=shared.sd_model.tokenizer,
|
||||
tokenizer_2=shared.sd_model.tokenizer_2,
|
||||
unet=shared.sd_model.unet,
|
||||
scheduler=shared.sd_model.scheduler,
|
||||
force_zeros_for_empty_prompt=shared.opts.diffusers_force_zeros,
|
||||
)
|
||||
sd_models.copy_diffuser_options(shared.sd_model, orig_pipeline) # copy options from original pipeline
|
||||
sd_models.set_diffuser_options(shared.sd_model) # set all model options such as fp16, offload, etc.
|
||||
if not ((shared.opts.diffusers_model_cpu_offload or shared.cmd_opts.medvram) or (shared.opts.diffusers_seq_cpu_offload or shared.cmd_opts.lowvram)):
|
||||
shared.sd_model.to(shared.device) # move pipeline if needed, but don't touch if its under automatic managment
|
||||
|
||||
orig_prompt_attention = shared.opts.prompt_attention
|
||||
shared.opts.data['prompt_attention'] = 'Fixed attention' # otherwise need to deal with class_tokens_mask
|
||||
p.task_args['input_id_images'] = input_images
|
||||
p.task_args['start_merge_step'] = int(start * p.steps)
|
||||
p.task_args['prompt'] = p.prompt # override all logic
|
||||
|
||||
photomaker_path = hf.hf_hub_download(repo_id="TencentARC/PhotoMaker", filename="photomaker-v1.bin", repo_type="model", cache_dir=shared.opts.diffusers_dir)
|
||||
shared.log.debug(f'PhotoMaker: model={photomaker_path} images={len(input_images)} trigger={trigger} args={p.task_args}')
|
||||
|
||||
# load photomaker adapter
|
||||
shared.sd_model.load_photomaker_adapter(
|
||||
os.path.dirname(photomaker_path),
|
||||
subfolder="",
|
||||
weight_name=os.path.basename(photomaker_path),
|
||||
trigger_word=trigger
|
||||
)
|
||||
shared.sd_model.set_adapters(["photomaker"], adapter_weights=[strength])
|
||||
|
||||
# run processing
|
||||
processed: processing.Processed = processing.process_images(p)
|
||||
p.extra_generation_params['PhotoMaker'] = f'{strength}'
|
||||
|
||||
# restore original pipeline
|
||||
shared.opts.data['prompt_attention'] = orig_prompt_attention
|
||||
shared.sd_model = orig_pipeline
|
||||
return processed
|
||||
@@ -1,555 +0,0 @@
|
||||
from typing import Any, Callable, Dict, List, Optional, Union, Tuple
|
||||
import PIL
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
from safetensors import safe_open
|
||||
from huggingface_hub.utils import validate_hf_hub_args
|
||||
from diffusers import StableDiffusionXLPipeline
|
||||
from diffusers.pipelines.stable_diffusion_xl.pipeline_output import StableDiffusionXLPipelineOutput
|
||||
from diffusers.utils import _get_model_file
|
||||
from transformers import CLIPImageProcessor
|
||||
from transformers.models.clip.modeling_clip import CLIPVisionModelWithProjection
|
||||
from transformers.models.clip.configuration_clip import CLIPVisionConfig
|
||||
|
||||
|
||||
PipelineImageInput = Union[
|
||||
PIL.Image.Image,
|
||||
torch.FloatTensor,
|
||||
List[PIL.Image.Image],
|
||||
List[torch.FloatTensor],
|
||||
]
|
||||
|
||||
|
||||
VISION_CONFIG_DICT = {
|
||||
"hidden_size": 1024,
|
||||
"intermediate_size": 4096,
|
||||
"num_attention_heads": 16,
|
||||
"num_hidden_layers": 24,
|
||||
"patch_size": 14,
|
||||
"projection_dim": 768
|
||||
}
|
||||
|
||||
|
||||
# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.rescale_noise_cfg
|
||||
def rescale_noise_cfg(noise_cfg, noise_pred_text, guidance_rescale=0.0):
|
||||
"""
|
||||
Rescale `noise_cfg` according to `guidance_rescale`. Based on findings of [Common Diffusion Noise Schedules and
|
||||
Sample Steps are Flawed](https://arxiv.org/pdf/2305.08891.pdf). See Section 3.4
|
||||
"""
|
||||
std_text = noise_pred_text.std(dim=list(range(1, noise_pred_text.ndim)), keepdim=True)
|
||||
std_cfg = noise_cfg.std(dim=list(range(1, noise_cfg.ndim)), keepdim=True)
|
||||
# rescale the results from guidance (fixes overexposure)
|
||||
noise_pred_rescaled = noise_cfg * (std_text / std_cfg)
|
||||
# mix with the original results from guidance by factor guidance_rescale to avoid "plain looking" images
|
||||
noise_cfg = guidance_rescale * noise_pred_rescaled + (1 - guidance_rescale) * noise_cfg
|
||||
return noise_cfg
|
||||
|
||||
|
||||
class MLP(nn.Module):
|
||||
def __init__(self, in_dim, out_dim, hidden_dim, use_residual=True):
|
||||
super().__init__()
|
||||
if use_residual:
|
||||
assert in_dim == out_dim
|
||||
self.layernorm = nn.LayerNorm(in_dim)
|
||||
self.fc1 = nn.Linear(in_dim, hidden_dim)
|
||||
self.fc2 = nn.Linear(hidden_dim, out_dim)
|
||||
self.use_residual = use_residual
|
||||
self.act_fn = nn.GELU()
|
||||
|
||||
def forward(self, x):
|
||||
residual = x
|
||||
x = self.layernorm(x)
|
||||
x = self.fc1(x)
|
||||
x = self.act_fn(x)
|
||||
x = self.fc2(x)
|
||||
if self.use_residual:
|
||||
x = x + residual
|
||||
return x
|
||||
|
||||
|
||||
class FuseModule(nn.Module):
|
||||
def __init__(self, embed_dim):
|
||||
super().__init__()
|
||||
self.mlp1 = MLP(embed_dim * 2, embed_dim, embed_dim, use_residual=False)
|
||||
self.mlp2 = MLP(embed_dim, embed_dim, embed_dim, use_residual=True)
|
||||
self.layer_norm = nn.LayerNorm(embed_dim)
|
||||
|
||||
def fuse_fn(self, prompt_embeds, id_embeds):
|
||||
stacked_id_embeds = torch.cat([prompt_embeds, id_embeds], dim=-1)
|
||||
stacked_id_embeds = self.mlp1(stacked_id_embeds) + prompt_embeds
|
||||
stacked_id_embeds = self.mlp2(stacked_id_embeds)
|
||||
stacked_id_embeds = self.layer_norm(stacked_id_embeds)
|
||||
return stacked_id_embeds
|
||||
|
||||
def forward(
|
||||
self,
|
||||
prompt_embeds,
|
||||
id_embeds,
|
||||
class_tokens_mask,
|
||||
) -> torch.Tensor:
|
||||
# id_embeds shape: [b, max_num_inputs, 1, 2048]
|
||||
id_embeds = id_embeds.to(prompt_embeds.dtype)
|
||||
num_inputs = class_tokens_mask.sum().unsqueeze(0)
|
||||
batch_size, max_num_inputs = id_embeds.shape[:2]
|
||||
# seq_length: 77
|
||||
seq_length = prompt_embeds.shape[1]
|
||||
# flat_id_embeds shape: [b*max_num_inputs, 1, 2048]
|
||||
flat_id_embeds = id_embeds.view(
|
||||
-1, id_embeds.shape[-2], id_embeds.shape[-1]
|
||||
)
|
||||
# valid_id_mask [b*max_num_inputs]
|
||||
valid_id_mask = (
|
||||
torch.arange(max_num_inputs, device=flat_id_embeds.device)[None, :]
|
||||
< num_inputs[:, None]
|
||||
)
|
||||
valid_id_embeds = flat_id_embeds[valid_id_mask.flatten()]
|
||||
|
||||
prompt_embeds = prompt_embeds.view(-1, prompt_embeds.shape[-1])
|
||||
class_tokens_mask = class_tokens_mask.view(-1)
|
||||
valid_id_embeds = valid_id_embeds.view(-1, valid_id_embeds.shape[-1])
|
||||
# slice out the image token embeddings
|
||||
image_token_embeds = prompt_embeds[class_tokens_mask]
|
||||
stacked_id_embeds = self.fuse_fn(image_token_embeds, valid_id_embeds)
|
||||
assert class_tokens_mask.sum() == stacked_id_embeds.shape[0], f"{class_tokens_mask.sum()} != {stacked_id_embeds.shape[0]}"
|
||||
prompt_embeds.masked_scatter_(class_tokens_mask[:, None], stacked_id_embeds.to(prompt_embeds.dtype))
|
||||
updated_prompt_embeds = prompt_embeds.view(batch_size, seq_length, -1)
|
||||
return updated_prompt_embeds
|
||||
|
||||
class PhotoMakerIDEncoder(CLIPVisionModelWithProjection):
|
||||
def __init__(self):
|
||||
super().__init__(CLIPVisionConfig(**VISION_CONFIG_DICT))
|
||||
self.visual_projection_2 = nn.Linear(1024, 1280, bias=False)
|
||||
self.fuse_module = FuseModule(2048)
|
||||
|
||||
def forward(self, id_pixel_values, prompt_embeds, class_tokens_mask):
|
||||
b, num_inputs, c, h, w = id_pixel_values.shape
|
||||
id_pixel_values = id_pixel_values.view(b * num_inputs, c, h, w)
|
||||
|
||||
shared_id_embeds = self.vision_model(id_pixel_values)[1]
|
||||
id_embeds = self.visual_projection(shared_id_embeds)
|
||||
id_embeds_2 = self.visual_projection_2(shared_id_embeds)
|
||||
|
||||
id_embeds = id_embeds.view(b, num_inputs, 1, -1)
|
||||
id_embeds_2 = id_embeds_2.view(b, num_inputs, 1, -1)
|
||||
|
||||
id_embeds = torch.cat((id_embeds, id_embeds_2), dim=-1)
|
||||
updated_prompt_embeds = self.fuse_module(prompt_embeds, id_embeds, class_tokens_mask)
|
||||
|
||||
return updated_prompt_embeds
|
||||
|
||||
|
||||
class PhotoMakerStableDiffusionXLPipeline(StableDiffusionXLPipeline):
|
||||
@validate_hf_hub_args
|
||||
def load_photomaker_adapter(
|
||||
self,
|
||||
pretrained_model_name_or_path_or_dict: Union[str, Dict[str, torch.Tensor]],
|
||||
weight_name: str,
|
||||
subfolder: str = '',
|
||||
trigger_word: str = 'img',
|
||||
**kwargs,
|
||||
):
|
||||
# Load the main state dict first.
|
||||
cache_dir = kwargs.pop("cache_dir", None)
|
||||
force_download = kwargs.pop("force_download", False)
|
||||
resume_download = kwargs.pop("resume_download", False)
|
||||
proxies = kwargs.pop("proxies", None)
|
||||
local_files_only = kwargs.pop("local_files_only", None)
|
||||
token = kwargs.pop("token", None)
|
||||
revision = kwargs.pop("revision", None)
|
||||
|
||||
user_agent = {
|
||||
"file_type": "attn_procs_weights",
|
||||
"framework": "pytorch",
|
||||
}
|
||||
|
||||
if not isinstance(pretrained_model_name_or_path_or_dict, dict):
|
||||
model_file = _get_model_file(
|
||||
pretrained_model_name_or_path_or_dict,
|
||||
weights_name=weight_name,
|
||||
cache_dir=cache_dir,
|
||||
force_download=force_download,
|
||||
resume_download=resume_download,
|
||||
proxies=proxies,
|
||||
local_files_only=local_files_only,
|
||||
token=token,
|
||||
revision=revision,
|
||||
subfolder=subfolder,
|
||||
user_agent=user_agent,
|
||||
)
|
||||
if weight_name.endswith(".safetensors"):
|
||||
state_dict = {"id_encoder": {}, "lora_weights": {}}
|
||||
with safe_open(model_file, framework="pt", device="cpu") as f:
|
||||
for key in f.keys():
|
||||
if key.startswith("id_encoder."):
|
||||
state_dict["id_encoder"][key.replace("id_encoder.", "")] = f.get_tensor(key)
|
||||
elif key.startswith("lora_weights."):
|
||||
state_dict["lora_weights"][key.replace("lora_weights.", "")] = f.get_tensor(key)
|
||||
else:
|
||||
state_dict = torch.load(model_file, map_location="cpu")
|
||||
else:
|
||||
state_dict = pretrained_model_name_or_path_or_dict
|
||||
|
||||
keys = list(state_dict.keys())
|
||||
if keys != ["id_encoder", "lora_weights"]:
|
||||
raise ValueError("Required keys are (`id_encoder` and `lora_weights`) missing from the state dict.")
|
||||
|
||||
self.trigger_word = trigger_word
|
||||
# load finetuned CLIP image encoder and fuse module here if it has not been registered to the pipeline yet
|
||||
id_encoder = PhotoMakerIDEncoder()
|
||||
id_encoder.load_state_dict(state_dict["id_encoder"], strict=True)
|
||||
id_encoder = id_encoder.to(self.device, dtype=self.unet.dtype)
|
||||
self.id_encoder = id_encoder
|
||||
self.id_image_processor = CLIPImageProcessor()
|
||||
|
||||
# load lora into models
|
||||
self.load_lora_weights(state_dict["lora_weights"], adapter_name="photomaker")
|
||||
|
||||
# Add trigger word token
|
||||
if self.tokenizer is not None:
|
||||
self.tokenizer.add_tokens([self.trigger_word], special_tokens=True)
|
||||
self.tokenizer_2.add_tokens([self.trigger_word], special_tokens=True)
|
||||
|
||||
def encode_prompt_with_trigger_word(
|
||||
self,
|
||||
prompt: str,
|
||||
prompt_2: Optional[str] = None,
|
||||
num_id_images: int = 1,
|
||||
device: Optional[torch.device] = None,
|
||||
prompt_embeds: Optional[torch.FloatTensor] = None,
|
||||
pooled_prompt_embeds: Optional[torch.FloatTensor] = None,
|
||||
class_tokens_mask: Optional[torch.LongTensor] = None,
|
||||
):
|
||||
device = device or self._execution_device
|
||||
|
||||
"""
|
||||
if prompt is not None and isinstance(prompt, str):
|
||||
batch_size = 1
|
||||
elif prompt is not None and isinstance(prompt, list):
|
||||
batch_size = len(prompt)
|
||||
else:
|
||||
batch_size = prompt_embeds.shape[0]
|
||||
"""
|
||||
|
||||
# Find the token id of the trigger word
|
||||
image_token_id = self.tokenizer_2.convert_tokens_to_ids(self.trigger_word)
|
||||
|
||||
# Define tokenizers and text encoders
|
||||
tokenizers = [self.tokenizer, self.tokenizer_2] if self.tokenizer is not None else [self.tokenizer_2]
|
||||
text_encoders = (
|
||||
[self.text_encoder, self.text_encoder_2] if self.text_encoder is not None else [self.text_encoder_2]
|
||||
)
|
||||
|
||||
if prompt_embeds is None:
|
||||
prompt_2 = prompt_2 or prompt
|
||||
prompt_embeds_list = []
|
||||
prompts = [prompt, prompt_2]
|
||||
for prompt, tokenizer, text_encoder in zip(prompts, tokenizers, text_encoders):
|
||||
input_ids = tokenizer.encode(prompt) # TODO: batch encode
|
||||
clean_index = 0
|
||||
clean_input_ids = []
|
||||
class_token_index = []
|
||||
# Find out the corrresponding class word token based on the newly added trigger word token
|
||||
for _i, token_id in enumerate(input_ids):
|
||||
if token_id == image_token_id:
|
||||
class_token_index.append(clean_index - 1)
|
||||
else:
|
||||
clean_input_ids.append(token_id)
|
||||
clean_index += 1
|
||||
|
||||
if len(class_token_index) != 1:
|
||||
raise ValueError(
|
||||
f"PhotoMaker currently does not support multiple trigger words in a single prompt.\
|
||||
Trigger word: {self.trigger_word}, Prompt: {prompt}."
|
||||
)
|
||||
class_token_index = class_token_index[0]
|
||||
|
||||
# Expand the class word token and corresponding mask
|
||||
class_token = clean_input_ids[class_token_index]
|
||||
clean_input_ids = clean_input_ids[:class_token_index] + [class_token] * num_id_images + \
|
||||
clean_input_ids[class_token_index+1:]
|
||||
|
||||
# Truncation or padding
|
||||
max_len = tokenizer.model_max_length
|
||||
if len(clean_input_ids) > max_len:
|
||||
clean_input_ids = clean_input_ids[:max_len]
|
||||
else:
|
||||
clean_input_ids = clean_input_ids + [tokenizer.pad_token_id] * (
|
||||
max_len - len(clean_input_ids)
|
||||
)
|
||||
|
||||
class_tokens_mask = [True if class_token_index <= i < class_token_index+num_id_images else False \
|
||||
for i in range(len(clean_input_ids))]
|
||||
|
||||
clean_input_ids = torch.tensor(clean_input_ids, dtype=torch.long).unsqueeze(0)
|
||||
class_tokens_mask = torch.tensor(class_tokens_mask, dtype=torch.bool).unsqueeze(0)
|
||||
|
||||
prompt_embeds = text_encoder(
|
||||
clean_input_ids.to(device),
|
||||
output_hidden_states=True,
|
||||
)
|
||||
|
||||
# We are only ALWAYS interested in the pooled output of the final text encoder
|
||||
pooled_prompt_embeds = prompt_embeds[0]
|
||||
prompt_embeds = prompt_embeds.hidden_states[-2]
|
||||
prompt_embeds_list.append(prompt_embeds)
|
||||
|
||||
prompt_embeds = torch.concat(prompt_embeds_list, dim=-1)
|
||||
|
||||
prompt_embeds = prompt_embeds.to(dtype=self.text_encoder_2.dtype, device=device)
|
||||
class_tokens_mask = class_tokens_mask.to(device=device) # TODO: ignoring two-prompt case
|
||||
|
||||
return prompt_embeds, pooled_prompt_embeds, class_tokens_mask
|
||||
|
||||
|
||||
@torch.no_grad()
|
||||
def __call__(
|
||||
self,
|
||||
prompt: Union[str, List[str]] = None,
|
||||
prompt_2: Optional[Union[str, List[str]]] = None,
|
||||
height: Optional[int] = None,
|
||||
width: Optional[int] = None,
|
||||
num_inference_steps: int = 50,
|
||||
denoising_end: Optional[float] = None,
|
||||
guidance_scale: float = 5.0,
|
||||
negative_prompt: Optional[Union[str, List[str]]] = None,
|
||||
negative_prompt_2: Optional[Union[str, List[str]]] = None,
|
||||
num_images_per_prompt: Optional[int] = 1,
|
||||
eta: float = 0.0,
|
||||
generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None,
|
||||
latents: Optional[torch.FloatTensor] = None,
|
||||
prompt_embeds: Optional[torch.FloatTensor] = None,
|
||||
negative_prompt_embeds: Optional[torch.FloatTensor] = None,
|
||||
pooled_prompt_embeds: Optional[torch.FloatTensor] = None,
|
||||
negative_pooled_prompt_embeds: Optional[torch.FloatTensor] = None,
|
||||
output_type: Optional[str] = "pil",
|
||||
return_dict: bool = True,
|
||||
cross_attention_kwargs: Optional[Dict[str, Any]] = None,
|
||||
guidance_rescale: float = 0.0,
|
||||
original_size: Optional[Tuple[int, int]] = None,
|
||||
crops_coords_top_left: Tuple[int, int] = (0, 0),
|
||||
target_size: Optional[Tuple[int, int]] = None,
|
||||
callback: Optional[Callable[[int, int, torch.FloatTensor], None]] = None,
|
||||
callback_steps: int = 1,
|
||||
# Added parameters (for PhotoMaker)
|
||||
input_id_images: PipelineImageInput = None,
|
||||
start_merge_step: int = 0, # TODO: change to `style_strength_ratio` in the future
|
||||
class_tokens_mask: Optional[torch.LongTensor] = None,
|
||||
prompt_embeds_text_only: Optional[torch.FloatTensor] = None,
|
||||
pooled_prompt_embeds_text_only: Optional[torch.FloatTensor] = None,
|
||||
):
|
||||
# 0. Default height and width to unet
|
||||
height = height or self.unet.config.sample_size * self.vae_scale_factor
|
||||
width = width or self.unet.config.sample_size * self.vae_scale_factor
|
||||
|
||||
original_size = original_size or (height, width)
|
||||
target_size = target_size or (height, width)
|
||||
|
||||
# 1. Check inputs. Raise error if not correct
|
||||
self.check_inputs(
|
||||
prompt,
|
||||
prompt_2,
|
||||
height,
|
||||
width,
|
||||
callback_steps,
|
||||
negative_prompt,
|
||||
negative_prompt_2,
|
||||
prompt_embeds,
|
||||
negative_prompt_embeds,
|
||||
pooled_prompt_embeds,
|
||||
negative_pooled_prompt_embeds,
|
||||
)
|
||||
#
|
||||
if prompt_embeds is not None and class_tokens_mask is None:
|
||||
raise ValueError(
|
||||
"If `prompt_embeds` are provided, `class_tokens_mask` also have to be passed. Make sure to generate `class_tokens_mask` from the same tokenizer that was used to generate `prompt_embeds`."
|
||||
)
|
||||
# check the input id images
|
||||
if input_id_images is None:
|
||||
raise ValueError(
|
||||
"Provide `input_id_images`. Cannot leave `input_id_images` undefined for PhotoMaker pipeline."
|
||||
)
|
||||
if not isinstance(input_id_images, list):
|
||||
input_id_images = [input_id_images]
|
||||
|
||||
# 2. Define call parameters
|
||||
if prompt is not None and isinstance(prompt, str):
|
||||
batch_size = 1
|
||||
elif prompt is not None and isinstance(prompt, list):
|
||||
batch_size = len(prompt)
|
||||
else:
|
||||
batch_size = prompt_embeds.shape[0]
|
||||
|
||||
device = self._execution_device
|
||||
|
||||
# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
|
||||
# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
|
||||
# corresponds to doing no classifier free guidance.
|
||||
do_classifier_free_guidance = guidance_scale > 1.0
|
||||
|
||||
assert do_classifier_free_guidance
|
||||
|
||||
# 3. Encode input prompt
|
||||
num_id_images = len(input_id_images)
|
||||
|
||||
(
|
||||
prompt_embeds,
|
||||
pooled_prompt_embeds,
|
||||
class_tokens_mask,
|
||||
) = self.encode_prompt_with_trigger_word(
|
||||
prompt=prompt,
|
||||
prompt_2=prompt_2,
|
||||
device=device,
|
||||
num_id_images=num_id_images,
|
||||
prompt_embeds=prompt_embeds,
|
||||
pooled_prompt_embeds=pooled_prompt_embeds,
|
||||
class_tokens_mask=class_tokens_mask,
|
||||
)
|
||||
|
||||
# 4. Encode input prompt without the trigger word for delayed conditioning
|
||||
prompt_text_only = prompt.replace(" "+self.trigger_word, "") # sensitive to white space
|
||||
(
|
||||
prompt_embeds_text_only,
|
||||
negative_prompt_embeds,
|
||||
pooled_prompt_embeds_text_only, # TODO: replace the pooled_prompt_embeds with text only prompt
|
||||
negative_pooled_prompt_embeds,
|
||||
) = self.encode_prompt(
|
||||
prompt=prompt_text_only,
|
||||
prompt_2=prompt_2,
|
||||
device=device,
|
||||
num_images_per_prompt=num_images_per_prompt,
|
||||
do_classifier_free_guidance=do_classifier_free_guidance,
|
||||
negative_prompt=negative_prompt,
|
||||
negative_prompt_2=negative_prompt_2,
|
||||
prompt_embeds=prompt_embeds_text_only,
|
||||
negative_prompt_embeds=negative_prompt_embeds,
|
||||
pooled_prompt_embeds=pooled_prompt_embeds_text_only,
|
||||
negative_pooled_prompt_embeds=negative_pooled_prompt_embeds,
|
||||
)
|
||||
|
||||
# 5. Prepare the input ID images
|
||||
dtype = next(self.id_encoder.parameters()).dtype
|
||||
if not isinstance(input_id_images[0], torch.Tensor):
|
||||
id_pixel_values = self.id_image_processor(input_id_images, return_tensors="pt").pixel_values
|
||||
|
||||
id_pixel_values = id_pixel_values.unsqueeze(0).to(device=device, dtype=dtype) # TODO: multiple prompts
|
||||
|
||||
# 6. Get the update text embedding with the stacked ID embedding
|
||||
prompt_embeds = self.id_encoder(id_pixel_values, prompt_embeds, class_tokens_mask)
|
||||
|
||||
bs_embed, seq_len, _ = prompt_embeds.shape
|
||||
# duplicate text embeddings for each generation per prompt, using mps friendly method
|
||||
prompt_embeds = prompt_embeds.repeat(1, num_images_per_prompt, 1)
|
||||
prompt_embeds = prompt_embeds.view(bs_embed * num_images_per_prompt, seq_len, -1)
|
||||
pooled_prompt_embeds = pooled_prompt_embeds.repeat(1, num_images_per_prompt).view(
|
||||
bs_embed * num_images_per_prompt, -1
|
||||
)
|
||||
|
||||
# 7. Prepare timesteps
|
||||
self.scheduler.set_timesteps(num_inference_steps, device=device)
|
||||
timesteps = self.scheduler.timesteps
|
||||
|
||||
# 8. Prepare latent variables
|
||||
num_channels_latents = self.unet.config.in_channels
|
||||
latents = self.prepare_latents(
|
||||
batch_size * num_images_per_prompt,
|
||||
num_channels_latents,
|
||||
height,
|
||||
width,
|
||||
prompt_embeds.dtype,
|
||||
device,
|
||||
generator,
|
||||
latents,
|
||||
)
|
||||
|
||||
# 9. Prepare extra step kwargs.
|
||||
extra_step_kwargs = self.prepare_extra_step_kwargs(generator, eta)
|
||||
|
||||
# 10. Prepare added time ids & embeddings
|
||||
if self.text_encoder_2 is None:
|
||||
text_encoder_projection_dim = int(pooled_prompt_embeds.shape[-1])
|
||||
else:
|
||||
text_encoder_projection_dim = self.text_encoder_2.config.projection_dim
|
||||
|
||||
add_time_ids = self._get_add_time_ids(
|
||||
original_size,
|
||||
crops_coords_top_left,
|
||||
target_size,
|
||||
dtype=prompt_embeds.dtype,
|
||||
text_encoder_projection_dim=text_encoder_projection_dim,
|
||||
)
|
||||
add_time_ids = torch.cat([add_time_ids, add_time_ids], dim=0)
|
||||
add_time_ids = add_time_ids.to(device).repeat(batch_size * num_images_per_prompt, 1)
|
||||
|
||||
# 11. Denoising loop
|
||||
num_warmup_steps = len(timesteps) - num_inference_steps * self.scheduler.order
|
||||
with self.progress_bar(total=num_inference_steps) as progress_bar:
|
||||
for i, t in enumerate(timesteps):
|
||||
latent_model_input = (
|
||||
torch.cat([latents] * 2) if do_classifier_free_guidance else latents
|
||||
)
|
||||
latent_model_input = self.scheduler.scale_model_input(latent_model_input, t)
|
||||
|
||||
if i <= start_merge_step:
|
||||
current_prompt_embeds = torch.cat(
|
||||
[negative_prompt_embeds, prompt_embeds_text_only], dim=0
|
||||
)
|
||||
add_text_embeds = torch.cat([negative_pooled_prompt_embeds, pooled_prompt_embeds_text_only], dim=0)
|
||||
else:
|
||||
current_prompt_embeds = torch.cat(
|
||||
[negative_prompt_embeds, prompt_embeds], dim=0
|
||||
)
|
||||
add_text_embeds = torch.cat([negative_pooled_prompt_embeds, pooled_prompt_embeds], dim=0)
|
||||
# predict the noise residual
|
||||
added_cond_kwargs = {"text_embeds": add_text_embeds, "time_ids": add_time_ids}
|
||||
noise_pred = self.unet(
|
||||
latent_model_input,
|
||||
t,
|
||||
encoder_hidden_states=current_prompt_embeds,
|
||||
cross_attention_kwargs=cross_attention_kwargs,
|
||||
added_cond_kwargs=added_cond_kwargs,
|
||||
return_dict=False,
|
||||
)[0]
|
||||
|
||||
# perform guidance
|
||||
if do_classifier_free_guidance:
|
||||
noise_pred_uncond, noise_pred_text = noise_pred.chunk(2)
|
||||
noise_pred = noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
|
||||
|
||||
if do_classifier_free_guidance and guidance_rescale > 0.0:
|
||||
# Based on 3.4. in https://arxiv.org/pdf/2305.08891.pdf
|
||||
noise_pred = rescale_noise_cfg(noise_pred, noise_pred_text, guidance_rescale=guidance_rescale)
|
||||
|
||||
# compute the previous noisy sample x_t -> x_t-1
|
||||
latents = self.scheduler.step(noise_pred, t, latents, **extra_step_kwargs, return_dict=False)[0]
|
||||
|
||||
# call the callback, if provided
|
||||
if i == len(timesteps) - 1 or ((i + 1) > num_warmup_steps and (i + 1) % self.scheduler.order == 0):
|
||||
progress_bar.update()
|
||||
if callback is not None and i % callback_steps == 0:
|
||||
callback(i, t, latents)
|
||||
|
||||
# make sure the VAE is in float32 mode, as it overflows in float16
|
||||
if self.vae.dtype == torch.float16 and self.vae.config.force_upcast:
|
||||
self.upcast_vae()
|
||||
latents = latents.to(next(iter(self.vae.post_quant_conv.parameters())).dtype)
|
||||
|
||||
if not output_type == "latent":
|
||||
image = self.vae.decode(latents / self.vae.config.scaling_factor, return_dict=False)[0]
|
||||
else:
|
||||
image = latents
|
||||
return StableDiffusionXLPipelineOutput(images=image)
|
||||
|
||||
# apply watermark if available
|
||||
# if self.watermark is not None:
|
||||
# image = self.watermark.apply_watermark(image)
|
||||
|
||||
image = self.image_processor.postprocess(image, output_type=output_type)
|
||||
|
||||
# Offload last model to CPU
|
||||
if hasattr(self, "final_offload_hook") and self.final_offload_hook is not None:
|
||||
self.final_offload_hook.offload()
|
||||
|
||||
if not return_dict:
|
||||
return (image,)
|
||||
|
||||
return StableDiffusionXLPipelineOutput(images=image)
|
||||
Reference in New Issue
Block a user