sdnext/modules/ipadapter.py

"""
Lightweight IP-Adapter applied to existing pipeline in Diffusers
- Downloads image_encoder or first usage (2.5GB)
- Introduced via: https://github.com/huggingface/diffusers/pull/5713
- IP adapters: https://huggingface.co/h94/IP-Adapter
"""

import os
import time
import json
from PIL import Image
import diffusers
import transformers
from modules import processing, shared, devices, sd_models, errors, model_quant


clip_loaded = None
adapters_loaded = []
CLIP_ID = "h94/IP-Adapter"
OPEN_ID = "openai/clip-vit-large-patch14"
SIGLIP_ID = 'google/siglip-so400m-patch14-384'
ADAPTERS_NONE = {
    'None': { 'name': 'none', 'repo': 'none', 'subfolder': 'none' },
}
ADAPTERS_SD15 = {
    'None': { 'name': 'none', 'repo': 'none', 'subfolder': 'none' },
    'Base': { 'name': 'ip-adapter_sd15.safetensors', 'repo': 'h94/IP-Adapter', 'subfolder': 'models' },
    'Base ViT-G': { 'name': 'ip-adapter_sd15_vit-G.safetensors', 'repo': 'h94/IP-Adapter', 'subfolder': 'models' },
    'Light': { 'name': 'ip-adapter_sd15_light.safetensors', 'repo': 'h94/IP-Adapter', 'subfolder': 'models' },
    'Plus': { 'name': 'ip-adapter-plus_sd15.safetensors', 'repo': 'h94/IP-Adapter', 'subfolder': 'models' },
    'Plus Face': { 'name': 'ip-adapter-plus-face_sd15.safetensors', 'repo': 'h94/IP-Adapter', 'subfolder': 'models' },
    'Full Face': { 'name': 'ip-adapter-full-face_sd15.safetensors', 'repo': 'h94/IP-Adapter', 'subfolder': 'models' },
    'Ostris Composition ViT-H': { 'name': 'ip_plus_composition_sd15.safetensors', 'repo': 'ostris/ip-composition-adapter', 'subfolder': '' },
}
ADAPTERS_SDXL = {
    'None': { 'name': 'none', 'repo': 'none', 'subfolder': 'none' },
    'Base SDXL': { 'name': 'ip-adapter_sdxl.safetensors', 'repo': 'h94/IP-Adapter', 'subfolder': 'sdxl_models' },
    'Base ViT-H SDXL': { 'name': 'ip-adapter_sdxl_vit-h.safetensors', 'repo': 'h94/IP-Adapter', 'subfolder': 'sdxl_models' },
    'Plus ViT-H SDXL': { 'name': 'ip-adapter-plus_sdxl_vit-h.safetensors', 'repo': 'h94/IP-Adapter', 'subfolder': 'sdxl_models' },
    'Plus Face ViT-H SDXL': { 'name': 'ip-adapter-plus-face_sdxl_vit-h.safetensors', 'repo': 'h94/IP-Adapter', 'subfolder': 'sdxl_models' },
    'Ostris Composition ViT-H SDXL': { 'name': 'ip_plus_composition_sdxl.safetensors', 'repo': 'ostris/ip-composition-adapter', 'subfolder': '' },
}
ADAPTERS_SD3 = {
    'None': { 'name': 'none', 'repo': 'none', 'subfolder': 'none' },
    'InstantX Large': { 'name': 'ip-adapter_diffusers.safetensors', 'repo': 'InstantX/SD3.5-Large-IP-Adapter', 'subfolder': 'none', 'revision': 'refs/pr/10' },
}
ADAPTERS_F1 = {
    'None': { 'name': 'none', 'repo': 'none', 'subfolder': 'none' },
    'XLabs AI v1': { 'name': 'ip_adapter.safetensors', 'repo': 'XLabs-AI/flux-ip-adapter', 'subfolder': 'none' },
    'XLabs AI v2': { 'name': 'ip_adapter.safetensors', 'repo': 'XLabs-AI/flux-ip-adapter-v2', 'subfolder': 'none' },
}
ADAPTERS = { **ADAPTERS_SD15, **ADAPTERS_SDXL, **ADAPTERS_SD3, **ADAPTERS_F1 }
ADAPTERS_ALL = { **ADAPTERS_SD15, **ADAPTERS_SDXL, **ADAPTERS_SD3, **ADAPTERS_F1 }


