diff --git a/CHANGELOG.md b/CHANGELOG.md
index 574faee76..2fc24a459 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -21,14 +21,15 @@ OPTIONAL:
- masking api
- preprocess api
-## Update for 2023-01-25
+## Update for 2023-01-27
Another big release, highlights being:
- A lot more functionality in the **Control** module:
- Inpaint and outpaint support, flexible resizing options, optional hires
- Built-in support for many new processors and models which are auto-downloaded on first use
- Full support for scripts and extensions
-- Fully baked-in **FaceID**, **FaceSwap** and **PhotoMaker** modules
+- Complete **Face** module
+ implements all variations of **FaceID**, **FaceSwap** and latest **PhotoMaker** and **InstantID**
- Much enhanced **IPAdapter** modules
- Brand new **intelligent masking**, manual or automatic
Using ML models (object removal, background removal, segmentation, etc.) and with live previews
@@ -92,27 +93,34 @@ As of this release, default backend is set to **diffusers** as its more feature
- fix batch/folder/video modes
- fix processor switching within same unit
- fix pipeline switching between different modes
-- [FaceID/FaceSwap](https://huggingface.co/h94/IP-Adapter-FaceID)
- - full implementation for *SD15* and *SD-XL*, to use simply select from *Scripts*
- **Base** (93MB) uses *InsightFace* to generate face embeds and *OpenCLIP-ViT-H-14* (2.5GB) as image encoder
- **SXDL** (1022MB) uses *InsightFace* to generate face embeds and *OpenCLIP-ViT-bigG-14* (3.7GB) as image encoder
- **Plus** (150MB) uses *InsightFace* to generate face embeds and *CLIP-ViT-H-14-laion2B* (3.8GB) as image encoder
- - **FaceSwap**
- you can use just faceid or just faceswap or both at the same time
- faceid guides image generation given the input image while face swap performs face swapping at the end of generation
- - *note*: all models are downloaded on first use
- - enable use via api, thanks @trojaner
+- **Face** module
+ implements all variations of **FaceID**, **FaceSwap** and latest **PhotoMaker** and **InstantID**
+ simply select from scripts and choose your favorite method and model
+ *note*: all models are auto-downloaded on first use
+ - [FaceID](https://huggingface.co/h94/IP-Adapter-FaceID)
+ - faceid guides image generation given the input image
+ - full implementation for *SD15* and *SD-XL*, to use simply select from *Scripts*
+ **Base** (93MB) uses *InsightFace* to generate face embeds and *OpenCLIP-ViT-H-14* (2.5GB) as image encoder
+ **Plus** (150MB) uses *InsightFace* to generate face embeds and *CLIP-ViT-H-14-laion2B* (3.8GB) as image encoder
+ **SXDL** (1022MB) uses *InsightFace* to generate face embeds and *OpenCLIP-ViT-bigG-14* (3.7GB) as image encoder
+ - [FaceSwap](https://github.com/deepinsight/insightface/blob/master/examples/in_swapper/README.md)
+ - face swap performs face swapping at the end of generation
+ - based on InsightFace in-swapper
+ - [PhotoMaker](https://github.com/TencentARC/PhotoMaker)
+ - for *SD-XL* only
+ - new model from TenencentARC using similar concept as IPAdapter, but with different implementation and
+ allowing full concept swaps between input images and generated images using trigger words
+ - note: trigger word must match exactly one term in prompt for model to work
+ - [InstantID](https://github.com/InstantID/InstantID)
+ - for *SD-XL* only
+ - based on custom trained ip-adapter and controlnet combined concepts
+ - note: controlnet appears to be heavily watermarked
+ - enable use via api, thanks @trojaner
- [IPAdapter](https://huggingface.co/h94/IP-Adapter)
- additional models for *SD15* and *SD-XL*, to use simply select from *Scripts*:
**SD15**: Base, Base ViT-G, Light, Plus, Plus Face, Full Face
**SDXL**: Base SXDL, Base ViT-H SXDL, Plus ViT-H SXDL, Plus Face ViT-H SXDL
- enable use via api, thanks @trojaner
-- [PhotoMaker](https://github.com/TencentARC/PhotoMaker)
- - for *SD-XL* only
- - simply select from *scripts*
- - new model from TenencentARC using similar concept as IPAdapter, but with different implementation and
- allowing full concept swaps between input images and generated images using trigger words
- - note: trigger word must match exactly one term in prompt for model to work
- [Self-attention guidance](https://github.com/SusungHong/Self-Attention-Guidance)
- simply select scale in advanced menu
- can drastically improve image coherence as well as reduce artifacts
@@ -250,6 +258,7 @@ As of this release, default backend is set to **diffusers** as its more feature
- cli: fix cmd args parsing
- global crlf->lf switch
- model type switch if there is loaded submodels
+ - cleanup samplers use of compute devices, thanks @Disty0
- **other**
- updated core requirements
- major internal ui module refactoring
diff --git a/extensions-builtin/sd-webui-controlnet b/extensions-builtin/sd-webui-controlnet
index 82450fbf2..e081a3a0e 160000
--- a/extensions-builtin/sd-webui-controlnet
+++ b/extensions-builtin/sd-webui-controlnet
@@ -1 +1 @@
-Subproject commit 82450fbf2d92cc131d35a99842f1a7791e4d1101
+Subproject commit e081a3a0e25899777e40e9b19b3368488fadc57a
diff --git a/modules/control/proc/reference_sd15.py b/modules/control/proc/reference_sd15.py
index 53f8602d3..fc16d6f56 100644
--- a/modules/control/proc/reference_sd15.py
+++ b/modules/control/proc/reference_sd15.py
@@ -212,7 +212,7 @@ class StableDiffusionReferencePipeline(StableDiffusionPipeline):
num_images_per_prompt (`int`, *optional*, defaults to 1):
The number of images to generate per prompt.
eta (`float`, *optional*, defaults to 0.0):
- Corresponds to parameter eta (η) in the DDIM paper: https://arxiv.org/abs/2010.02502. Only applies to
+ Corresponds to parameter eta in the DDIM paper: https://arxiv.org/abs/2010.02502. Only applies to
[`schedulers.DDIMScheduler`], will be ignored for others.
generator (`torch.Generator` or `List[torch.Generator]`, *optional*):
One or a list of [torch generator(s)](https://pytorch.org/docs/stable/generated/torch.Generator.html)
@@ -246,7 +246,7 @@ class StableDiffusionReferencePipeline(StableDiffusionPipeline):
[diffusers.models.attention_processor](https://github.com/huggingface/diffusers/blob/main/src/diffusers/models/attention_processor.py).
guidance_rescale (`float`, *optional*, defaults to 0.0):
Guidance rescale factor proposed by [Common Diffusion Noise Schedules and Sample Steps are
- Flawed](https://arxiv.org/pdf/2305.08891.pdf) `guidance_scale` is defined as `φ` in equation 16. of
+ Flawed](https://arxiv.org/pdf/2305.08891.pdf) `guidance_scale` is defined as . in equation 16. of
[Common Diffusion Noise Schedules and Sample Steps are Flawed](https://arxiv.org/pdf/2305.08891.pdf).
Guidance rescale factor should fix overexposure when using zero terminal SNR.
attention_auto_machine_weight (`float`):
diff --git a/modules/face/__init__.py b/modules/face/__init__.py
new file mode 100644
index 000000000..f7d35b6e0
--- /dev/null
+++ b/modules/face/__init__.py
@@ -0,0 +1,140 @@
+import os
+import gradio as gr
+from PIL import Image
+from modules import scripts, processing, shared, images
+
+
+debug = shared.log.trace if os.environ.get('SD_FACE_DEBUG', None) is not None else lambda *args, **kwargs: None
+
+
+class Script(scripts.Script):
+ def title(self):
+ return 'Face'
+
+ 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 mode_change(self, mode):
+ return [
+ gr.update(visible=mode=='FaceID'),
+ gr.update(visible=mode=='FaceSwap'),
+ gr.update(visible=mode=='InstantID'),
+ gr.update(visible=mode=='PhotoMaker'),
+ ]
+
+ # return signature is array of gradio components
+ def ui(self, _is_img2img):
+ with gr.Row():
+ mode = gr.Dropdown(label='Mode', choices=['None', 'FaceID', 'FaceSwap', 'InstantID', 'PhotoMaker'], value='None')
+ with gr.Group(visible=False) as cfg_faceid:
+ with gr.Row():
+ gr.HTML(' Tencent AI Lab IP-Adapter FaceID
')
+ with gr.Row():
+ from modules.face.faceid import FACEID_MODELS
+ ip_model = gr.Dropdown(choices=list(FACEID_MODELS), label='FaceID Model', value='FaceID Base')
+ with gr.Row(visible=True):
+ ip_override = gr.Checkbox(label='Override sampler', value=True)
+ ip_cache = gr.Checkbox(label='Cache model', value=True)
+ with gr.Row(visible=True):
+ ip_strength = gr.Slider(label='Strength', minimum=0.0, maximum=2.0, step=0.01, value=1.0)
+ ip_structure = gr.Slider(label='Structure', minimum=0.0, maximum=1.0, step=0.01, value=1.0)
+ with gr.Group(visible=False) as cfg_faceswap:
+ with gr.Row():
+ gr.HTML(' InsightFace InSwapper
')
+ with gr.Row(visible=True):
+ fs_cache = gr.Checkbox(label='Cache model', value=True)
+ with gr.Group(visible=False) as cfg_instantid:
+ with gr.Row():
+ gr.HTML(' InstantX InstantID
')
+ with gr.Row():
+ id_strength = gr.Slider(label='Strength', minimum=0.0, maximum=2.0, step=0.01, value=1.0)
+ id_conditioning = gr.Slider(label='Control', minimum=0.0, maximum=2.0, step=0.01, value=0.5)
+ with gr.Row(visible=True):
+ id_cache = gr.Checkbox(label='Cache model', value=True)
+ with gr.Group(visible=False) as cfg_photomaker:
+ with gr.Row():
+ gr.HTML(' Tenecent ARC Lab PhotoMaker
')
+ with gr.Row():
+ pm_trigger = gr.Text(label='Trigger word', value="person")
+ pm_strength = gr.Slider(label='Strength', minimum=0.0, maximum=2.0, step=0.01, value=1.0)
+ pm_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=[])
+ files.change(fn=self.load_images, inputs=[files], outputs=[gallery])
+ mode.change(fn=self.mode_change, inputs=[mode], outputs=[cfg_faceid, cfg_faceswap, cfg_instantid, cfg_photomaker])
+
+ return [mode, gallery, ip_model, ip_override, ip_cache, ip_strength, ip_structure, id_strength, id_conditioning, id_cache, pm_trigger, pm_strength, pm_start, fs_cache]
+
+ def run(self, p: processing.StableDiffusionProcessing, mode, input_images, ip_model, ip_override, ip_cache, ip_strength, ip_structure, id_strength, id_conditioning, id_cache, pm_trigger, pm_strength, pm_start, fs_cache): # pylint: disable=arguments-differ, unused-argument
+ if input_images is None or len(input_images) == 0:
+ shared.log.error('Face: no init images')
+ return None
+ if shared.sd_model_type != 'sd' and shared.sd_model_type != 'sdxl':
+ shared.log.error('Face: base model not supported')
+ return None
+
+ for i, image in enumerate(input_images):
+ if isinstance(image, str):
+ from modules.api.api import decode_base64_to_image
+ input_images[i] = decode_base64_to_image(image).convert("RGB")
+
+ processed = None
+ for i, image in enumerate(input_images):
+ input_images[i] = Image.open(image['name'])
+ source_image = input_images[0]
+
+ if mode == 'FaceID': # faceid runs as ipadapter in its own pipeline
+ from modules.face.faceid import face_id
+ from modules.face.insightface import get_app
+ processed_images = face_id(p, app=get_app('buffalo_l'), source_image=source_image, model=ip_model, override=ip_override, cache=ip_cache, scale=ip_strength, structure=ip_structure) # run faceid pipeline
+ processed = processing.Processed(p, images_list=processed_images, seed=p.seed, subseed=p.subseed, index_of_first_image=0) # manually created processed object
+ elif mode == 'PhotoMaker': # photomaker creates pipeline and triggers original process_images
+ from modules.face.photomaker import photo_maker
+ processed = photo_maker(p, input_images=input_images, trigger=pm_trigger, strength=pm_strength, start=pm_start)
+ elif mode == 'InstantID':
+ from modules.face.instantid import instant_id # instantid creates pipeline and triggers original process_images
+ from modules.face.insightface import get_app
+ processed = instant_id(p, app=get_app('antelopev2'), source_image=source_image, strength=id_strength, conditioning=id_conditioning, cache=id_cache)
+
+ if processed is None: # run normal pipeline
+ processed = processing.process_images(p)
+
+ if mode == 'FaceSwap': # faceswap runs as postprocessing
+ from modules.face.faceswap import face_swap
+ from modules.face.insightface import get_app
+ 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-faceswap")
+ processed.images = face_swap(p, app=get_app('buffalo_l'), input_images=processed.images, source_image=source_image, cache=fs_cache)
+
+ processed.info = processed.infotext(p, 0)
+ processed.infotexts = [processed.info]
+ 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)
+
+ return processed
diff --git a/modules/face/faceid.py b/modules/face/faceid.py
new file mode 100644
index 000000000..d139851ab
--- /dev/null
+++ b/modules/face/faceid.py
@@ -0,0 +1,125 @@
+import os
+import cv2
+import torch
+import numpy as np
+import diffusers
+import huggingface_hub as hf
+from PIL import Image
+from modules import processing, shared, devices
+
+
+FACEID_MODELS = {
+ 'FaceID Base': 'h94/IP-Adapter-FaceID/ip-adapter-faceid_sd15.bin',
+ 'FaceID Plus v1': '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'
+}
+faceid_model = None
+faceid_model_name = None
+debug = shared.log.trace if os.environ.get('SD_FACE_DEBUG', None) is not None else lambda *args, **kwargs: None
+
+
+def face_id(p: processing.StableDiffusionProcessing, app, source_image: Image.Image, model: str, override: bool, cache: bool, scale: float, structure: float):
+ global faceid_model, faceid_model_name # pylint: disable=global-statement
+ from insightface.utils import face_align
+ from ip_adapter.ip_adapter_faceid import IPAdapterFaceID, IPAdapterFaceIDPlus, IPAdapterFaceIDXL
+
+ ip_ckpt = FACEID_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 faceid_model is None or faceid_model_name != model or not cache:
+ shared.log.debug(f'FaceID load: model={model} file={ip_ckpt}')
+ if 'Plus' in model:
+ image_encoder_path = "laion/CLIP-ViT-H-14-laion2B-s32B-b79K"
+ faceid_model = IPAdapterFaceIDPlus(
+ sd_pipe=shared.sd_model,
+ image_encoder_path=image_encoder_path,
+ ip_ckpt=model_path,
+ lora_rank=128, num_tokens=4, device=devices.device, torch_dtype=devices.dtype,
+ )
+ shortcut = 'v2' in model
+ elif 'XL' in model:
+ faceid_model = IPAdapterFaceIDXL(
+ sd_pipe=shared.sd_model,
+ ip_ckpt=model_path,
+ lora_rank=128, num_tokens=4, device=devices.device, torch_dtype=devices.dtype,
+ )
+ else:
+ faceid_model = IPAdapterFaceID(
+ sd_pipe=shared.sd_model,
+ ip_ckpt=model_path,
+ lora_rank=128, num_tokens=4, device=devices.device, torch_dtype=devices.dtype,
+ )
+ faceid_model_name = model
+ else:
+ shared.log.debug(f'FaceID cached: model={model} file={ip_ckpt}')
+
+ processed_images = []
+ np_image = cv2.cvtColor(np.array(source_image), cv2.COLOR_RGB2BGR)
+ faces = app.get(np_image)
+ if len(faces) == 0:
+ shared.log.error('FaceID: no faces found')
+ return None
+ face_embeds = torch.from_numpy(faces[0].normed_embedding).unsqueeze(0)
+ face_image = face_align.norm_crop(np_image, landmark=faces[0].kps, image_size=224) # you can also segment the face
+
+ 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'
+ ip_model_dict = { # main generate 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
+ faceid_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 = faceid_model.generate(**ip_model_dict)
+ if isinstance(res, list):
+ processed_images += res
+ faceid_model.set_scale(0)
+
+ if not cache:
+ faceid_model = None
+ faceid_model_name = None
+ devices.torch_gc()
+
+ p.extra_generation_params["IP Adapter"] = f'{basename}:{scale}'
+ return processed_images
diff --git a/modules/face/faceswap.py b/modules/face/faceswap.py
new file mode 100644
index 000000000..2d03a5232
--- /dev/null
+++ b/modules/face/faceswap.py
@@ -0,0 +1,41 @@
+from typing import List
+import os
+import cv2
+import numpy as np
+import huggingface_hub as hf
+from PIL import Image
+from modules import processing, shared, devices
+
+
+debug = shared.log.trace if os.environ.get('SD_FACE_DEBUG', None) is not None else lambda *args, **kwargs: None
+insightface_app = None
+swapper = None
+
+
+def face_swap(p: processing.StableDiffusionProcessing, app, input_images: List[Image.Image], source_image: Image.Image, cache: bool):
+ 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.INSwapper = insightface.model_zoo.model_zoo.ModelRouter(model_path)
+ swapper = router.get_model()
+
+ np_image = cv2.cvtColor(np.array(source_image), cv2.COLOR_RGB2BGR)
+ faces = app.get(np_image)
+ source_face = faces[0]
+ processed_images = []
+ for image in input_images:
+ np_image = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
+ faces = app.get(np_image)
+ for i, face in enumerate(faces):
+ debug(f'FaceSwap: face={i} source={source_face.bbox} target={face.bbox}')
+ np_image = swapper.get(img=np_image, target_face=face, source_face=source_face, paste_back=True) # pylint: disable=unexpected-keyword-arg, no-value-for-parameter
+ p.extra_generation_params["FaceSwap"] = f'{len(faces)}'
+ np_image = cv2.cvtColor(np_image, cv2.COLOR_BGR2RGB)
+ processed_images.append(Image.fromarray(np_image))
+
+ if not cache:
+ swapper = None
+ devices.torch_gc()
+
+ return processed_images
diff --git a/modules/face/insightface.py b/modules/face/insightface.py
new file mode 100644
index 000000000..20d6b5019
--- /dev/null
+++ b/modules/face/insightface.py
@@ -0,0 +1,49 @@
+import os
+from modules.shared import log, opts
+
+
+insightface_app = None
+instightface_mp = None
+
+
+def get_app(mp_name):
+ 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)
+ global insightface_app, instightface_mp # pylint: disable=global-statement
+ if insightface_app is None or mp_name != instightface_mp:
+ import onnxruntime
+ from insightface.app import FaceAnalysis
+ import huggingface_hub as hf
+ import zipfile
+ log.debug(f"InsightFace: mp={mp_name} device={onnxruntime.get_device()} providers={onnxruntime.get_available_providers()}")
+ root_dir = os.path.join(opts.diffusers_dir, 'models--vladmandic--insightface-faceanalysis')
+ local_dir = os.path.join(root_dir, 'models')
+ extract_dir = os.path.join(local_dir, mp_name)
+ model_path = hf.hf_hub_download(
+ repo_id='vladmandic/insightface-faceanalysis',
+ filename=f'{mp_name}.zip',
+ local_dir_use_symlinks=False,
+ cache_dir=opts.diffusers_dir,
+ local_dir=local_dir
+ )
+ if not os.path.exists(extract_dir):
+ log.debug(f"InsightFace extract: folder={extract_dir}")
+ os.makedirs(extract_dir)
+ with zipfile.ZipFile(model_path) as zf:
+ zf.extractall(local_dir)
+ kwargs = {
+ 'root': root_dir,
+ 'download': False,
+ 'download_zip': False,
+ }
+ insightface_app = FaceAnalysis(name=mp_name, providers=['CUDAExecutionProvider', 'CPUExecutionProvider'], **kwargs)
+ instightface_mp = mp_name
+ onnxruntime.set_default_logger_severity(3)
+ insightface_app.prepare(ctx_id=0, det_thresh=0.5, det_size=(640, 640))
+ return insightface_app
diff --git a/modules/face/instantid.py b/modules/face/instantid.py
new file mode 100644
index 000000000..ce8dde6c0
--- /dev/null
+++ b/modules/face/instantid.py
@@ -0,0 +1,86 @@
+import os
+import cv2
+import numpy as np
+import huggingface_hub as hf
+from modules import shared, processing, sd_models, devices
+
+
+REPO_ID = "InstantX/InstantID"
+controlnet_model = None
+debug = shared.log.trace if os.environ.get('SD_FACE_DEBUG', None) is not None else lambda *args, **kwargs: None
+
+
+def instant_id(p: processing.StableDiffusionProcessing, app, source_image, strength=1.0, conditioning=0.5, cache=True): # pylint: disable=arguments-differ
+ from modules.face.instantid_model import StableDiffusionXLInstantIDPipeline, draw_kps
+ from diffusers.models import ControlNetModel
+ global controlnet_model # pylint: disable=global-statement
+
+ # prepare pipeline
+ if source_image is None:
+ shared.log.warning('InstantID: 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'InstantID invalid base model: current={c} required=StableDiffusionXLPipeline')
+ return None
+
+ # prepare face emb
+ faces = app.get(cv2.cvtColor(np.array(source_image), cv2.COLOR_RGB2BGR))
+ face = sorted(faces, key=lambda x:(x['bbox'][2]-x['bbox'][0])*x['bbox'][3]-x['bbox'][1])[-1] # only use the maximum face
+ face_emb = face['embedding']
+ face_kps = draw_kps(source_image, face['kps'])
+
+ shared.log.debug(f'InstantID face: score={face.det_score:.2f} gender={"female" if face.gender==0 else "male"} age={face.age} bbox={face.bbox}')
+ shared.log.debug(f'InstantID loading: model={REPO_ID}')
+ face_adapter = hf.hf_hub_download(repo_id=REPO_ID, filename="ip-adapter.bin")
+ if controlnet_model is None:
+ controlnet_model = ControlNetModel.from_pretrained(REPO_ID, subfolder="ControlNetModel", torch_dtype=devices.dtype, cache_dir=shared.opts.diffusers_dir)
+
+ processing.process_init(p)
+
+ # create new pipeline
+ orig_pipeline = shared.sd_model # backup current pipeline definition
+ shared.sd_model = StableDiffusionXLInstantIDPipeline(
+ 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,
+ controlnet=controlnet_model,
+ 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.
+ shared.sd_model.load_ip_adapter_instantid(face_adapter, scale=strength)
+ shared.sd_model.set_ip_adapter_scale(strength)
+ 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, devices.dtype) # move pipeline if needed, but don't touch if its under automatic managment
+
+ # pipeline specific args
+ 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['prompt'] = p.prompt # override all logic
+ p.task_args['image_embeds'] = face_emb
+ p.task_args['image'] = face_kps
+ p.task_args['controlnet_conditioning_scale'] = float(conditioning)
+ p.task_args['ip_adapter_scale'] = float(strength)
+ debug(f'InstantID: args={p.task_args}')
+
+ # run processing
+ shared.log.debug(f'InstantID: strength={strength} conditioning={conditioning} image={source_image}')
+ processed: processing.Processed = processing.process_images(p)
+ shared.sd_model.set_ip_adapter_scale(0)
+ p.extra_generation_params['InstantID'] = f'{strength}/{conditioning}'
+ p.extra_generation_params["Face"] = f'{face.det_score:.2f} {"female" if face.gender==0 else "male"} {face.age}y'
+
+ if not cache:
+ controlnet_model = None
+ devices.torch_gc()
+
+ # restore original pipeline
+ shared.opts.data['prompt_attention'] = orig_prompt_attention
+ shared.sd_model = orig_pipeline
+ return processed
diff --git a/modules/face/instantid_model.py b/modules/face/instantid_model.py
new file mode 100644
index 000000000..5ae70b1fa
--- /dev/null
+++ b/modules/face/instantid_model.py
@@ -0,0 +1,1062 @@
+# Copyright 2024 The InstantX Team. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import math
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+
+import cv2
+import numpy as np
+import PIL.Image
+import torch
+import torch.nn as nn
+
+from diffusers import StableDiffusionXLControlNetPipeline
+from diffusers.image_processor import PipelineImageInput
+from diffusers.models import ControlNetModel
+from diffusers.pipelines.controlnet.multicontrolnet import MultiControlNetModel
+from diffusers.pipelines.stable_diffusion_xl import StableDiffusionXLPipelineOutput
+from diffusers.utils import (
+ deprecate,
+ logging,
+ replace_example_docstring,
+)
+from diffusers.utils.import_utils import is_xformers_available
+from diffusers.utils.torch_utils import is_compiled_module, is_torch_version
+
+
+try:
+ import xformers
+ import xformers.ops
+
+ xformers_available = True
+except Exception:
+ xformers_available = False
+
+logger = logging.get_logger(__name__) # pylint: disable=invalid-name
+
+
+def FeedForward(dim, mult=4):
+ inner_dim = int(dim * mult)
+ return nn.Sequential(
+ nn.LayerNorm(dim),
+ nn.Linear(dim, inner_dim, bias=False),
+ nn.GELU(),
+ nn.Linear(inner_dim, dim, bias=False),
+ )
+
+
+def reshape_tensor(x, heads):
+ bs, length, width = x.shape
+ # (bs, length, width) --> (bs, length, n_heads, dim_per_head)
+ x = x.view(bs, length, heads, -1)
+ # (bs, length, n_heads, dim_per_head) --> (bs, n_heads, length, dim_per_head)
+ x = x.transpose(1, 2)
+ # (bs, n_heads, length, dim_per_head) --> (bs*n_heads, length, dim_per_head)
+ x = x.reshape(bs, heads, length, -1)
+ return x
+
+
+class PerceiverAttention(nn.Module):
+ def __init__(self, *, dim, dim_head=64, heads=8):
+ super().__init__()
+ self.scale = dim_head**-0.5
+ self.dim_head = dim_head
+ self.heads = heads
+ inner_dim = dim_head * heads
+
+ self.norm1 = nn.LayerNorm(dim)
+ self.norm2 = nn.LayerNorm(dim)
+
+ self.to_q = nn.Linear(dim, inner_dim, bias=False)
+ self.to_kv = nn.Linear(dim, inner_dim * 2, bias=False)
+ self.to_out = nn.Linear(inner_dim, dim, bias=False)
+
+ def forward(self, x, latents):
+ """
+ Args:
+ x (torch.Tensor): image features
+ shape (b, n1, D)
+ latent (torch.Tensor): latent features
+ shape (b, n2, D)
+ """
+ x = self.norm1(x)
+ latents = self.norm2(latents)
+
+ b, l, _ = latents.shape # noqa:E741
+
+ q = self.to_q(latents)
+ kv_input = torch.cat((x, latents), dim=-2)
+ k, v = self.to_kv(kv_input).chunk(2, dim=-1)
+
+ q = reshape_tensor(q, self.heads)
+ k = reshape_tensor(k, self.heads)
+ v = reshape_tensor(v, self.heads)
+
+ # attention
+ scale = 1 / math.sqrt(math.sqrt(self.dim_head))
+ weight = (q * scale) @ (k * scale).transpose(-2, -1) # More stable with f16 than dividing afterwards
+ weight = torch.softmax(weight.float(), dim=-1).type(weight.dtype)
+ out = weight @ v
+
+ out = out.permute(0, 2, 1, 3).reshape(b, l, -1)
+
+ return self.to_out(out)
+
+
+class Resampler(nn.Module):
+ def __init__(
+ self,
+ dim=1024,
+ depth=8,
+ dim_head=64,
+ heads=16,
+ num_queries=8,
+ embedding_dim=768,
+ output_dim=1024,
+ ff_mult=4,
+ ):
+ super().__init__()
+
+ self.latents = nn.Parameter(torch.randn(1, num_queries, dim) / dim**0.5)
+
+ self.proj_in = nn.Linear(embedding_dim, dim)
+
+ self.proj_out = nn.Linear(dim, output_dim)
+ self.norm_out = nn.LayerNorm(output_dim)
+
+ self.layers = nn.ModuleList([])
+ for _ in range(depth):
+ self.layers.append(
+ nn.ModuleList(
+ [
+ PerceiverAttention(dim=dim, dim_head=dim_head, heads=heads),
+ FeedForward(dim=dim, mult=ff_mult),
+ ]
+ )
+ )
+
+ def forward(self, x):
+ latents = self.latents.repeat(x.size(0), 1, 1)
+ x = self.proj_in(x)
+
+ for attn, ff in self.layers:
+ latents = attn(x, latents) + latents
+ latents = ff(latents) + latents
+
+ latents = self.proj_out(latents)
+ return self.norm_out(latents)
+
+
+class AttnProcessor(nn.Module):
+ r"""
+ Default processor for performing attention-related computations.
+ """
+
+ def __init__(
+ self,
+ hidden_size=None,
+ cross_attention_dim=None,
+ ):
+ super().__init__()
+
+ def __call__(
+ self,
+ attn,
+ hidden_states,
+ encoder_hidden_states=None,
+ attention_mask=None,
+ temb=None,
+ ):
+ residual = hidden_states
+
+ if attn.spatial_norm is not None:
+ hidden_states = attn.spatial_norm(hidden_states, temb)
+
+ input_ndim = hidden_states.ndim
+
+ if input_ndim == 4:
+ batch_size, channel, height, width = hidden_states.shape
+ hidden_states = hidden_states.view(batch_size, channel, height * width).transpose(1, 2)
+
+ batch_size, sequence_length, _ = (
+ hidden_states.shape if encoder_hidden_states is None else encoder_hidden_states.shape
+ )
+ attention_mask = attn.prepare_attention_mask(attention_mask, sequence_length, batch_size)
+
+ if attn.group_norm is not None:
+ hidden_states = attn.group_norm(hidden_states.transpose(1, 2)).transpose(1, 2)
+
+ query = attn.to_q(hidden_states)
+
+ if encoder_hidden_states is None:
+ encoder_hidden_states = hidden_states
+ elif attn.norm_cross:
+ encoder_hidden_states = attn.norm_encoder_hidden_states(encoder_hidden_states)
+
+ key = attn.to_k(encoder_hidden_states)
+ value = attn.to_v(encoder_hidden_states)
+
+ query = attn.head_to_batch_dim(query)
+ key = attn.head_to_batch_dim(key)
+ value = attn.head_to_batch_dim(value)
+
+ attention_probs = attn.get_attention_scores(query, key, attention_mask)
+ hidden_states = torch.bmm(attention_probs, value)
+ hidden_states = attn.batch_to_head_dim(hidden_states)
+
+ # linear proj
+ hidden_states = attn.to_out[0](hidden_states)
+ # dropout
+ hidden_states = attn.to_out[1](hidden_states)
+
+ if input_ndim == 4:
+ hidden_states = hidden_states.transpose(-1, -2).reshape(batch_size, channel, height, width)
+
+ if attn.residual_connection:
+ hidden_states = hidden_states + residual
+
+ hidden_states = hidden_states / attn.rescale_output_factor
+
+ return hidden_states
+
+
+class IPAttnProcessor(nn.Module):
+ r"""
+ Attention processor for IP-Adapater.
+ Args:
+ hidden_size (`int`):
+ The hidden size of the attention layer.
+ cross_attention_dim (`int`):
+ The number of channels in the `encoder_hidden_states`.
+ scale (`float`, defaults to 1.0):
+ the weight scale of image prompt.
+ num_tokens (`int`, defaults to 4 when do ip_adapter_plus it should be 16):
+ The context length of the image features.
+ """
+
+ def __init__(self, hidden_size, cross_attention_dim=None, scale=1.0, num_tokens=4):
+ super().__init__()
+
+ self.hidden_size = hidden_size
+ self.cross_attention_dim = cross_attention_dim
+ self.scale = scale
+ self.num_tokens = num_tokens
+
+ self.to_k_ip = nn.Linear(cross_attention_dim or hidden_size, hidden_size, bias=False)
+ self.to_v_ip = nn.Linear(cross_attention_dim or hidden_size, hidden_size, bias=False)
+
+ def __call__(
+ self,
+ attn,
+ hidden_states,
+ encoder_hidden_states=None,
+ attention_mask=None,
+ temb=None,
+ ):
+ residual = hidden_states
+
+ if attn.spatial_norm is not None:
+ hidden_states = attn.spatial_norm(hidden_states, temb)
+
+ input_ndim = hidden_states.ndim
+
+ if input_ndim == 4:
+ batch_size, channel, height, width = hidden_states.shape
+ hidden_states = hidden_states.view(batch_size, channel, height * width).transpose(1, 2)
+
+ batch_size, sequence_length, _ = (
+ hidden_states.shape if encoder_hidden_states is None else encoder_hidden_states.shape
+ )
+ attention_mask = attn.prepare_attention_mask(attention_mask, sequence_length, batch_size)
+
+ if attn.group_norm is not None:
+ hidden_states = attn.group_norm(hidden_states.transpose(1, 2)).transpose(1, 2)
+
+ query = attn.to_q(hidden_states)
+
+ if encoder_hidden_states is None:
+ encoder_hidden_states = hidden_states
+ else:
+ # get encoder_hidden_states, ip_hidden_states
+ end_pos = encoder_hidden_states.shape[1] - self.num_tokens
+ encoder_hidden_states, ip_hidden_states = (
+ encoder_hidden_states[:, :end_pos, :],
+ encoder_hidden_states[:, end_pos:, :],
+ )
+ if attn.norm_cross:
+ encoder_hidden_states = attn.norm_encoder_hidden_states(encoder_hidden_states)
+
+ key = attn.to_k(encoder_hidden_states)
+ value = attn.to_v(encoder_hidden_states)
+
+ query = attn.head_to_batch_dim(query)
+ key = attn.head_to_batch_dim(key)
+ value = attn.head_to_batch_dim(value)
+
+ if xformers_available:
+ hidden_states = self._memory_efficient_attention_xformers(query, key, value, attention_mask)
+ else:
+ attention_probs = attn.get_attention_scores(query, key, attention_mask)
+ hidden_states = torch.bmm(attention_probs, value)
+ hidden_states = attn.batch_to_head_dim(hidden_states)
+
+ # for ip-adapter
+ ip_key = self.to_k_ip(ip_hidden_states)
+ ip_value = self.to_v_ip(ip_hidden_states)
+
+ ip_key = attn.head_to_batch_dim(ip_key)
+ ip_value = attn.head_to_batch_dim(ip_value)
+
+ if xformers_available:
+ ip_hidden_states = self._memory_efficient_attention_xformers(query, ip_key, ip_value, None)
+ else:
+ ip_attention_probs = attn.get_attention_scores(query, ip_key, None)
+ ip_hidden_states = torch.bmm(ip_attention_probs, ip_value)
+ ip_hidden_states = attn.batch_to_head_dim(ip_hidden_states)
+
+ hidden_states = hidden_states + self.scale * ip_hidden_states
+
+ # linear proj
+ hidden_states = attn.to_out[0](hidden_states)
+ # dropout
+ hidden_states = attn.to_out[1](hidden_states)
+
+ if input_ndim == 4:
+ hidden_states = hidden_states.transpose(-1, -2).reshape(batch_size, channel, height, width)
+
+ if attn.residual_connection:
+ hidden_states = hidden_states + residual
+
+ hidden_states = hidden_states / attn.rescale_output_factor
+
+ return hidden_states
+
+ def _memory_efficient_attention_xformers(self, query, key, value, attention_mask):
+ # TODO attention_mask
+ query = query.contiguous()
+ key = key.contiguous()
+ value = value.contiguous()
+ hidden_states = xformers.ops.memory_efficient_attention(query, key, value, attn_bias=attention_mask)
+ return hidden_states
+
+
+EXAMPLE_DOC_STRING = """
+ Examples:
+ ```py
+ >>> # !pip install opencv-python transformers accelerate insightface
+ >>> import diffusers
+ >>> from diffusers.utils import load_image
+ >>> from diffusers.models import ControlNetModel
+
+ >>> import cv2
+ >>> import torch
+ >>> import numpy as np
+ >>> from PIL import Image
+
+ >>> from insightface.app import FaceAnalysis
+ >>> from pipeline_stable_diffusion_xl_instantid import StableDiffusionXLInstantIDPipeline, draw_kps
+
+ >>> # download 'antelopev2' under ./models
+ >>> app = FaceAnalysis(name='antelopev2', root='./', providers=['CUDAExecutionProvider', 'CPUExecutionProvider'])
+ >>> app.prepare(ctx_id=0, det_size=(640, 640))
+
+ >>> # download models under ./checkpoints
+ >>> face_adapter = f'./checkpoints/ip-adapter.bin'
+ >>> controlnet_path = f'./checkpoints/ControlNetModel'
+
+ >>> # load IdentityNet
+ >>> controlnet = ControlNetModel.from_pretrained(controlnet_path, torch_dtype=torch.float16)
+
+ >>> pipe = StableDiffusionXLInstantIDPipeline.from_pretrained(
+ ... "stabilityai/stable-diffusion-xl-base-1.0", controlnet=controlnet, torch_dtype=torch.float16
+ ... )
+ >>> pipe.cuda()
+
+ >>> # load adapter
+ >>> pipe.load_ip_adapter_instantid(face_adapter)
+
+ >>> prompt = "analog film photo of a man. faded film, desaturated, 35mm photo, grainy, vignette, vintage, Kodachrome, Lomography, stained, highly detailed, found footage, masterpiece, best quality"
+ >>> negative_prompt = "(lowres, low quality, worst quality:1.2), (text:1.2), watermark, painting, drawing, illustration, glitch, deformed, mutated, cross-eyed, ugly, disfigured (lowres, low quality, worst quality:1.2), (text:1.2), watermark, painting, drawing, illustration, glitch,deformed, mutated, cross-eyed, ugly, disfigured"
+
+ >>> # load an image
+ >>> image = load_image("your-example.jpg")
+
+ >>> face_info = app.get(cv2.cvtColor(np.array(face_image), cv2.COLOR_RGB2BGR))[-1]
+ >>> face_emb = face_info['embedding']
+ >>> face_kps = draw_kps(face_image, face_info['kps'])
+
+ >>> pipe.set_ip_adapter_scale(0.8)
+
+ >>> # generate image
+ >>> image = pipe(
+ ... prompt, image_embeds=face_emb, image=face_kps, controlnet_conditioning_scale=0.8
+ ... ).images[0]
+ ```
+"""
+
+
+def draw_kps(image_pil, kps, color_list=None):
+ if color_list is None:
+ color_list = [(255, 0, 0), (0, 255, 0), (0, 0, 255), (255, 255, 0), (255, 0, 255)]
+ stickwidth = 4
+ limbSeq = np.array([[0, 2], [1, 2], [3, 2], [4, 2]])
+ kps = np.array(kps)
+
+ w, h = image_pil.size
+ out_img = np.zeros([h, w, 3])
+
+ for i in range(len(limbSeq)):
+ index = limbSeq[i]
+ color = color_list[index[0]]
+
+ x = kps[index][:, 0]
+ y = kps[index][:, 1]
+ length = ((x[0] - x[1]) ** 2 + (y[0] - y[1]) ** 2) ** 0.5
+ angle = math.degrees(math.atan2(y[0] - y[1], x[0] - x[1]))
+ polygon = cv2.ellipse2Poly(
+ (int(np.mean(x)), int(np.mean(y))), (int(length / 2), stickwidth), int(angle), 0, 360, 1
+ )
+ out_img = cv2.fillConvexPoly(out_img.copy(), polygon, color)
+ out_img = (out_img * 0.6).astype(np.uint8)
+
+ for idx_kp, kp in enumerate(kps):
+ color = color_list[idx_kp]
+ x, y = kp
+ out_img = cv2.circle(out_img.copy(), (int(x), int(y)), 10, color, -1)
+
+ out_img_pil = PIL.Image.fromarray(out_img.astype(np.uint8))
+ return out_img_pil
+
+
+class StableDiffusionXLInstantIDPipeline(StableDiffusionXLControlNetPipeline):
+ def cuda(self, dtype=torch.float16, use_xformers=False):
+ self.to("cuda", dtype)
+
+ if hasattr(self, "image_proj_model"):
+ self.image_proj_model.to(self.unet.device).to(self.unet.dtype)
+
+ if use_xformers:
+ if is_xformers_available():
+ import xformers
+ from packaging import version
+
+ xformers_version = version.parse(xformers.__version__)
+ if xformers_version == version.parse("0.0.16"):
+ logger.warn(
+ "xFormers 0.0.16 cannot be used for training in some GPUs. If you observe problems during training, please update xFormers to at least 0.0.17. See https://huggingface.co/docs/diffusers/main/en/optimization/xformers for more details."
+ )
+ self.enable_xformers_memory_efficient_attention()
+ else:
+ raise ValueError("xformers is not available. Make sure it is installed correctly")
+
+ def load_ip_adapter_instantid(self, model_ckpt, image_emb_dim=512, num_tokens=16, scale=0.5):
+ self.set_image_proj_model(model_ckpt, image_emb_dim, num_tokens)
+ self.set_ip_adapter(model_ckpt, num_tokens, scale)
+
+ def set_image_proj_model(self, model_ckpt, image_emb_dim=512, num_tokens=16):
+ image_proj_model = Resampler(
+ dim=1280,
+ depth=4,
+ dim_head=64,
+ heads=20,
+ num_queries=num_tokens,
+ embedding_dim=image_emb_dim,
+ output_dim=self.unet.config.cross_attention_dim,
+ ff_mult=4,
+ )
+
+ image_proj_model.eval()
+
+ self.image_proj_model = image_proj_model.to(self.device, dtype=self.dtype)
+ state_dict = torch.load(model_ckpt, map_location="cpu")
+ if "image_proj" in state_dict:
+ state_dict = state_dict["image_proj"]
+ self.image_proj_model.load_state_dict(state_dict)
+
+ self.image_proj_model_in_features = image_emb_dim
+
+ def set_ip_adapter(self, model_ckpt, num_tokens, scale):
+ unet = self.unet
+ attn_procs = {}
+ for name in unet.attn_processors.keys():
+ cross_attention_dim = None if name.endswith("attn1.processor") else unet.config.cross_attention_dim
+ if name.startswith("mid_block"):
+ hidden_size = unet.config.block_out_channels[-1]
+ elif name.startswith("up_blocks"):
+ block_id = int(name[len("up_blocks.")])
+ hidden_size = list(reversed(unet.config.block_out_channels))[block_id]
+ elif name.startswith("down_blocks"):
+ block_id = int(name[len("down_blocks.")])
+ hidden_size = unet.config.block_out_channels[block_id]
+ if cross_attention_dim is None:
+ attn_procs[name] = AttnProcessor().to(unet.device, dtype=unet.dtype)
+ else:
+ attn_procs[name] = IPAttnProcessor(
+ hidden_size=hidden_size,
+ cross_attention_dim=cross_attention_dim,
+ scale=scale,
+ num_tokens=num_tokens,
+ ).to(unet.device, dtype=unet.dtype)
+ unet.set_attn_processor(attn_procs)
+
+ state_dict = torch.load(model_ckpt, map_location="cpu")
+ ip_layers = torch.nn.ModuleList(self.unet.attn_processors.values())
+ if "ip_adapter" in state_dict:
+ state_dict = state_dict["ip_adapter"]
+ ip_layers.load_state_dict(state_dict)
+
+ def set_ip_adapter_scale(self, scale):
+ unet = getattr(self, self.unet_name) if not hasattr(self, "unet") else self.unet
+ for attn_processor in unet.attn_processors.values():
+ if isinstance(attn_processor, IPAttnProcessor):
+ attn_processor.scale = scale
+
+ def _encode_prompt_image_emb(self, prompt_image_emb, device, dtype, do_classifier_free_guidance):
+ if isinstance(prompt_image_emb, torch.Tensor):
+ prompt_image_emb = prompt_image_emb.clone().detach()
+ else:
+ prompt_image_emb = torch.tensor(prompt_image_emb)
+
+ prompt_image_emb = prompt_image_emb.to(device=device, dtype=dtype)
+ prompt_image_emb = prompt_image_emb.reshape([1, -1, self.image_proj_model_in_features])
+
+ if do_classifier_free_guidance:
+ prompt_image_emb = torch.cat([torch.zeros_like(prompt_image_emb), prompt_image_emb], dim=0)
+ else:
+ prompt_image_emb = torch.cat([prompt_image_emb], dim=0)
+
+ prompt_image_emb = self.image_proj_model(prompt_image_emb)
+ return prompt_image_emb
+
+ @torch.no_grad()
+ @replace_example_docstring(EXAMPLE_DOC_STRING)
+ def __call__(
+ self,
+ prompt: Union[str, List[str]] = None,
+ prompt_2: Optional[Union[str, List[str]]] = None,
+ image: PipelineImageInput = None,
+ height: Optional[int] = None,
+ width: Optional[int] = None,
+ num_inference_steps: int = 50,
+ 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,
+ image_embeds: Optional[torch.FloatTensor] = None,
+ output_type: Optional[str] = "pil",
+ return_dict: bool = True,
+ cross_attention_kwargs: Optional[Dict[str, Any]] = None,
+ controlnet_conditioning_scale: Union[float, List[float]] = 1.0,
+ guess_mode: bool = False,
+ control_guidance_start: Union[float, List[float]] = 0.0,
+ control_guidance_end: Union[float, List[float]] = 1.0,
+ original_size: Tuple[int, int] = None,
+ crops_coords_top_left: Tuple[int, int] = (0, 0),
+ target_size: Tuple[int, int] = None,
+ negative_original_size: Optional[Tuple[int, int]] = None,
+ negative_crops_coords_top_left: Tuple[int, int] = (0, 0),
+ negative_target_size: Optional[Tuple[int, int]] = None,
+ clip_skip: Optional[int] = None,
+ callback_on_step_end: Optional[Callable[[int, int, Dict], None]] = None,
+ callback_on_step_end_tensor_inputs: List[str] = None,
+ **kwargs,
+ ):
+ r"""
+ The call function to the pipeline for generation.
+
+ Args:
+ prompt (`str` or `List[str]`, *optional*):
+ The prompt or prompts to guide image generation. If not defined, you need to pass `prompt_embeds`.
+ prompt_2 (`str` or `List[str]`, *optional*):
+ The prompt or prompts to be sent to `tokenizer_2` and `text_encoder_2`. If not defined, `prompt` is
+ used in both text-encoders.
+ image (`torch.FloatTensor`, `PIL.Image.Image`, `np.ndarray`, `List[torch.FloatTensor]`, `List[PIL.Image.Image]`, `List[np.ndarray]`,:
+ `List[List[torch.FloatTensor]]`, `List[List[np.ndarray]]` or `List[List[PIL.Image.Image]]`):
+ The ControlNet input condition to provide guidance to the `unet` for generation. If the type is
+ specified as `torch.FloatTensor`, it is passed to ControlNet as is. `PIL.Image.Image` can also be
+ accepted as an image. The dimensions of the output image defaults to `image`'s dimensions. If height
+ and/or width are passed, `image` is resized accordingly. If multiple ControlNets are specified in
+ `init`, images must be passed as a list such that each element of the list can be correctly batched for
+ input to a single ControlNet.
+ height (`int`, *optional*, defaults to `self.unet.config.sample_size * self.vae_scale_factor`):
+ The height in pixels of the generated image. Anything below 512 pixels won't work well for
+ [stabilityai/stable-diffusion-xl-base-1.0](https://huggingface.co/stabilityai/stable-diffusion-xl-base-1.0)
+ and checkpoints that are not specifically fine-tuned on low resolutions.
+ width (`int`, *optional*, defaults to `self.unet.config.sample_size * self.vae_scale_factor`):
+ The width in pixels of the generated image. Anything below 512 pixels won't work well for
+ [stabilityai/stable-diffusion-xl-base-1.0](https://huggingface.co/stabilityai/stable-diffusion-xl-base-1.0)
+ and checkpoints that are not specifically fine-tuned on low resolutions.
+ num_inference_steps (`int`, *optional*, defaults to 50):
+ The number of denoising steps. More denoising steps usually lead to a higher quality image at the
+ expense of slower inference.
+ guidance_scale (`float`, *optional*, defaults to 5.0):
+ A higher guidance scale value encourages the model to generate images closely linked to the text
+ `prompt` at the expense of lower image quality. Guidance scale is enabled when `guidance_scale > 1`.
+ negative_prompt (`str` or `List[str]`, *optional*):
+ The prompt or prompts to guide what to not include in image generation. If not defined, you need to
+ pass `negative_prompt_embeds` instead. Ignored when not using guidance (`guidance_scale < 1`).
+ negative_prompt_2 (`str` or `List[str]`, *optional*):
+ The prompt or prompts to guide what to not include in image generation. This is sent to `tokenizer_2`
+ and `text_encoder_2`. If not defined, `negative_prompt` is used in both text-encoders.
+ num_images_per_prompt (`int`, *optional*, defaults to 1):
+ The number of images to generate per prompt.
+ eta (`float`, *optional*, defaults to 0.0):
+ Corresponds to parameter eta (η) from the [DDIM](https://arxiv.org/abs/2010.02502) paper. Only applies
+ to the [`~schedulers.DDIMScheduler`], and is ignored in other schedulers.
+ generator (`torch.Generator` or `List[torch.Generator]`, *optional*):
+ A [`torch.Generator`](https://pytorch.org/docs/stable/generated/torch.Generator.html) to make
+ generation deterministic.
+ latents (`torch.FloatTensor`, *optional*):
+ Pre-generated noisy latents sampled from a Gaussian distribution, to be used as inputs for image
+ generation. Can be used to tweak the same generation with different prompts. If not provided, a latents
+ tensor is generated by sampling using the supplied random `generator`.
+ prompt_embeds (`torch.FloatTensor`, *optional*):
+ Pre-generated text embeddings. Can be used to easily tweak text inputs (prompt weighting). If not
+ provided, text embeddings are generated from the `prompt` input argument.
+ negative_prompt_embeds (`torch.FloatTensor`, *optional*):
+ Pre-generated negative text embeddings. Can be used to easily tweak text inputs (prompt weighting). If
+ not provided, `negative_prompt_embeds` are generated from the `negative_prompt` input argument.
+ pooled_prompt_embeds (`torch.FloatTensor`, *optional*):
+ Pre-generated pooled text embeddings. Can be used to easily tweak text inputs (prompt weighting). If
+ not provided, pooled text embeddings are generated from `prompt` input argument.
+ negative_pooled_prompt_embeds (`torch.FloatTensor`, *optional*):
+ Pre-generated negative pooled text embeddings. Can be used to easily tweak text inputs (prompt
+ weighting). If not provided, pooled `negative_prompt_embeds` are generated from `negative_prompt` input
+ argument.
+ image_embeds (`torch.FloatTensor`, *optional*):
+ Pre-generated image embeddings.
+ output_type (`str`, *optional*, defaults to `"pil"`):
+ The output format of the generated image. Choose between `PIL.Image` or `np.array`.
+ return_dict (`bool`, *optional*, defaults to `True`):
+ Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a
+ plain tuple.
+ cross_attention_kwargs (`dict`, *optional*):
+ A kwargs dictionary that if specified is passed along to the [`AttentionProcessor`] as defined in
+ [`self.processor`](https://github.com/huggingface/diffusers/blob/main/src/diffusers/models/attention_processor.py).
+ controlnet_conditioning_scale (`float` or `List[float]`, *optional*, defaults to 1.0):
+ The outputs of the ControlNet are multiplied by `controlnet_conditioning_scale` before they are added
+ to the residual in the original `unet`. If multiple ControlNets are specified in `init`, you can set
+ the corresponding scale as a list.
+ guess_mode (`bool`, *optional*, defaults to `False`):
+ The ControlNet encoder tries to recognize the content of the input image even if you remove all
+ prompts. A `guidance_scale` value between 3.0 and 5.0 is recommended.
+ control_guidance_start (`float` or `List[float]`, *optional*, defaults to 0.0):
+ The percentage of total steps at which the ControlNet starts applying.
+ control_guidance_end (`float` or `List[float]`, *optional*, defaults to 1.0):
+ The percentage of total steps at which the ControlNet stops applying.
+ original_size (`Tuple[int]`, *optional*, defaults to (1024, 1024)):
+ If `original_size` is not the same as `target_size` the image will appear to be down- or upsampled.
+ `original_size` defaults to `(height, width)` if not specified. Part of SDXL's micro-conditioning as
+ explained in section 2.2 of
+ [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952).
+ crops_coords_top_left (`Tuple[int]`, *optional*, defaults to (0, 0)):
+ `crops_coords_top_left` can be used to generate an image that appears to be "cropped" from the position
+ `crops_coords_top_left` downwards. Favorable, well-centered images are usually achieved by setting
+ `crops_coords_top_left` to (0, 0). Part of SDXL's micro-conditioning as explained in section 2.2 of
+ [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952).
+ target_size (`Tuple[int]`, *optional*, defaults to (1024, 1024)):
+ For most cases, `target_size` should be set to the desired height and width of the generated image. If
+ not specified it will default to `(height, width)`. Part of SDXL's micro-conditioning as explained in
+ section 2.2 of [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952).
+ negative_original_size (`Tuple[int]`, *optional*, defaults to (1024, 1024)):
+ To negatively condition the generation process based on a specific image resolution. Part of SDXL's
+ micro-conditioning as explained in section 2.2 of
+ [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952). For more
+ information, refer to this issue thread: https://github.com/huggingface/diffusers/issues/4208.
+ negative_crops_coords_top_left (`Tuple[int]`, *optional*, defaults to (0, 0)):
+ To negatively condition the generation process based on a specific crop coordinates. Part of SDXL's
+ micro-conditioning as explained in section 2.2 of
+ [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952). For more
+ information, refer to this issue thread: https://github.com/huggingface/diffusers/issues/4208.
+ negative_target_size (`Tuple[int]`, *optional*, defaults to (1024, 1024)):
+ To negatively condition the generation process based on a target image resolution. It should be as same
+ as the `target_size` for most cases. Part of SDXL's micro-conditioning as explained in section 2.2 of
+ [https://huggingface.co/papers/2307.01952](https://huggingface.co/papers/2307.01952). For more
+ information, refer to this issue thread: https://github.com/huggingface/diffusers/issues/4208.
+ clip_skip (`int`, *optional*):
+ Number of layers to be skipped from CLIP while computing the prompt embeddings. A value of 1 means that
+ the output of the pre-final layer will be used for computing the prompt embeddings.
+ callback_on_step_end (`Callable`, *optional*):
+ A function that calls at the end of each denoising steps during the inference. The function is called
+ with the following arguments: `callback_on_step_end(self: DiffusionPipeline, step: int, timestep: int,
+ callback_kwargs: Dict)`. `callback_kwargs` will include a list of all tensors as specified by
+ `callback_on_step_end_tensor_inputs`.
+ callback_on_step_end_tensor_inputs (`List`, *optional*):
+ The list of tensor inputs for the `callback_on_step_end` function. The tensors specified in the list
+ will be passed as `callback_kwargs` argument. You will only be able to include variables listed in the
+ `._callback_tensor_inputs` attribute of your pipeine class.
+
+ Examples:
+
+ Returns:
+ [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] or `tuple`:
+ If `return_dict` is `True`, [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] is returned,
+ otherwise a `tuple` is returned containing the output images.
+ """
+
+ if callback_on_step_end_tensor_inputs is None:
+ callback_on_step_end_tensor_inputs = ["latents"]
+ callback = kwargs.pop("callback", None)
+ callback_steps = kwargs.pop("callback_steps", None)
+
+ if callback is not None:
+ deprecate(
+ "callback",
+ "1.0.0",
+ "Passing `callback` as an input argument to `__call__` is deprecated, consider using `callback_on_step_end`",
+ )
+ if callback_steps is not None:
+ deprecate(
+ "callback_steps",
+ "1.0.0",
+ "Passing `callback_steps` as an input argument to `__call__` is deprecated, consider using `callback_on_step_end`",
+ )
+
+ controlnet = self.controlnet._orig_mod if is_compiled_module(self.controlnet) else self.controlnet
+
+ # align format for control guidance
+ if not isinstance(control_guidance_start, list) and isinstance(control_guidance_end, list):
+ control_guidance_start = len(control_guidance_end) * [control_guidance_start]
+ elif not isinstance(control_guidance_end, list) and isinstance(control_guidance_start, list):
+ control_guidance_end = len(control_guidance_start) * [control_guidance_end]
+ elif not isinstance(control_guidance_start, list) and not isinstance(control_guidance_end, list):
+ mult = len(controlnet.nets) if isinstance(controlnet, MultiControlNetModel) else 1
+ control_guidance_start, control_guidance_end = (
+ mult * [control_guidance_start],
+ mult * [control_guidance_end],
+ )
+
+ # 1. Check inputs. Raise error if not correct
+ self.check_inputs(
+ prompt,
+ prompt_2,
+ image,
+ callback_steps,
+ negative_prompt,
+ negative_prompt_2,
+ prompt_embeds,
+ negative_prompt_embeds,
+ pooled_prompt_embeds,
+ negative_pooled_prompt_embeds,
+ controlnet_conditioning_scale,
+ control_guidance_start,
+ control_guidance_end,
+ callback_on_step_end_tensor_inputs,
+ )
+
+ self._guidance_scale = guidance_scale
+ self._clip_skip = clip_skip
+ self._cross_attention_kwargs = cross_attention_kwargs
+
+ # 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
+
+ if isinstance(controlnet, MultiControlNetModel) and isinstance(controlnet_conditioning_scale, float):
+ controlnet_conditioning_scale = [controlnet_conditioning_scale] * len(controlnet.nets)
+
+ global_pool_conditions = (
+ controlnet.config.global_pool_conditions
+ if isinstance(controlnet, ControlNetModel)
+ else controlnet.nets[0].config.global_pool_conditions
+ )
+ guess_mode = guess_mode or global_pool_conditions
+
+ # 3.1 Encode input prompt
+ text_encoder_lora_scale = (
+ self.cross_attention_kwargs.get("scale", None) if self.cross_attention_kwargs is not None else None
+ )
+ (
+ prompt_embeds,
+ negative_prompt_embeds,
+ pooled_prompt_embeds,
+ negative_pooled_prompt_embeds,
+ ) = self.encode_prompt(
+ prompt,
+ prompt_2,
+ device,
+ num_images_per_prompt,
+ self.do_classifier_free_guidance,
+ negative_prompt,
+ negative_prompt_2,
+ prompt_embeds=prompt_embeds,
+ negative_prompt_embeds=negative_prompt_embeds,
+ pooled_prompt_embeds=pooled_prompt_embeds,
+ negative_pooled_prompt_embeds=negative_pooled_prompt_embeds,
+ lora_scale=text_encoder_lora_scale,
+ clip_skip=self.clip_skip,
+ )
+
+ # 3.2 Encode image prompt
+ prompt_image_emb = self._encode_prompt_image_emb(
+ image_embeds, device, self.unet.dtype, self.do_classifier_free_guidance
+ )
+
+ # 4. Prepare image
+ if isinstance(controlnet, ControlNetModel):
+ image = self.prepare_image(
+ image=image,
+ width=width,
+ height=height,
+ batch_size=batch_size * num_images_per_prompt,
+ num_images_per_prompt=num_images_per_prompt,
+ device=device,
+ dtype=controlnet.dtype,
+ do_classifier_free_guidance=self.do_classifier_free_guidance,
+ guess_mode=guess_mode,
+ )
+ height, width = image.shape[-2:]
+ elif isinstance(controlnet, MultiControlNetModel):
+ images = []
+
+ for image_ in image:
+ image_ = self.prepare_image(
+ image=image_,
+ width=width,
+ height=height,
+ batch_size=batch_size * num_images_per_prompt,
+ num_images_per_prompt=num_images_per_prompt,
+ device=device,
+ dtype=controlnet.dtype,
+ do_classifier_free_guidance=self.do_classifier_free_guidance,
+ guess_mode=guess_mode,
+ )
+
+ images.append(image_)
+
+ image = images
+ height, width = image[0].shape[-2:]
+ else:
+ raise AssertionError
+
+ # 5. Prepare timesteps
+ self.scheduler.set_timesteps(num_inference_steps, device=device)
+ timesteps = self.scheduler.timesteps
+ self._num_timesteps = len(timesteps)
+
+ # 6. 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,
+ )
+
+ # 6.5 Optionally get Guidance Scale Embedding
+ timestep_cond = None
+ if self.unet.config.time_cond_proj_dim is not None:
+ guidance_scale_tensor = torch.tensor(self.guidance_scale - 1).repeat(batch_size * num_images_per_prompt)
+ timestep_cond = self.get_guidance_scale_embedding(
+ guidance_scale_tensor, embedding_dim=self.unet.config.time_cond_proj_dim
+ ).to(device=device, dtype=latents.dtype)
+
+ # 7. Prepare extra step kwargs. TODO: Logic should ideally just be moved out of the pipeline
+ extra_step_kwargs = self.prepare_extra_step_kwargs(generator, eta)
+
+ # 7.1 Create tensor stating which controlnets to keep
+ controlnet_keep = []
+ for i in range(len(timesteps)):
+ keeps = [
+ 1.0 - float(i / len(timesteps) < s or (i + 1) / len(timesteps) > e)
+ for s, e in zip(control_guidance_start, control_guidance_end)
+ ]
+ controlnet_keep.append(keeps[0] if isinstance(controlnet, ControlNetModel) else keeps)
+
+ # 7.2 Prepare added time ids & embeddings
+ if isinstance(image, list):
+ original_size = original_size or image[0].shape[-2:]
+ else:
+ original_size = original_size or image.shape[-2:]
+ target_size = target_size or (height, width)
+
+ add_text_embeds = pooled_prompt_embeds
+ 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,
+ )
+
+ if negative_original_size is not None and negative_target_size is not None:
+ negative_add_time_ids = self._get_add_time_ids(
+ negative_original_size,
+ negative_crops_coords_top_left,
+ negative_target_size,
+ dtype=prompt_embeds.dtype,
+ text_encoder_projection_dim=text_encoder_projection_dim,
+ )
+ else:
+ negative_add_time_ids = add_time_ids
+
+ if self.do_classifier_free_guidance:
+ prompt_embeds = torch.cat([negative_prompt_embeds, prompt_embeds], dim=0)
+ add_text_embeds = torch.cat([negative_pooled_prompt_embeds, add_text_embeds], dim=0)
+ add_time_ids = torch.cat([negative_add_time_ids, add_time_ids], dim=0)
+
+ prompt_embeds = prompt_embeds.to(device)
+ add_text_embeds = add_text_embeds.to(device)
+ add_time_ids = add_time_ids.to(device).repeat(batch_size * num_images_per_prompt, 1)
+ encoder_hidden_states = torch.cat([prompt_embeds, prompt_image_emb], dim=1)
+
+ # 8. Denoising loop
+ num_warmup_steps = len(timesteps) - num_inference_steps * self.scheduler.order
+ is_unet_compiled = is_compiled_module(self.unet)
+ is_controlnet_compiled = is_compiled_module(self.controlnet)
+ is_torch_higher_equal_2_1 = is_torch_version(">=", "2.1")
+
+ with self.progress_bar(total=num_inference_steps) as progress_bar:
+ for i, t in enumerate(timesteps):
+ # Relevant thread:
+ # https://dev-discuss.pytorch.org/t/cudagraphs-in-pytorch-2-0/1428
+ if (is_unet_compiled and is_controlnet_compiled) and is_torch_higher_equal_2_1:
+ torch._inductor.cudagraph_mark_step_begin()
+ # expand the latents if we are doing classifier free guidance
+ latent_model_input = torch.cat([latents] * 2) if self.do_classifier_free_guidance else latents
+ latent_model_input = self.scheduler.scale_model_input(latent_model_input, t)
+
+ added_cond_kwargs = {"text_embeds": add_text_embeds, "time_ids": add_time_ids}
+
+ # controlnet(s) inference
+ if guess_mode and self.do_classifier_free_guidance:
+ # Infer ControlNet only for the conditional batch.
+ control_model_input = latents
+ control_model_input = self.scheduler.scale_model_input(control_model_input, t)
+ controlnet_prompt_embeds = prompt_embeds.chunk(2)[1]
+ controlnet_added_cond_kwargs = {
+ "text_embeds": add_text_embeds.chunk(2)[1],
+ "time_ids": add_time_ids.chunk(2)[1],
+ }
+ else:
+ control_model_input = latent_model_input
+ controlnet_prompt_embeds = prompt_embeds
+ controlnet_added_cond_kwargs = added_cond_kwargs
+
+ if isinstance(controlnet_keep[i], list):
+ cond_scale = [c * s for c, s in zip(controlnet_conditioning_scale, controlnet_keep[i])]
+ else:
+ controlnet_cond_scale = controlnet_conditioning_scale
+ if isinstance(controlnet_cond_scale, list):
+ controlnet_cond_scale = controlnet_cond_scale[0]
+ cond_scale = controlnet_cond_scale * controlnet_keep[i]
+
+ down_block_res_samples, mid_block_res_sample = self.controlnet(
+ control_model_input,
+ t,
+ encoder_hidden_states=prompt_image_emb,
+ controlnet_cond=image,
+ conditioning_scale=cond_scale,
+ guess_mode=guess_mode,
+ added_cond_kwargs=controlnet_added_cond_kwargs,
+ return_dict=False,
+ )
+
+ if guess_mode and self.do_classifier_free_guidance:
+ # Infered ControlNet only for the conditional batch.
+ # To apply the output of ControlNet to both the unconditional and conditional batches,
+ # add 0 to the unconditional batch to keep it unchanged.
+ down_block_res_samples = [torch.cat([torch.zeros_like(d), d]) for d in down_block_res_samples]
+ mid_block_res_sample = torch.cat([torch.zeros_like(mid_block_res_sample), mid_block_res_sample])
+
+ # predict the noise residual
+ noise_pred = self.unet(
+ latent_model_input,
+ t,
+ encoder_hidden_states=encoder_hidden_states,
+ timestep_cond=timestep_cond,
+ cross_attention_kwargs=self.cross_attention_kwargs,
+ down_block_additional_residuals=down_block_res_samples,
+ mid_block_additional_residual=mid_block_res_sample,
+ added_cond_kwargs=added_cond_kwargs,
+ return_dict=False,
+ )[0]
+
+ # perform guidance
+ if self.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)
+
+ # 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]
+
+ if callback_on_step_end is not None:
+ callback_kwargs = {}
+ for k in callback_on_step_end_tensor_inputs:
+ callback_kwargs[k] = locals()[k]
+ callback_outputs = callback_on_step_end(self, i, t, callback_kwargs)
+
+ latents = callback_outputs.pop("latents", latents)
+ prompt_embeds = callback_outputs.pop("prompt_embeds", prompt_embeds)
+ negative_prompt_embeds = callback_outputs.pop("negative_prompt_embeds", negative_prompt_embeds)
+
+ # 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:
+ step_idx = i // getattr(self.scheduler, "order", 1)
+ callback(step_idx, t, latents)
+
+ if not output_type == "latent":
+ # make sure the VAE is in float32 mode, as it overflows in float16
+ needs_upcasting = self.vae.dtype == torch.float16 and self.vae.config.force_upcast
+ if needs_upcasting:
+ self.upcast_vae()
+ latents = latents.to(next(iter(self.vae.post_quant_conv.parameters())).dtype)
+
+ image = self.vae.decode(latents / self.vae.config.scaling_factor, return_dict=False)[0]
+
+ # cast back to fp16 if needed
+ if needs_upcasting:
+ self.vae.to(dtype=torch.float16)
+ else:
+ image = latents
+
+ if not output_type == "latent":
+ # 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 all models
+ self.maybe_free_model_hooks()
+
+ if not return_dict:
+ return (image,)
+
+ return StableDiffusionXLPipelineOutput(images=image)
diff --git a/modules/face/photomaker.py b/modules/face/photomaker.py
new file mode 100644
index 000000000..a12322b4c
--- /dev/null
+++ b/modules/face/photomaker.py
@@ -0,0 +1,73 @@
+import os
+import huggingface_hub as hf
+from modules import shared, processing, sd_models
+
+
+def photo_maker(p: processing.StableDiffusionProcessing, input_images, trigger, strength, start): # pylint: disable=arguments-differ
+ from modules.face.photomaker_model import PhotoMakerStableDiffusionXLPipeline
+
+ # prepare pipeline
+ 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
diff --git a/scripts/photomaker_model.py b/modules/face/photomaker_model.py
similarity index 99%
rename from scripts/photomaker_model.py
rename to modules/face/photomaker_model.py
index 25bbbe353..b62fe73b8 100755
--- a/scripts/photomaker_model.py
+++ b/modules/face/photomaker_model.py
@@ -533,7 +533,7 @@ class PhotoMakerStableDiffusionXLPipeline(StableDiffusionXLPipeline):
self.upcast_vae()
latents = latents.to(next(iter(self.vae.post_quant_conv.parameters())).dtype)
- if not output_type == "latent":
+ if output_type != "latent":
image = self.vae.decode(latents / self.vae.config.scaling_factor, return_dict=False)[0]
else:
image = latents
diff --git a/modules/lora b/modules/lora
index d5ab97b69..cd19df49c 160000
--- a/modules/lora
+++ b/modules/lora
@@ -1 +1 @@
-Subproject commit d5ab97b69b4822d0ef3ae7ca5cf8d37a92384305
+Subproject commit cd19df49cd512e13ac90db115c424d19c0e8868a
diff --git a/modules/processing_diffusers.py b/modules/processing_diffusers.py
index c735fb2bb..1a323131f 100644
--- a/modules/processing_diffusers.py
+++ b/modules/processing_diffusers.py
@@ -297,6 +297,8 @@ def process_diffusers(p: processing.StableDiffusionProcessing):
clean['negative_prompt_embeds'] = clean['negative_prompt_embeds'].shape if torch.is_tensor(clean['negative_prompt_embeds']) else type(clean['negative_prompt_embeds'])
if 'negative_pooled_prompt_embeds' in clean:
clean['negative_pooled_prompt_embeds'] = clean['negative_pooled_prompt_embeds'].shape if torch.is_tensor(clean['negative_pooled_prompt_embeds']) else type(clean['negative_pooled_prompt_embeds'])
+ if 'image_embeds' in clean:
+ clean['image_embeds'] = clean['image_embeds'].shape if torch.is_tensor(clean['image_embeds']) else type(clean['image_embeds'])
clean['generator'] = generator_device
clean['parser'] = parser
shared.log.debug(f'Diffuser pipeline: {model.__class__.__name__} task={sd_models.get_diffusers_task(model)} set={clean}')
diff --git a/modules/scripts.py b/modules/scripts.py
index 0f308a13f..498fc5db6 100644
--- a/modules/scripts.py
+++ b/modules/scripts.py
@@ -247,7 +247,7 @@ def load_scripts():
scripts_data.clear()
postprocessing_scripts_data.clear()
script_callbacks.clear_callbacks()
- scripts_list = list_scripts("scripts", ".py")
+ scripts_list = list_scripts('scripts', '.py') + list_scripts(os.path.join('modules', 'face'), '.py')
syspath = sys.path
def register_scripts_from_module(module, scriptfile):
diff --git a/modules/ui_control.py b/modules/ui_control.py
index 326a6f063..b1f4b5fb2 100644
--- a/modules/ui_control.py
+++ b/modules/ui_control.py
@@ -10,7 +10,6 @@ from modules.control.units import xs # vislearn ControlNet-XS
from modules.control.units import lite # vislearn ControlNet-XS
from modules.control.units import t2iadapter # TencentARC T2I-Adapter
from modules.control.units import reference # reference pipeline
-from scripts import ipadapter # pylint: disable=no-name-in-module
from modules import errors, shared, progress, sd_samplers, ui_components, ui_symbols, ui_common, ui_sections, generation_parameters_copypaste, call_queue, scripts, masking # pylint: disable=ungrouped-imports
@@ -438,7 +437,8 @@ def create_ui(_blocks: gr.Blocks=None):
with gr.Row():
with gr.Column():
gr.HTML('IP-Adapter')
- ip_adapter = gr.Dropdown(label='Adapter', choices=ipadapter.ADAPTERS, value='none')
+ from scripts.ipadapter import ADAPTERS # pylint: disable=no-name-in-module
+ ip_adapter = gr.Dropdown(label='Adapter', choices=ADAPTERS, value='none')
ip_scale = gr.Slider(label='Scale', minimum=0.0, maximum=1.0, step=0.01, value=0.5)
with gr.Column():
ip_image = gr.Image(label="Input", show_label=False, type="pil", source="upload", interactive=True, tool="editor", height=256, width=256)
diff --git a/modules/ui_extensions.py b/modules/ui_extensions.py
index 30fff153c..2cd626b0b 100644
--- a/modules/ui_extensions.py
+++ b/modules/ui_extensions.py
@@ -6,7 +6,7 @@ import html
from datetime import datetime, timedelta
import git
import gradio as gr
-from modules import extensions, shared, paths, errors
+from modules import extensions, shared, paths, errors, ui_symbols
extensions_index = "https://vladmandic.github.io/sd-data/pages/extensions.json"
@@ -372,27 +372,27 @@ def create_html(search_text, sort_column):
if ext.get('status', None) is None or type(ext['status']) == str: # old format
ext['status'] = 0
if ext['url'] is None or ext['url'] == '':
- status = "⬤"
+ status = f"{ui_symbols.bullet}"
elif ext['status'] > 0:
if ext['status'] == 1:
- status = "⬤"
+ status = f"{ui_symbols.bullet}"
elif ext['status'] == 2:
- status = "⬤"
+ status = f"{ui_symbols.bullet}"
elif ext['status'] == 3:
- status = "⬤"
+ status = f"{ui_symbols.bullet}"
elif ext['status'] == 4:
- status = f"⬤"
+ status = f"{ui_symbols.bullet}"
elif ext['status'] == 5:
- status = "⬤"
+ status = f"{ui_symbols.bullet}"
elif ext['status'] == 6:
- status = "⬤"
+ status = f"{ui_symbols.bullet}"
else:
- status = "⬤"
+ status = f"{ui_symbols.bullet}"
else:
if updated < datetime.timestamp(datetime.now() - timedelta(6*30)):
- status = "⬤"
+ status = f"{ui_symbols.bullet}"
else:
- status = "⬤"
+ status = f"{ui_symbols.bullet}"
code += f"""
@@ -434,7 +434,12 @@ def create_ui():
check = gr.Button(value="Update all installed", variant="primary")
apply = gr.Button(value="Apply changes", variant="primary")
list_extensions()
- gr.HTML('Extension list
⯀ Refesh extension list to download latest list with status
⯀ Check status of an extension by looking at status icon before installing it
⯀ After any operation such as install/uninstall or enable/disable, please restart the server
')
+ gr.HTML('''
+ Extension list
+ - Refesh extension list to download latest list with status
+ - Check status of an extension by looking at status icon before installing it
+ - After any operation such as install/uninstall or enable/disable, please restart the server
+ ''')
gr.HTML('')
info = gr.HTML('')
extensions_table = gr.HTML(create_html(search_text.value, sort_column.value))
diff --git a/modules/ui_symbols.py b/modules/ui_symbols.py
index 6f93abb7d..57945835b 100644
--- a/modules/ui_symbols.py
+++ b/modules/ui_symbols.py
@@ -27,6 +27,7 @@ mark_diag = '※'
mark_flag = '⁜'
int_clip = '✎'
int_blip = '✐'
+bullet = '⃝'
"""
refresh = '🔄'
close = '🛗'
diff --git a/pyproject.toml b/pyproject.toml
index 1cc12d98a..12e1f8750 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -42,6 +42,7 @@ exclude = [
"modules/control/proc/normalbae/nets/submodules/efficientnet_repo/geffnet",
"modules/control/units/*_model.py",
"modules/control/units/*_pipe.py",
+ "modules/pipelines/*.py",
]
ignore = [
"A003", # Class attirbute shadowing builtin
diff --git a/scripts/faceid.py b/scripts/faceid.py
deleted file mode 100644
index b605c77d0..000000000
--- a/scripts/faceid.py
+++ /dev/null
@@ -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
diff --git a/scripts/photomaker.py b/scripts/photomaker.py
deleted file mode 100644
index c6e9c9b5e..000000000
--- a/scripts/photomaker.py
+++ /dev/null
@@ -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('