mirror of
https://github.com/vladmandic/automatic
synced 2026-09-18 16:54:33 +02:00
Vladimir Mandic
+97
-86
@@ -1,33 +1,37 @@
|
||||
import time
|
||||
import io
|
||||
import os
|
||||
import contextlib
|
||||
import gradio as gr
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
from modules import shared, devices, errors, sd_models, scripts, processing, processing_helpers
|
||||
from modules import shared, devices, errors, scripts, processing, processing_helpers, sd_models
|
||||
|
||||
|
||||
pulid = None
|
||||
debug = os.environ.get('SD_PULID_DEBUG', None) is not None
|
||||
|
||||
|
||||
class Script(scripts.Script):
|
||||
def __init__(self):
|
||||
self.images = []
|
||||
self.pulid = None
|
||||
self.cache = None
|
||||
super().__init__()
|
||||
# self.register() # pulid is script with processing override so xyz doesnt execute
|
||||
self.register() # pulid is script with processing override so xyz doesnt execute
|
||||
|
||||
def title(self):
|
||||
return 'PuLID'
|
||||
|
||||
def show(self, _is_img2img):
|
||||
return not _is_img2img
|
||||
return shared.native
|
||||
|
||||
def dependencies(self):
|
||||
from installer import install, installed
|
||||
# if not installed('apex', reload=False, quiet=True):
|
||||
# install('apex', 'apex', ignore=False)
|
||||
if not installed('insightface', reload=False, quiet=True):
|
||||
install('insightface', 'insightface', ignore=False)
|
||||
install('albumentations==1.4.3', 'albumentations', ignore=False, reinstall=True)
|
||||
install('pydantic==1.10.15', 'pydantic', ignore=False, reinstall=True)
|
||||
# if not installed('apex', reload=False, quiet=True):
|
||||
# install('apex', 'apex', ignore=False)
|
||||
|
||||
def register(self): # register xyz grid elements
|
||||
def apply_field(field):
|
||||
@@ -39,7 +43,8 @@ class Script(scripts.Script):
|
||||
import sys
|
||||
xyz_classes = [v for k, v in sys.modules.items() if 'xyz_grid_classes' in k][0]
|
||||
xyz_classes.axis_options.append(xyz_classes.AxisOption("[PuLID] Strength", float, apply_field("pulid_strength")))
|
||||
xyz_classes.axis_options.append(xyz_classes.AxisOption("[PuLID] Zero", float, apply_field("pulid_zero")))
|
||||
xyz_classes.axis_options.append(xyz_classes.AxisOption("[PuLID] Zero", int, apply_field("pulid_zero")))
|
||||
xyz_classes.axis_options.append(xyz_classes.AxisOption("[PuLID] Ortho", str, apply_field("pulid_ortho"), choices=lambda: ['off', 'v1', 'v2']))
|
||||
|
||||
def load_images(self, files):
|
||||
self.images = []
|
||||
@@ -79,117 +84,123 @@ class Script(scripts.Script):
|
||||
return [strength, zero, sampler, ortho, gallery]
|
||||
|
||||
def run(self, p: processing.StableDiffusionProcessing, strength: float = 0.8, zero: int = 20, sampler: str = 'dpmpp_sde', ortho: str = 'v2', gallery: list = []): # pylint: disable=arguments-differ
|
||||
global pulid # pylint: disable=global-statement
|
||||
images = []
|
||||
try:
|
||||
if len(gallery) == 0:
|
||||
gallery = self.images
|
||||
images = [Image.open(f['name']) for f in gallery if isinstance(f, dict)]
|
||||
from modules.api.api import decode_base64_to_image
|
||||
images = getattr(p, 'pulid_images', self.images)
|
||||
images = [decode_base64_to_image(image) if isinstance(image, str) else image for image in images]
|
||||
else:
|
||||
images = [Image.open(f['name']) if isinstance(f, dict) else f for f in gallery]
|
||||
images = [np.array(image) for image in images]
|
||||
except Exception as e:
|
||||
shared.log.error(f'PuLID: failed to load images: {e}')
|
||||
return None
|
||||
if len(images) == 0:
|
||||
shared.log.error('PuLID: no images loaded')
|
||||
shared.log.error('PuLID: no images')
|
||||
return None
|
||||
supported_model_list = ['sdxl']
|
||||
if shared.sd_model_type not in supported_model_list:
|
||||
shared.log.error(f'PuLID: class={shared.sd_model.__class__.__name__} model={shared.sd_model_type} required={supported_model_list}')
|
||||
return None
|
||||
if pulid is None:
|
||||
if self.pulid is None:
|
||||
self.dependencies()
|
||||
try:
|
||||
from modules import pulid # pylint: disable=redefined-outer-name
|
||||
self.pulid = pulid
|
||||
# from diffusers import pipelines
|
||||
# pipelines.auto_pipeline.AUTO_TEXT2IMAGE_PIPELINES_MAPPING["pilid"] = pulid.StableDiffusionXLPuLIDPipeline
|
||||
# pipelines.auto_pipeline.AUTO_IMAGE2IMAGE_PIPELINES_MAPPING["omnigen"] = pulid.StableDiffusionXLPuLIDPipelineImg2Img
|
||||
except Exception as e:
|
||||
shared.log.error(f'PuLID: failed to import library: {e}')
|
||||
return None
|
||||
# import os
|
||||
# import importlib
|
||||
# module_path = os.path.join(os.path.dirname(__file__), '..', 'pulid', '__init__.py')
|
||||
# module_spec = importlib.util.spec_from_file_location('pulid', module_path)
|
||||
# pulid = importlib.util.module_from_spec(module_spec)
|
||||
# module_spec.loader.exec_module(pulid)
|
||||
if pulid is None:
|
||||
shared.log.error('PuLID: failed to load PuLID library')
|
||||
return None
|
||||
if self.pulid is None:
|
||||
shared.log.error('PuLID: failed to load PuLID library')
|
||||
return None
|
||||
if p.batch_size > 1:
|
||||
shared.log.warning('PuLID: batch size not supported')
|
||||
p.batch_size = 1
|
||||
|
||||
strength = getattr(p, 'pulid_strength', strength)
|
||||
zero = getattr(p, 'pulid_zero', zero)
|
||||
ortho = getattr(p, 'pulid_ortho', ortho)
|
||||
|
||||
processing.fix_seed(p)
|
||||
pipe = None
|
||||
if shared.sd_model_type == 'sdxl':
|
||||
# TODO pulid has monolithic inference so not really working with offloading
|
||||
sd_models.move_model(shared.sd_model, devices.device)
|
||||
sd_models.move_model(shared.sd_model.vae, devices.device)
|
||||
sd_models.move_model(shared.sd_model.unet, devices.device)
|
||||
sd_models.move_model(shared.sd_model.text_encoder, devices.device)
|
||||
sd_models.move_model(shared.sd_model.text_encoder_2, devices.device)
|
||||
if shared.sd_model_type == 'sdxl' and not hasattr(shared.sd_model, 'pipe'):
|
||||
try:
|
||||
pipe = pulid.PuLIDPipelineXL(
|
||||
pipe =shared.sd_model,
|
||||
device=devices.device,
|
||||
sampler=sampler,
|
||||
cache_dir=shared.opts.hfcache_dir,
|
||||
)
|
||||
stdout = io.StringIO()
|
||||
ctx = contextlib.nullcontext if debug else contextlib.redirect_stdout(stdout)
|
||||
with ctx:
|
||||
shared.sd_model = self.pulid.StableDiffusionXLPuLIDPipeline(
|
||||
pipe =shared.sd_model,
|
||||
device=devices.device,
|
||||
sampler=sampler,
|
||||
cache_dir=shared.opts.hfcache_dir,
|
||||
)
|
||||
shared.sd_model.no_recurse = True
|
||||
sd_models.copy_diffuser_options(shared.sd_model, shared.sd_model.pipe)
|
||||
sd_models.move_model(shared.sd_model, devices.device) # move pipeline to device
|
||||
sd_models.set_diffuser_options(shared.sd_model, vae=None, op='model')
|
||||
devices.torch_gc()
|
||||
except Exception as e:
|
||||
shared.log.error(f'PuLID: failed to create pipeline: {e}')
|
||||
errors.display(e, 'PuLID')
|
||||
return None
|
||||
if pipe is None:
|
||||
return None
|
||||
shared.state.begin('PuLID')
|
||||
shared.log.info(f'PuLID: class={pipe.__class__.__name__} strength={strength} zero={zero} ortho={ortho} sampler={sampler} images={[i.shape for i in images]}')
|
||||
|
||||
pipe.debug_img_list = []
|
||||
pulid.attention.NUM_ZERO = zero
|
||||
if ortho == 'v2':
|
||||
pulid.attention.ORTHO = False
|
||||
pulid.attention.ORTHO_v2 = True
|
||||
elif ortho == 'v1':
|
||||
pulid.attention.ORTHO = True
|
||||
pulid.attention.ORTHO_v2 = False
|
||||
else:
|
||||
pulid.attention.ORTHO = False
|
||||
pulid.attention.ORTHO_v2 = False
|
||||
shared.log.info(f'PuLID: class={shared.sd_model.__class__.__name__} strength={strength} zero={zero} ortho={ortho} sampler={sampler} images={[i.shape for i in images]}')
|
||||
self.pulid.attention.NUM_ZERO = zero
|
||||
self.pulid.attention.ORTHO = ortho == 'v1'
|
||||
self.pulid.attention.ORTHO_v2 = ortho == 'v2'
|
||||
images = [self.pulid.resize(image, 1024) for image in images]
|
||||
shared.sd_model.debug_img_list = []
|
||||
uncond_id_embedding, id_embedding = shared.sd_model.get_id_embedding(images)
|
||||
|
||||
t0 = time.time()
|
||||
images = [pulid.resize(image, 1024) for image in images]
|
||||
outputs = []
|
||||
infotexts = []
|
||||
seeds = []
|
||||
prompts = []
|
||||
negative_prompts = []
|
||||
|
||||
for _n in range(p.n_iter):
|
||||
seed = processing_helpers.get_fixed_seed(p.seed)
|
||||
prompt = shared.prompt_styles.apply_styles_to_prompt(p.prompt, p.styles)
|
||||
negative_prompt = shared.prompt_styles.apply_negative_styles_to_prompt(p.negative_prompt, p.styles)
|
||||
if debug: # run pipeline directly
|
||||
shared.state.begin('PuLID')
|
||||
processing.fix_seed(p)
|
||||
p.seed = processing_helpers.get_fixed_seed(p.seed)
|
||||
p.prompt = shared.prompt_styles.apply_styles_to_prompt(p.prompt, p.styles)
|
||||
p.negative_prompt = shared.prompt_styles.apply_negative_styles_to_prompt(p.negative_prompt, p.styles)
|
||||
with devices.inference_context():
|
||||
uncond_id_embedding, id_embedding = pipe.get_id_embedding(images)
|
||||
output = pipe.inference(prompt, (1, p.height, p.width), negative_prompt, id_embedding, uncond_id_embedding, strength, p.cfg_scale, p.steps, seed)[0]
|
||||
if output is not None:
|
||||
outputs.append(output)
|
||||
infotexts.append(processing.create_infotext(p))
|
||||
seeds.append(seed)
|
||||
prompts.append(prompt)
|
||||
negative_prompts.append(negative_prompt)
|
||||
|
||||
interim = [Image.fromarray(face) for face in pipe.debug_img_list]
|
||||
t1 = time.time()
|
||||
shared.log.debug(f'PuLID: output={output} interim={interim} time={t1-t0:.2f}')
|
||||
|
||||
if len(outputs) > 0:
|
||||
p.prompt = prompts[0]
|
||||
p.negative_prompt = negative_prompts[0]
|
||||
p.seed = seeds[0]
|
||||
p.all_prompts = prompts
|
||||
p.all_negative_prompts = negative_prompts
|
||||
p.all_seeds = seeds
|
||||
output = shared.sd_model(
|
||||
prompt=p.prompt,
|
||||
negative_prompt=p.negative_prompt,
|
||||
width=p.width,
|
||||
height=p.height,
|
||||
seed=p.seed,
|
||||
num_inference_steps=p.steps,
|
||||
guidance_scale=p.cfg_scale,
|
||||
id_embedding=id_embedding,
|
||||
uncond_id_embedding=uncond_id_embedding,
|
||||
id_scale=strength,
|
||||
)[0]
|
||||
info = processing.create_infotext(p)
|
||||
processed = processing.Processed(p, [output], info=info)
|
||||
shared.state.end('PuLID')
|
||||
else: # let processing run the pipeline
|
||||
p.task_args['id_embedding'] = id_embedding
|
||||
p.task_args['uncond_id_embedding'] = uncond_id_embedding
|
||||
p.task_args['id_scale'] = strength
|
||||
if len(getattr(p, 'init_images', [])) > 0:
|
||||
p.task_args['image'] = p.init_images[0]
|
||||
p.task_args['strength'] = p.denoising_strength
|
||||
p.extra_generation_params["PuLID"] = f'Strength={strength} Zero={zero} Ortho={ortho}'
|
||||
processed = processing.Processed(p, outputs, infotexts=infotexts)
|
||||
if getattr(p, 'xyz', False): # xyz will run its own processing
|
||||
return None
|
||||
processed: processing.Processed = processing.process_images(p) # runs processing using main loop
|
||||
|
||||
shared.state.end('PuLID')
|
||||
# interim = [Image.fromarray(img) for img in shared.sd_model.debug_img_list]
|
||||
# shared.log.debug(f'PuLID: time={t1-t0:.2f}')
|
||||
return processed
|
||||
|
||||
def after(self, p: processing.StableDiffusionProcessing, processed: processing.Processed, *args): # pylint: disable=unused-argument
|
||||
if hasattr(shared.sd_model, 'pipe') and shared.sd_model_type == "sdxl":
|
||||
if hasattr(shared.sd_model, 'app'):
|
||||
shared.sd_model.app = None
|
||||
shared.sd_model.ip_adapter = None
|
||||
shared.sd_model.face_helper = None
|
||||
shared.sd_model.clip_vision_model = None
|
||||
shared.sd_model.handler_ante = None
|
||||
devices.torch_gc(force=True)
|
||||
shared.sd_model = shared.sd_model.pipe
|
||||
# shared.log.debug(f'PuLID restore: class={shared.sd_model.__class__.__name__}')
|
||||
return processed
|
||||
|
||||
@@ -258,6 +258,7 @@ class Script(scripts.Script):
|
||||
def cell(x, y, z, ix, iy, iz):
|
||||
if shared.state.interrupted:
|
||||
return processing.Processed(p, [], p.seed, "")
|
||||
p.xyz = True
|
||||
pc = copy(p)
|
||||
pc.override_settings_restore_afterwards = False
|
||||
pc.styles = pc.styles[:]
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from scripts.xyz_grid_shared import apply_field, apply_task_args, apply_setting, apply_prompt, apply_order, apply_sampler, apply_hr_sampler_name, confirm_samplers, apply_checkpoint, apply_refiner, apply_unet, apply_dict, apply_clip_skip, apply_vae, list_lora, apply_lora, apply_te, apply_styles, apply_upscaler, apply_context, apply_detailer, apply_override, apply_processing, apply_options, apply_seed, format_value_add_label, format_value, format_value_join_list, do_nothing, format_nothing, str_permutations # pylint: disable=no-name-in-module
|
||||
from scripts.xyz_grid_shared import apply_field, apply_task_args, apply_setting, apply_prompt, apply_order, apply_sampler, apply_hr_sampler_name, confirm_samplers, apply_checkpoint, apply_refiner, apply_unet, apply_dict, apply_clip_skip, apply_vae, list_lora, apply_lora, apply_te, apply_styles, apply_upscaler, apply_context, apply_detailer, apply_override, apply_processing, apply_options, apply_seed, format_value_add_label, format_value, format_value_join_list, do_nothing, format_nothing, str_permutations # pylint: disable=no-name-in-module, unused-import
|
||||
from modules import shared, shared_items, sd_samplers, ipadapter, sd_models, sd_vae, sd_unet
|
||||
|
||||
|
||||
|
||||
@@ -273,6 +273,7 @@ class Script(scripts.Script):
|
||||
def cell(x, y, z, ix, iy, iz):
|
||||
if shared.state.interrupted:
|
||||
return processing.Processed(p, [], p.seed, "")
|
||||
p.xyz = True
|
||||
pc = copy(p)
|
||||
pc.override_settings_restore_afterwards = False
|
||||
pc.styles = pc.styles[:]
|
||||
|
||||
Reference in New Issue
Block a user