def get_adapters():
    global ADAPTERS # pylint: disable=global-statement
    if shared.sd_model_type == 'sd':
        ADAPTERS = ADAPTERS_SD15
    elif shared.sd_model_type == 'sdxl':
        ADAPTERS = ADAPTERS_SDXL
    elif shared.sd_model_type == 'sd3':
        ADAPTERS = ADAPTERS_SD3
    elif shared.sd_model_type == 'f1':
        ADAPTERS = ADAPTERS_F1
    else:
        ADAPTERS = ADAPTERS_NONE
    return list(ADAPTERS)


def get_images(input_images):
    output_images = []
    if input_images is None or len(input_images) == 0:
        shared.log.error('IP adapter: no init images')
        return None
    if shared.sd_model_type not in ['sd', 'sdxl', 'sd3', 'f1']:
        shared.log.error('IP adapter: base model not supported')
        return None
    if isinstance(input_images, str):
        from modules.api.api import decode_base64_to_image
        input_images = decode_base64_to_image(input_images).convert("RGB")
    input_images = input_images.copy()
    if not isinstance(input_images, list):
        input_images = [input_images]
    for image in input_images:
        if image is None:
            continue
        if isinstance(image, list):
            output_images.append(get_images(image)) # recursive
        elif isinstance(image, Image.Image):
            output_images.append(image)
        elif isinstance(image, str):
            from modules.api.api import decode_base64_to_image
            decoded_image = decode_base64_to_image(image).convert("RGB")
            output_images.append(decoded_image)
        elif hasattr(image, 'name'): # gradio gallery entry
            pil_image = Image.open(image.name)
            pil_image.load()
            output_images.append(pil_image)
        else:
            shared.log.error(f'IP adapter: unknown input: {image}')
    return output_images


def get_scales(adapter_scales, adapter_images):
    output_scales = [adapter_scales] if not isinstance(adapter_scales, list) else adapter_scales
    while len(output_scales) < len(adapter_images):
        output_scales.append(output_scales[-1])
    return output_scales


def get_crops(adapter_crops, adapter_images):
    output_crops = [adapter_crops] if not isinstance(adapter_crops, list) else adapter_crops
    while len(output_crops) < len(adapter_images):
        output_crops.append(output_crops[-1])
    return output_crops


def crop_images(images, crops):
    try:
        for i in range(len(images)):
            if crops[i]:
                from modules.shared import yolo # pylint: disable=no-name-in-module
                cropped = []
                for image in images[i]:
                    faces = yolo.predict('face-yolo8n', image)
                    if len(faces) > 0:
                        cropped.append(faces[0].item)
                if len(cropped) == len(images[i]):
                    images[i] = cropped
                else:
                    shared.log.error(f'IP adapter: failed to crop image: source={len(images[i])} faces={len(cropped)}')
    except Exception as e:
        shared.log.error(f'IP adapter: failed to crop image: {e}')
    if shared.sd_model_type == 'sd3' and len(images) == 1:
        return images[0]
    return images


def unapply(pipe, unload: bool = False): # pylint: disable=arguments-differ
    if len(adapters_loaded) == 0:
        return
    try:
        if hasattr(pipe, 'set_ip_adapter_scale'):
            pipe.set_ip_adapter_scale(0)
            if unload:
                shared.log.debug('IP adapter unload')
                pipe.unload_ip_adapter()
        if hasattr(pipe, 'unet') and pipe.unet is not None:
            module = pipe.unet
        elif hasattr(pipe, 'transformer'):
            module = pipe.transformer
        else:
            module = None
        if module is not None and hasattr(module, 'config') and module.config.encoder_hid_dim_type == 'ip_image_proj':
            pipe.unet.encoder_hid_proj = None
            pipe.config.encoder_hid_dim_type = None
            pipe.unet.set_default_attn_processor()
    except Exception:
        pass


def load_image_encoder(pipe: diffusers.DiffusionPipeline, adapter_names: list[str]):
    global clip_loaded # pylint: disable=global-statement
    for adapter_name in adapter_names:
        # which clip to use
        clip_repo = CLIP_ID
        if 'ViT' not in adapter_name: # defaults per model
            clip_subfolder = 'models/image_encoder' if shared.sd_model_type == 'sd' else 'sdxl_models/image_encoder'
        if 'ViT-H' in adapter_name:
            clip_subfolder = 'models/image_encoder' # this is vit-h
        elif 'ViT-G' in adapter_name:
            clip_subfolder = 'sdxl_models/image_encoder' # this is vit-g
        else:
            if shared.sd_model_type == 'sd':
                clip_subfolder = 'models/image_encoder'
            elif shared.sd_model_type == 'sdxl':
                clip_subfolder = 'sdxl_models/image_encoder'
            elif shared.sd_model_type == 'sd3':
                clip_repo = SIGLIP_ID
                clip_subfolder = None
            elif shared.sd_model_type == 'f1':
                clip_repo = OPEN_ID
                clip_subfolder = None
            else:
                shared.log.error(f'IP adapter: unknown model type: {adapter_name}')
                return False

    # load image encoder used by ip adapter
    if pipe.image_encoder is None or clip_loaded != f'{clip_repo}/{clip_subfolder}':
        jobid = shared.state.begin('Load encoder')
        try:
            offline_config = { 'local_files_only': True } if shared.opts.offline_mode else {}
            if shared.sd_model_type == 'sd3':
                image_encoder = transformers.SiglipVisionModel.from_pretrained(clip_repo, torch_dtype=devices.dtype, cache_dir=shared.opts.hfcache_dir, **offline_config)
            else:
                if clip_subfolder is None:
                    image_encoder = transformers.CLIPVisionModelWithProjection.from_pretrained(clip_repo, torch_dtype=devices.dtype, cache_dir=shared.opts.hfcache_dir, use_safetensors=True, **offline_config)
                    shared.log.debug(f'IP adapter load: encoder="{clip_repo}" cls={pipe.image_encoder.__class__.__name__}')
                else:
                    image_encoder = transformers.CLIPVisionModelWithProjection.from_pretrained(clip_repo, subfolder=clip_subfolder, torch_dtype=devices.dtype, cache_dir=shared.opts.hfcache_dir, use_safetensors=True, **offline_config)
                    shared.log.debug(f'IP adapter load: encoder="{clip_repo}/{clip_subfolder}" cls={pipe.image_encoder.__class__.__name__}')
            sd_models.clear_caches()
            image_encoder = model_quant.do_post_load_quant(image_encoder, allow=True)
            if hasattr(pipe, 'register_modules'):
                pipe.register_modules(image_encoder=image_encoder)
            else:
                pipe.image_encoder = image_encoder
            clip_loaded = f'{clip_repo}/{clip_subfolder}'
        except Exception as e:
            shared.log.error(f'IP adapter load: encoder="{clip_repo}/{clip_subfolder}" {e}')
            errors.display(e, 'IP adapter: type=encoder')
            return False
        shared.state.end(jobid)
    sd_models.move_model(pipe.image_encoder, devices.device)
    return True


def load_feature_extractor(pipe):
    # load feature extractor used by ip adapter
    if pipe.feature_extractor is None:
        try:
            jobid = shared.state.begin('Load extractor')
            offline_config = { 'local_files_only': True } if shared.opts.offline_mode else {}
            if shared.sd_model_type == 'sd3':
                feature_extractor = transformers.SiglipImageProcessor.from_pretrained(SIGLIP_ID, torch_dtype=devices.dtype, cache_dir=shared.opts.hfcache_dir, **offline_config)
            else:
                feature_extractor = transformers.CLIPImageProcessor()
            if hasattr(pipe, 'register_modules'):
                pipe.register_modules(feature_extractor=feature_extractor)
            else:
                pipe.feature_extractor = feature_extractor
                sd_models.apply_balanced_offload(pipe.feature_extractor)
            shared.log.debug(f'IP adapter load: extractor={pipe.feature_extractor.__class__.__name__}')
        except Exception as e:
            shared.log.error(f'IP adapter load: extractor {e}')
            errors.display(e, 'IP adapter: type=extractor')
            return False
        shared.state.end(jobid)
    return True


def parse_params(p: processing.StableDiffusionProcessing, adapters: list, adapter_scales: list[float], adapter_crops: list[bool], adapter_starts: list[float], adapter_ends: list[float], adapter_images: list):
    if hasattr(p, 'ip_adapter_scales'):
        adapter_scales = p.ip_adapter_scales
    if hasattr(p, 'ip_adapter_crops'):
        adapter_crops = p.ip_adapter_crops
    if hasattr(p, 'ip_adapter_starts'):
        adapter_starts = p.ip_adapter_starts
    if hasattr(p, 'ip_adapter_ends'):
        adapter_ends = p.ip_adapter_ends
    if hasattr(p, 'ip_adapter_images'):
        adapter_images = p.ip_adapter_images
    adapter_images = get_images(adapter_images)
    if hasattr(p, 'ip_adapter_masks') and len(p.ip_adapter_masks) > 0:
        adapter_masks = p.ip_adapter_masks
        adapter_masks = get_images(adapter_masks)
    else:
        adapter_masks = []
    if len(adapter_masks) > 0:
        from diffusers.image_processor import IPAdapterMaskProcessor
        mask_processor = IPAdapterMaskProcessor()
        for i in range(len(adapter_masks)):
            adapter_masks[i] = mask_processor.preprocess(adapter_masks[i], height=p.height, width=p.width)
        adapter_masks = mask_processor.preprocess(adapter_masks, height=p.height, width=p.width)
    if adapter_images is None:
        shared.log.error('IP adapter: no image provided')
        return [], [], [], [], [], []
    if len(adapters) < len(adapter_images):
        adapter_images = adapter_images[:len(adapters)]
    if len(adapters) < len(adapter_masks):
        adapter_masks = adapter_masks[:len(adapters)]
    if len(adapter_masks) > 0 and len(adapter_masks) != len(adapter_images):
        shared.log.error('IP adapter: image and mask count mismatch')
        return [], [], [], [], [], []
    adapter_scales = get_scales(adapter_scales, adapter_images)
    p.ip_adapter_scales = adapter_scales.copy()
    adapter_crops = get_crops(adapter_crops, adapter_images)
    p.ip_adapter_crops = adapter_crops.copy()
    adapter_starts = get_scales(adapter_starts, adapter_images)
    p.ip_adapter_starts = adapter_starts.copy()
    adapter_ends = get_scales(adapter_ends, adapter_images)
    p.ip_adapter_ends = adapter_ends.copy()
    return adapter_images, adapter_masks, adapter_scales, adapter_crops, adapter_starts, adapter_ends


def apply(pipe, p: processing.StableDiffusionProcessing, adapter_names=[], adapter_scales=[1.0], adapter_crops=[False], adapter_starts=[0.0], adapter_ends=[1.0], adapter_images=[]):
    global adapters_loaded # pylint: disable=global-statement
    # overrides
    if hasattr(p, 'ip_adapter_names'):
        if isinstance(p.ip_adapter_names, str):
            p.ip_adapter_names = [p.ip_adapter_names]
        adapters = [ADAPTERS_ALL.get(adapter_name, None) for adapter_name in p.ip_adapter_names if adapter_name is not None and adapter_name.lower() != 'none']
        adapter_names = p.ip_adapter_names
    else:
        if isinstance(adapter_names, str):
            adapter_names = [adapter_names]
        adapters = [ADAPTERS.get(adapter_name, None) for adapter_name in adapter_names if adapter_name.lower() != 'none']

    if len(adapters) == 0:
        unapply(pipe, getattr(p, 'ip_adapter_unload', False))
        if hasattr(p, 'ip_adapter_images'):
            del p.ip_adapter_images
        return False
    if shared.sd_model_type not in ['sd', 'sdxl', 'sd3', 'f1']:
        shared.log.error(f'IP adapter: model={shared.sd_model_type} class={pipe.__class__.__name__} not supported')
        return False

    adapter_images, adapter_masks, adapter_scales, adapter_crops, adapter_starts, adapter_ends = parse_params(p, adapters, adapter_scales, adapter_crops, adapter_starts, adapter_ends, adapter_images)

    # init code
    if pipe is None:
        return False
    if len(adapter_images) == 0:
        shared.log.error('IP adapter: no image provided')
        adapters = [] # unload adapter if previously loaded as it will cause runtime errors
    if len(adapters) == 0:
        unapply(pipe, getattr(p, 'ip_adapter_unload', False))
        if hasattr(p, 'ip_adapter_images'):
            del p.ip_adapter_images
        return False
    if not hasattr(pipe, 'load_ip_adapter'):
        shared.log.error(f'IP adapter: pipeline not supported: {pipe.__class__.__name__}')
        return False

    if not load_image_encoder(pipe, adapter_names):
        return False

    if not load_feature_extractor(pipe):
        return False

    # main code
    try:
        t0 = time.time()
        repos = [adapter.get('repo', None) for adapter in adapters if adapter.get('repo', 'none') != 'none']
        subfolders = [adapter.get('subfolder', None) for adapter in adapters if adapter.get('subfolder', 'none') != 'none']
        names = [adapter.get('name', None) for adapter in adapters if adapter.get('name', 'none') != 'none']
        revisions = [adapter.get('revision', None) for adapter in adapters if adapter.get('revision', 'none') != 'none']
        kwargs = {}
        if len(repos) == 1:
            repos = repos[0]
        if len(subfolders) > 0:
            kwargs['subfolder'] = subfolders if len(subfolders) > 1 else subfolders[0]
        if len(names) > 0:
            kwargs['weight_name'] = names if len(names) > 1 else names[0]
        if len(revisions) > 0:
            kwargs['revision'] = revisions[0]
        if shared.opts.offline_mode:
            kwargs["local_files_only"] = True
        pipe.load_ip_adapter(repos, **kwargs)
        adapters_loaded = names
        if hasattr(p, 'ip_adapter_layers'):
            pipe.set_ip_adapter_scale(p.ip_adapter_layers)
            ip_str = ';'.join(adapter_names) + ':' + json.dumps(p.ip_adapter_layers)
        else:
            for i in range(len(adapter_scales)):
                if adapter_starts[i] > 0:
                    adapter_scales[i] = 0.00
            pipe.set_ip_adapter_scale(adapter_scales if len(adapter_scales) > 1 else adapter_scales[0])
            ip_str =  [f'{os.path.splitext(adapter)[0]}:{scale}:{start}:{end}:{crop}' for adapter, scale, start, end, crop in zip(adapter_names, adapter_scales, adapter_starts, adapter_ends, adapter_crops)]
        if hasattr(pipe, 'transformer') and 'Nunchaku' in pipe.transformer.__class__.__name__:
            if isinstance(repos, str):
                sd_models.clear_caches(full=True)
                import accelerate
                accelerate.hooks.remove_hook_from_module(pipe.transformer, recurse=True)
                pipe.transformer = pipe.transformer.to(devices.device)
                from nunchaku.models.ip_adapter.diffusers_adapters import apply_IPA_on_pipe
                apply_IPA_on_pipe(pipe, ip_adapter_scale=adapter_scales[0], repo_id=repos)
                pipe = sd_models.apply_balanced_offload(pipe)
                shared.log.debug(f'IP adapter load: engine=nunchaku scale={adapter_scales[0]} repo="{repos}"')
            else:
                shared.log.error('IP adapter: Nunchaku only supports single adapter')
        p.task_args['ip_adapter_image'] = crop_images(adapter_images, adapter_crops)
        if len(adapter_masks) > 0:
            p.cross_attention_kwargs = { 'ip_adapter_masks': adapter_masks }
        p.extra_generation_params["IP Adapter"] = ';'.join(ip_str)
        t1 = time.time()
        shared.log.info(f'IP adapter: {ip_str} image={adapter_images} mask={adapter_masks is not None} time={t1-t0:.2f}')
    except Exception as e:
        shared.log.error(f'IP adapter load: adapters={adapter_names} repo={repos} folders={subfolders} names={names} {e}')
        errors.display(e, 'IP adapter: type=adapter')
    return True