sdxlサーバ、モデルリロード版
モデルをクライエントから指定できるようにしたバージョン。
api_sdxl_v1_1_server_t.py サーバ本体
import torch
import cv2
from PIL import Image
#from n_lcm_lora_sdxl_i2i_t2i_class_v1 import DiffusersPipelineManager
from n_lcm_lora_sdxl_i2i_t2i_class_v1_1_t import DiffusersPipelineManager
import pickle
# =================================== FastAPI ==============================
from fastapi import FastAPI, File, UploadFile, Form, Query
from fastapi.responses import HTMLResponse,StreamingResponse,JSONResponse,Response
from pydantic import BaseModel
from io import BytesIO
import io
import json
import base64
import datetime
import string
import time
import gc
app = FastAPI()
#------ DiffusersPipelineManagerの定義と初期化
Df=DiffusersPipelineManager()
#------ 独自LoRAの定義
lola_models_list=[
["./models/animagine-xl-2.0/megu_sports_sdxl_v02.safetensors","style",0.8],
["./models/animagine-xl-2.0/anime-detailer-xl.safetensors","detaile",1.0],
["./models/animagine-xl-2.0/style-enhancer-xl.safetensors","enhancer",0.5]]
#----- pipelineのロード時の引数定義
model="./models/animagine-xl-2.0/animagine-xl-2.0.safetensors"
#model="cagliostrolab/animagine-xl-4.0"
#model="./models/animagine-xl-2.0/animagine-xl-3.1.safetensors"
#model="cagliostrolab/animagine-xl-3.1"
#model="stabilityai/stable-diffusion-xl-base-1.0"
lcm_lora_en=True
lcm_lora_path="./models/SSD-1B-anime/lcm-ssd1b-anime.safetensors"
vae_en=True
vae_path="madebyollin/sdxl-vae-fp16-fix"
lora_en=False
lola_models=lola_models_list
ddism_en=False
set_unet=True
torch_compile=False
def load_model(model, torch_compile=None, lcm_lora_en=None):
#----- pipelineのロード "t2i"・"i2i_pipe"・"inpaint_pipe"はデフォルトでロードでOK、他はVRAMの容量次第
print("pipelineのロード開始")
pipeline_name_list=[]
pipe_t2i=Df.pipeline_init("t2i",
pipeline=None,
model=model,
vae_en=vae_en,
vae_path=vae_path,
lcm_lora_en=lcm_lora_en,
lcm_lora_path=lcm_lora_path,
lora_en=lora_en,
lola_models=lola_models,
ddism_en=ddism_en,
set_unet=set_unet,
torch_compile=torch_compile
)
pipeline_name_list.append("t2i")
pipe_i2i=Df.pipeline_init("i2i_pipe",pipeline=pipe_t2i)#pipelineの使い回し
pipeline_name_list.append("pipe_i2i")
pipe_inpaint=Df.pipeline_init("inpaint_pipe",pipeline=pipe_t2i)#pipelineの使い回し
pipeline_name_list.append("pipe_inpaint")
return pipe_t2i,pipe_i2i,pipe_inpaint,pipeline_name_list
pipe_t2i,pipe_i2i,pipe_inpaint,pipeline_name_list=load_model(model)
def load_contrilnet(control_type):
if control_type=="canny":
control_pipe=Df.pipeline_init("canny",model=model) #Canny ContorolNET用pipeの新規定義
pipeline_name_list.append("canny")
elif control_type=="openpose":
control_pipe=Df.pipeline_init("openpose",model=model) #openpose ContorolNET用pipeの新規定義
pipeline_name_list.append("openpose")
elif control_type=="openpose_lola":
control_pipe=Df.pipeline_init("openpose_lola",model=model, lola_models=lola_models) #lola_models指定
pipeline_name_list.append("openpose_lola")
elif control_type=="depth":
control_pipe = Df.pipeline_init("depth",model=model)
pipeline_name_list.append("depth")
elif control_type=="t2i_adapter":
control_pipe=Df.pipeline_init("t2i_adapter",model=model_t2i, lora_en=False)#t2i_adapter用
pipeline_name_list.append("t2i_adapter")
elif control_type=="t2i_ip":
control_pipe=Df.pipeline_init("t2i_ip",model=model, lcm_lora_en=True, mul_ip=True, ip_scal1=0.9,ip_scal2=0.5)
pipeline_name_list.append("t2i_ip")
elif control_type=="i2i_ip":
control_pipe=Df.pipeline_init("i2i_ip",model=model, lcm_lora_en=True, mul_ip=True, ip_scal1=0.9,ip_scal2=0.5)
pipeline_name_list.append("i2i_ip")
elif control_type=="t2i_sip":
control_pipe=Df.pipeline_init("t2i_ip",model=model, lcm_lora_en=True, single_ip=True,ip_scal1=0.6)
pipeline_name_list.append("t2i_ip")
elif control_type=="i2i_sip":
control_pipe=Df.pipeline_init("i2i_ip",model=model, lcm_lora_en=True, single_ip=True,ip_scal1=0.6)
pipeline_name_list.append("i2i_ip")
return control_pipe
# 利用できるpipeline名 → i2i/canny/pipe_i2i/pipe_inpaint/canny/openpose//depth/t2i_adapter
#--------pipeline名からpipeを選択して返す
def pipline_sel(pipeline):
if pipeline=="t2i":
return pipe_t2i
elif pipeline=="pipe_i2i":
return pipe_i2i
elif pipeline=="pipe_inpaint":
return pipe_inpaint
elif pipeline=="canny":
return pipe_canny
elif pipeline=="openpose":
return pipe_openpose
elif pipeline=="depth":
return pipe_depth
elif pipeline=="t2i_adapter":
return pipe_t2i_adapter
elif pipeline=="t2i_sip":
return pipe_t2i_sip
elif pipeline=="t2i_mip":
return pipe_t2i_mip
elif pipeline=="i2i_sip":
return pipe_i2i_sip
elif pipeline=="i2i_mip":
return pipe_i2i_mip
elif pipeline=="inpaint_sip":
return pipe_inpaint_sip
else:
return pipe_t2i
#-------------------- FastAPI エンドポイント--------------------------
@app.post("/get_pipeline/")
def get_pipeline():
return pipeline_name_list
@app.post("/reload_model/")
def reload_model(model_id: str = Form(...), del_pipe: bool = Form(False),torch_compile: bool = Form(False),lcm_lora_en: bool = Form(False),):
global pipe_t2i, pipe_i2i, pipe_inpaint, pipeline_name_list
try:
if del_pipe:
print("Deleting previously loaded pipelines and freeing GPU VRAM.")
# 既存のパイプラインを CPU に移動し、メモリを解放
for attr in ["pipe_t2i", "pipe_i2i", "pipe_inpaint"]:
pipeline = globals().get(attr)
if pipeline is not None:
try:
print(f"Releasing {attr} components before moving to CPU...")
# 各コンポーネントを明示的に削除
if hasattr(pipeline, "unet"):
pipeline.unet = None
if hasattr(pipeline, "vae"):
pipeline.vae = None
if hasattr(pipeline, "text_encoder"):
pipeline.text_encoder = None
if hasattr(pipeline, "tokenizer"):
pipeline.tokenizer = None
# CPU に移動
pipeline.to("cpu")
print(f"{attr} moved to CPU.")
except Exception as e:
print(f"Failed to move {attr} to CPU:", e)
# 明示的に削除
del pipeline
globals()[attr] = None
# 明示的に削除
try:
del pipe_t2i, pipe_i2i, pipe_inpaint
except:
print("some pipe was not loaded")
# メモリ解放
torch.cuda.empty_cache() # GPU キャッシュを解放
gc.collect() # Python のガベージコレクション実行
pipeline_name_list = []
# モデルを再ロード
print("Reloading pipeline for", model_id)
pipe_t2i, pipe_i2i, pipe_inpaint, pipeline_name_list = load_model(model_id, torch_compile=torch_compile, lcm_lora_en=lcm_lora_en)
except:
print("Failed to reload pipeline.")
return {"status": "failed"}
return pipeline_name_list
@app.post("/get_canny/")
def get_canny(image: UploadFile = File(...), low_threshold : int = Form(100) , high_threshold: int = Form(200)):
image_data = image.file.read()
image_data = Image.open(BytesIO( image_data)) # バイナリデータをPIL形式に変換
image_data = image_data.convert("RGB")
canny_image = Df.get_canny(original_image=image_data , low_threshold = low_threshold , high_threshold =high_threshold)
canny_data = pickle.dumps(canny_image, 5) # tx_dataはpklデータ
return Response(content=canny_data , media_type="application/octet-stream")
@app.post("/get_openpose/")
def get_openpose(image: UploadFile = File(...)):
image_data = image.file.read()
image_data = Image.open(BytesIO( image_data)) # バイナリデータをPIL形式に変換
image_data = image_data.convert("RGB")
openpose_image = Df.get_openpose(image_data )
openpose_data = pickle.dumps(openpose_image, 5) # tx_dataはpklデータ
return Response(content= openpose_data , media_type="application/octet-stream")
@app.post("/get_depth_map/")
def get_depth_map(image: UploadFile = File(...)):
image_data = image.file.read()
image_data = Image.open(BytesIO( image_data)) # バイナリデータをPIL形式に変換
image_data = image_data.convert("RGB")
depth_image = Df.get_depth_map( image_data)
depth_data = pickle.dumps(depth_image , 5) # tx_dataはpklデータ
return Response(content= depth_data, media_type="application/octet-stream")
@app.post("/generate/")
def generate(
pipeline: str = Form(...),
image: UploadFile = File(None),
mask_image: UploadFile = File(None),
embe:bool = Form(False),
prompt_embe:list = Form(None),
prompt_embe0:str = Form(None),
prompt_embe1:str = Form(None),
prompt:str = Form(None),
prompt_2:str = Form(None),
negative_prompt:str = Form(None),
negative_prompt_2:str = Form(None),
num_inference_steps:int =Form(4),
strength:float = Form(0.1),
guidance_scale:float = Form(1.0),
seed:int= Form(0),
auto_seed:bool = Form(True),
height :int= Form(None),
width:int = Form(None),
num_images_per_prompt:int = Form(1),
guess_mode:bool = Form(None),
freeu:bool = Form(False),
freeu_list:list=Form(None),
output_type:str = Form("pil"),
crops_coords_top_left:list = Form([0,0]),
controlnet_conditioning_scale:float = Form(1.0),
adapter_conditioning_scale:float=Form(None),
adapter_conditioning_factor:int=Form(None),
ip_adapter:bool=Form(False),
style_images:UploadFile =File(None),
ip_image:UploadFile =File(None),
):
#−−−−−pipeの設定
try:
pipe = pipline_sel(pipeline)
#−−−−−conditioning, pooledがGPUにTensor形式でロードされるのでprompt_embeddingをサーバ側で実行
if embe:
prompt_embe=[prompt_embe0,prompt_embe1]
conditioning, pooled = Df.prompt_embedding(pipe , prompt_embe)
else: #embeがFalseならconditioning、pooled はNoneを設定
conditioning = None
pooled = None
#−−−−−Seed値を取得 auto_seedがTrueならtorch.initial_seed( )
if seed == None:
if auto_seed: #seedが無くてauto_seedが有効(True)の時
seedx= torch.initial_seed( )
generator = torch.manual_seed(seedx )
else: #seedが無くてauto_seedが無効(False)の時
generator = None
else: #seedが指定された場合
generator = torch.manual_seed(seed )
"""
if auto_seed:
seedx= torch.initial_seed( )
generator = torch.manual_seed(seedx )
else:
if seed == None: #auto_seedがFalseでseedがない場合
generator = None
else: #auto_seedがFalseでseedが指定された場合
generator = torch.manual_seed(seed )
"""
# −−−−− 受け取ったイメージをPIL(RGB)へ変換
if image: #inpaint input イメージ
image_data =image.file.read()
image = Image.open(BytesIO(image_data)) # バイナリデータをPIL形式に変換
image = image.convert("RGB")
else:
image = None
if mask_image: #inpaint マスク イメージ
mask_data =mask_image.file.read()
mask_image = Image.open(BytesIO( mask_data)) # バイナリデータをPIL形式に変換
mask_image = mask_image.convert("RGB")
else:
mask_image = None
# −−−−− IP-Adapter用の画像の再現
if ip_adapter:
if ip_image:
ip_data= ip_image.file.read()
ip_image = Image.open(BytesIO(ip_data)) # バイナリデータをPIL形式に変換
ip_image = ip_image.convert("RGB")
#face_image .show()
# −−−−− "t2i_mip"と"i2i_mip"の時、受け取ったStyle IPイメージをstyle_images_listへ復元
if pipeline=="t2i_mip" or pipeline=="i2i_mip":
style_images_rb=style_images.file.read()
style_images_list = pickle.loads(style_images_rb)
else:
style_images_list=None
else:
ip_image =None
style_images_list=None
#−−−−−freeuの設定
if freeu:
#freeu_list=eval(freeu_list[0])
freeu_list=[float(item) for item in freeu_list]
print("++++++++++++++++++++++++++++++++++++++ freeu_list", freeu_list)
crops_coords_top_left=(crops_coords_top_left[0],crops_coords_top_left[1])
#−−−−−こから画像の生成
print("prompt=",prompt)
print("prompt_2=",prompt_2)
# t2i_adapter専用
if pipeline=="t2i_adapter":
start_time = time.time()
image_list = Df.generate_t2a_adapter(
pipe,
image=image,
prompt=prompt,
prompt_2=prompt_2,
negative_prompt = negative_prompt,
negative_prompt_2 = negative_prompt_2,
generator=generator,
height = height,
width = width,
num_images_per_prompt = num_images_per_prompt,
freeu=freeu,
freeu_list=freeu_list,
output_type=output_type,
num_inference_steps= num_inference_steps,
adapter_conditioning_scale=adapter_conditioning_scale,
adapter_conditioning_factor= adapter_conditioning_factor,
guidance_scale=guidance_scale,
)
print("Generation time = ",(time.time() - start_time)*1000,"mS")
# t2i/i2i/inpaint/canny/openpose/depth
else:
start_time = time.time()
image_list = Df.generate(
pipe,
image=image,
prompt=prompt ,
prompt_2=prompt_2,
mask_image=mask_image,
embe=embe,
conditioning=conditioning ,#prompt_embeds=conditioning,
pooled=pooled ,#pooled_prompt_embeds=pooled,
negative_prompt = negative_prompt,
negative_prompt_2 = negative_prompt_2,
num_inference_steps= num_inference_steps,
strength=strength,
guidance_scale=guidance_scale,
generator=generator,
height = height,
width = width,
num_images_per_prompt = num_images_per_prompt,
guess_mode=guess_mode,
freeu=freeu,
freeu_list=freeu_list,
output_type=output_type,
crops_coords_top_left=crops_coords_top_left,
controlnet_conditioning_scale=controlnet_conditioning_scale,
ip_adapter=ip_adapter,
ip_image = ip_image ,
style_images=style_images_list,
)
print("Generation time = ",(time.time() - start_time)*1000,"mS")
#−−−−−生成画像を返信
images_data = pickle.dumps(image_list, 5) # tx_dataはpklデータ
except:
print("Failed to generate image.")
return {"status": "failed"}
return Response(content= images_data, media_type="application/octet-stream")
if __name__ == "__main__":
import uvicorn
#uvicorn.run(app, host="0.0.0.0", port=64656)
uvicorn.run(app, host="192.168.5.73", port=64656)
n_lcm_lora_sdxl_i2i_t2i_class_v1_1_t.py サーバで使うクラス
from diffusers import StableDiffusionXLImg2ImgPipeline , StableDiffusionXLPipeline , AutoPipelineForText2Image, AutoPipelineForImage2Image ,StableDiffusionDepth2ImgPipeline, StableDiffusionXLInpaintPipeline
from diffusers import DiffusionPipeline, LCMScheduler, UNet2DConditionModel , AutoencoderKL , AutoPipelineForInpainting,StableDiffusionXLAdapterPipeline
from diffusers import StableDiffusionXLControlNetPipeline, ControlNetModel, AutoencoderKL, UniPCMultistepScheduler, T2IAdapter, LCMScheduler,DDIMScheduler
from diffusers.utils import load_image,make_image_grid
from compel import Compel,ReturnedEmbeddingsType
from controlnet_aux import OpenposeDetector
#Depthで使用
from transformers import DPTFeatureExtractor, DPTForDepthEstimation,CLIPVisionModelWithProjection
import torch
from datetime import datetime
import time
import numpy as np
import cv2
from PIL import Image
class DiffusersPipelineManager:
def __init__(self):
#----- デフォルト用定義
#VAE
self.org_vae = "madebyollin/sdxl-vae-fp16-fix"
#model
self.org_model="./models/animagine-xl-2.0/animagine-xl-2.0.safetensors"
#self.org_model="./models/sd_xl_1.0/sd_xl_base_1.0.safetensors",
#self.org_model="./models/blue_pencil-XL-LCM/bluePencilXLLCM_v200.safetensors"
#LoRA
self.org_lora_model_list=[
["./models/animagine-xl-2.0/megu_sports_sdxl_v02.safetensors","style",0.8],
["./models/animagine-xl-2.0/anime-detailer-xl.safetensors","detaile",1.0],
["./models/animagine-xl-2.0/style-enhancer-xl.safetensors","enhancer",0.5]]
#embe TEST プロンプトの準備
self.prompt_embe=["masterpiece, best quality, 1girl, solo, flower, long hair, outdoors, letterboxed, school uniform, day, sky, looking up, short sleeves, parted lips, shirt, cloud,black hair, sunlight, white shirt, serafuku, upper body, from side, pink flower, blurry, brown hair, blue sky, depth of field",
"at shcool park"]
#self.lcm_lora="latent-consistency/lcm-lora-sdxl"
self.lcm_lora="./models/SSD-1B-anime/lcm-ssd1b-anime.safetensors"
def pipeline_init(self,
mode:str="t2i", #指定できるpipe名 t2i/i2i/canny/i2i_pipe/inpaint_pipe/canny/openpose/canny_openpose/depth/t2i_adapter_canny
pipeline=None, #pipeloneの再利用の時に参照する生成済みpipeline名指定 デフォルト：無し
model:str=None, #ロードするSDXLモデル名 デフォルト：無し
vae_en:bool=True, #VAE有効無効設定 デフォルト：有効
vae_path:str=None, #VAE有効時のVAEへのパス デフォルト：無し
lcm_lora_en:bool=True, #LCM-LoRA有効無効設定 無効にするとLCM-LoRAとLCMスケジュ-ラがロードされない デフォルト 有効
lcm_lora_path=None, #LCM-LoRA有効のLCM-LoRAへのパス デフォルト：無し
lora_en:bool=False, #Style-LoRAの有効無効設定 デフォルト 有効
lola_models:list=None, #Style-LoRAの有効時のStyle-LoRAへのパスリスト 複数のStyle-LoRAをロード出来る デフォルト：無し
ddism_en:bool=False, # ddism¥スケジューラ有効化
single_ip:bool=False, # IP-Adapter 有効化
mul_ip:bool=False, # Multi IP-Adapter 有効化
ip_scal1:float=None, # IP-AdapterのScale/Multi IP-Adapter のScale1
ip_scal2:float=None, # Multi IP-Adapter のScale2
set_unet:bool=False, #UNETの有効無効設定 デフォルト 有効
torch_compile:bool=False,
):
# デフォルト引数が None の場合、クラス属性を使用
if model is None:
model = self.org_model
if vae_path is None:
vae_path = self.org_vae
if lcm_lora_path is None:
lcm_lora_path = self.lcm_lora
if lola_models is None:
lola_models = self.org_lora_model_list
#UNETの準備
if lcm_lora_en:
unet = UNet2DConditionModel.from_pretrained(
"latent-consistency/lcm-sdxl",
torch_dtype=torch.float16,
variant="fp16",)
#VAEの準備
if vae_en:
vae = AutoencoderKL.from_pretrained(vae_path, torch_dtype=torch.float16, use_safetensors=True)
else:
print("VAE disabled")
vae = None
#pipelineの準備
if mode=="t2i":
if lcm_lora_en:
print("+++++t2i lcm_lora_en=True")
if model.split("/")[1]=="models": #V1.2 ./models/で始まる場合はHuggingFaceではなくfrom_single_fileでロード
try:
pipe = StableDiffusionXLPipeline.from_single_file(model,safety_checker=None,feature_extractor=None,unet=unet,torch_dtype=torch.float16,vae=vae, variant="fp16",).to("cuda")
print("Use singlefile & bf16")
except:
print("Use singlefile & fp32")
pipe = StableDiffusionXLPipeline.from_single_file(model,safety_checker=None,feature_extractor=None,unet=unet,torch_dtype=torch.float16,vae=vae,).to("cuda")
else:
try: #1.2 ./models/で始まらない場合はHuggingFaceからダウンロードする
pipe = StableDiffusionXLPipeline.from_pretrained(model,safety_checker=None,feature_extractor=None,unet=unet,torch_dtype=torch.float16,vae=vae,variant="fp16",).to("cuda")
print("Use hf & bf16")
except:
print("Use hf & fp32")
pipe = StableDiffusionXLPipeline.from_pretrained(model,safety_checker=None,feature_extractor=None,unet=unet,torch_dtype=torch.float16,vae=vae,).to("cuda")
else:
print("+++++t2i lcm_lora_en=False")
if model.split("/")[1]=="models": #V1.2 ./models/で始まる場合はHuggingFaceではなくfrom_single_fileでロード
try:
pipe = StableDiffusionXLPipeline.from_single_file(model,safety_checker=None,feature_extractor=None,torch_dtype=torch.float16,vae=vae,variant="fp16", ).to("cuda")
print("Use hf & bf16")
except:
print("Use hf & fp32")
pipe = StableDiffusionXLPipeline.from_single_file(model,safety_checker=None,feature_extractor=None,torch_dtype=torch.float16,vae=vae,).to("cuda")
else: #1.2 ./models/で始まらない場合はHuggingFaceからダウンロードする
try:
pipe = StableDiffusionXLPipeline.from_pretrained(model,safety_checker=None,feature_extractor=None,torch_dtype=torch.float16,vae=vae,variant="fp16",).to("cuda")
print("Use hf & bf16")
except:
print("Use hf & fp32")
pipe = StableDiffusionXLPipeline.from_pretrained(model,safety_checker=None,feature_extractor=None,torch_dtype=torch.float16,vae=vae,).to("cuda")
#"i2i_pipe","inpaint_pipe"は"t2i"の設定を引き継ぐので再設定はできない。設定を変える場合は"i2i"を指定すること。。
elif mode=="i2i_pipe":
print("+++++i2i pipe loading")
pipe = AutoPipelineForImage2Image.from_pipe(pipeline).to("cuda")
return pipe
elif mode=="inpaint_pipe":
print("+++++inpaint_pipe loading")
pipe = AutoPipelineForInpainting.from_pipe(pipeline).to("cuda")
return pipe
#ControlNet名を指定してpipeを設定するとき。 対象は="canny" , "openpose" , "depth"の3種類
elif mode=="canny" or mode=="openpose" or mode=="depth":
if mode=="canny" :
controlnet = ControlNetModel.from_pretrained("diffusers/controlnet-canny-sdxl-1.0", torch_dtype=torch.float16, use_safetensors=True )
elif mode=="openpose":
controlnet = ControlNetModel.from_pretrained("thibaud/controlnet-openpose-sdxl-1.0", torch_dtype=torch.float16)
elif mode=="depth":
controlnet = ControlNetModel.from_pretrained("diffusers/controlnet-depth-sdxl-1.0", variant="fp16", use_safetensors=True, torch_dtype=torch.float16,).to("cuda")
else:
print("ContorolNET error")
#pipe = StableDiffusionXLControlNetPipeline.from_pretrained( #HuggingFaceからダウンロードする場合
pipe = StableDiffusionXLControlNetPipeline.from_single_file(
model,
controlnet=controlnet,
vae=vae,
torch_dtype=torch.float16,
use_safetensors=True )
pipe.enable_model_cpu_offload()
#canny_openpose 指定 マルチControlNeのpipeを設定するとき。
elif mode=="canny_openpose":
print("+++++canny_openpose loading")
controlnets = [
ControlNetModel.from_pretrained("diffusers/controlnet-canny-sdxl-1.0", torch_dtype=torch.float16, use_safetensors=True ),
ControlNetModel.from_pretrained( "thibaud/controlnet-openpose-sdxl-1.0", torch_dtype=torch.float16 ),
]
#pipe = StableDiffusionXLControlNetPipeline.from_pretrained( #HuggingFaceからダウンロードする場合
pipe = StableDiffusionXLControlNetPipeline.from_single_file(
model,
controlnet=controlnets,
vae=vae,
torch_dtype=torch.float16,
use_safetensors=True)
pipe.scheduler = UniPCMultistepScheduler.from_config(pipe.scheduler.config)
pipe.enable_model_cpu_offload()
#T2I-Adapterのpipeを作成する時
elif mode=="t2i_adapter_canny":
print("+++++t2i_adapter_canny loading")
adapter = T2IAdapter.from_pretrained("TencentARC/t2i-adapter-canny-sdxl-1.0", torch_dtype=torch.float16, varient="fp16").to("cuda")
unet = UNet2DConditionModel.from_pretrained("latent-consistency/lcm-sdxl", torch_dtype=torch.float16, variant="fp16",)
pipe = StableDiffusionXLAdapterPipeline.from_pretrained( #HuggingFaceからダウンロードする場合
model,
unet=unet,
adapter=adapter,
vae=vae,
torch_dtype=torch.float16,
variant="fp16",
).to("cuda")
# IP-Adapterのpipeを設定
elif mul_ip:
print("-----mul_ip",mode)
image_encoder = CLIPVisionModelWithProjection.from_pretrained(
"h94/IP-Adapter",
subfolder="models/image_encoder",
torch_dtype=torch.float16,
)
if mode=="t2i_ip":
pipe = StableDiffusionXLPipeline.from_single_file(model, torch_dtype=torch.float16, image_encoder=image_encoder).to("cuda")
if mode=="i2i_ip":
pipe = StableDiffusionXLImg2ImgPipeline.from_single_file(model, torch_dtype=torch.float16, image_encoder=image_encoder).to("cuda")
"""
#当面 IP-Adapter inpaintは使えません
if mode=="inpaint_ip":
pipe = StableDiffusionXLInpaintPipeline.from_single_file(model, torch_dtype=torch.float16, image_encoder=image_encoder).to("cuda")
"""
print("+++++ multi Ip-Adapter loading")
pipe.load_ip_adapter("h94/IP-Adapter", subfolder="sdxl_models", weight_name=["ip-adapter-plus_sdxl_vit-h.safetensors", "ip-adapter-plus-face_sdxl_vit-h.safetensors"])
pipe.set_ip_adapter_scale([ip_scal1, ip_scal2])
elif single_ip:
print("-----single_ip",mode)
'''
if mode=="t2i_ip":
print("+++++t2i_ip loading")
pipe = StableDiffusionXLPipeline.from_single_file(model, torch_dtype=torch.float16 ).to("cuda")
if mode=="i2i_ip":
print("+++++ i2i_ip loading")
pipe = StableDiffusionXLImg2ImgPipeline.from_single_file(model, torch_dtype=torch.float16).to("cuda")
#IP-Adapter inpaintはVRAMが16G必要
if mode=="inpaint_ip":
print("+++++ inpaint_ip")
pipe = StableDiffusionXLInpaintPipeline.from_single_file(model, torch_dtype=torch.float16).to("cuda")
'''
print("+++++ single-Ip-Adapter loading")
pipe.load_ip_adapter("h94/IP-Adapter", subfolder="sdxl_models", weight_name="ip-adapter_sdxl.bin")
#pipe.load_ip_adapter("h94/IP-Adapter", subfolder="sdxl_models", weight_name="ip-adapter-plus-face_sdxl_vit-h.safetensors")#Need DDIMScheduler and EulerDiscreteScheduler
pipe.set_ip_adapter_scale(ip_scal1)
else:
print("error=100") #pipe名指定のエラー
return "100"
# Set scheduler LCMスケージューラーのセット
if lcm_lora_en: # Reload the scheduler with custom config
pipe.scheduler = LCMScheduler.from_config(pipe.scheduler.config)# Reload the scheduler with custom config
if ddism_en: # Reload the scheduler with custom config
pipe.scheduler = DDIMScheduler.from_config(pipe.scheduler.config)
#LCM-LoRAのロード
if lcm_lora_en:
#pipe.load_lora_weights("latent-consistency/lcm-lora-sdxl", adapter_name="lcm") #HuggingFaceからダウンロードする場合
pipe.load_lora_weights(lcm_lora_path, adapter_name="lcm")
adapters_list=["lcm"]
else:
adapters_list=[]
#Style-LoRAのロード
if lora_en :
n_of_lora=len( lola_models)
if lcm_lora_en:
adapter_weights_list=[1.0]
else:
adapter_weights_list=[]
#loRAのモデルとアダプタ名及びウエイトリスト作成
for n in range(n_of_lora):
lora_name=lola_models[n][0]
adapter_name=lola_models[n][1]
adapters_list.append(adapter_name)
weight=lola_models[n][2]
adapter_weights_list.append(weight)
pipe.load_lora_weights(lora_name, adapter_name=adapter_name)
# Combine LoRAs
pipe.set_adapters(adapters_list, adapter_weights=adapter_weights_list)
print("adapters_list",adapters_list,"adapter_weights",adapter_weights_list)
# Set the unet to channels last, which accelerates inference a little bit
if set_unet:
pipe.unet = pipe.unet.to(memory_format=torch.channels_last)
if torch_compile:
print("torch.compile")
pipe.unet = torch.compile(pipe.unet, mode="reduce-overhead", fullgraph=True)
#Disable the progress bar
#pipe.set_progress_bar_config(disable=True)
return pipe
#-----prompt_embeddings プロンプトをロードするクラス
#prompt = "a red cat++ playing with a ball----" prompt Enhace / Reduce
def prompt_embedding(self, pipe, prompt):
#compel_proc = Compel(tokenizer=pipe.tokenizer, text_encoder=pipe.text_encoder)
compel = Compel(
tokenizer=[pipe.tokenizer, pipe.tokenizer_2] ,
text_encoder=[pipe.text_encoder, pipe.text_encoder_2],
returned_embeddings_type=ReturnedEmbeddingsType.PENULTIMATE_HIDDEN_STATES_NON_NORMALIZED,
requires_pooled=[False, True]
)
conditioning, pooled = compel(prompt)
return conditioning, pooled
#-----load_image for i2i 指定イメージをロードするクラス
def load_image_img(self,image_path):
image =load_image(image_path)
#image = pipe.image_processor.preprocess(
# Image.open(image_path) .convert("RGB").resize((512,) * 2, Image.Resampling.LANCZOS))
#image = Image.open(image_path)#同じ
return image
#cannyイメージ作成クラス
def get_canny(self , original_image , low_threshold = 100 , high_threshold = 200):
image = np.array(original_image)
image = cv2.Canny(image, low_threshold, high_threshold)
image = image[:, :, None]
image = np.concatenate([image, image, image], axis=2)
canny_image = Image.fromarray(image)
return canny_image
#openposeイメージ作成クラス
def get_openpose(self , image ):
openpose = OpenposeDetector.from_pretrained("lllyasviel/ControlNet")
openpose_image = openpose(image)
return openpose_image
#Depthイメージ作成クラス
def get_depth_map(self, image):
feature_extractor = DPTFeatureExtractor.from_pretrained("Intel/dpt-hybrid-midas")
depth_estimator = DPTForDepthEstimation.from_pretrained("Intel/dpt-hybrid-midas").to("cuda")
image = feature_extractor(images=image, return_tensors="pt").pixel_values.to("cuda")
with torch.no_grad(), torch.autocast("cuda"):
depth_map = depth_estimator(image).predicted_depth
depth_map = torch.nn.functional.interpolate(depth_map.unsqueeze(1), size=(1024, 1024), mode="bicubic", align_corners=False, )
depth_min = torch.amin(depth_map, dim=[1, 2, 3], keepdim=True)
depth_max = torch.amax(depth_map, dim=[1, 2, 3], keepdim=True)
depth_map = (depth_map - depth_min) / (depth_max - depth_min)
image = torch.cat([depth_map] * 3, dim=1)
image = image.permute(0, 2, 3, 1).cpu().numpy()[0]
image = Image.fromarray((image * 255.0).clip(0, 255).astype(np.uint8))
return image
#画像の生成
#generate t2i/i2i/inpaint/cannyで使用
#generate_controlnet openpose/depth/Multi-ContorolNETで使用
#generate_t2a_adapter T2I- Adapterで使用
"""3種類の生成モード共通のパラメータ
pipe
利用するpipeを指定。必ず指定しなければならない。
image=None,
入力イメージを指定する。パスではなくイメージオブジェクト（PILが標準、CV2も使えるはず（ためしていない）
mask_image=None,
インペイントなどのマスクイメージを指定。パスではなくイメージオブジェクト（PILが標準、CV2も使えるはず（ためしていない）
embe:bool=True,
プロンプトエンべデッディング有効・無効。有効時はCONPLEによるエンべディングを行う。プロンプトのウエイト指定も可能
ウエイト指定の例 prompt = "a red cat++ playing with a ball----" +はprompt Enhace /-は Reduce
prompt:str=None,
プロンプト 文章も使えます。ただし英語のみ
conditioning=None,
pooled=None,
conditioningとpooledはプロンプトエンべデッディングでCONPLEから得られる変数を記述
negative_prompt=None,
ネガティブプロンプト
num_inference_steps:int=4,
生成時のSTEP、大きいほうがきれいになる。lmcは4が標準だが適時上げる
strength:float=0.1,
プロンプトの効きを調整する0〜1.0
guidance_scale:float=1.0,
ガイダンススケール 1.0で無効、大きいと生成時間が伸びる。上げれば画像がきれいになるわけででない
generator=None,
SEEDを指定する。例 generator = torch.manual_seed(0 )でマニュアルでSEED指定
height:int = None,
画像の高さを指定する
width:int = None,
画像の幅を指定する
num_images_per_prompt:int = 1,
プロンプトあたりの生成画像数を指定して複数生成させることが出来る
guess_mode:bool=None,
ガウシアン効果の有効無効指定
freeu:bool=False,
freeu効果の有効無効指定。有効にすると生成画像が鮮やかな色になる。
output_type:str="pil",
画像出力形式を指定する。ndarreyが指定可能（"nd")
crops_coords_top_left=(0, 0),
生成する主画像の位置を指定する
controlnet_conditioning_scale=1.0,
コントロールネットの効き具合を指定する
ip_adapter:bool=False,
IP-Adaperを使う
style_images:list=None,
Multi-IP-Adaper用のStyle画像リスト
face_image=None,
IP-Adaper用の入力画像 / Multi-IP-Adaper用のface画像リスト
T2I-Adaper特有のパラメータ
adapter_conditioning_scale:float=None,
adapter_conditioning_factor:int=None,
アダプタの効き具合の設定をする
"""
#-----image generation t2i/i2i /inpaint and Multi-IP-Adapter t2i
def generate(self,
pipe,
image=None,
mask_image=None,
negative_prompt_embeds= None,
embe:bool=True,
prompt:str=None,
prompt_2:str=None,
conditioning=None,
pooled=None,
negative_prompt=None,
negative_prompt_2=None,
num_inference_steps:int=4,
strength:float=0.1,
guidance_scale:float=1.0,
generator=None,
height:int = None,
width:int = None,
num_images_per_prompt:int = 1,
guess_mode:bool=None,
freeu:bool=False,
freeu_list:list=[0.9, 0.2,1.2,1.4],
output_type:str="pil",
crops_coords_top_left=(0, 0),
controlnet_conditioning_scale=1.0,
ip_adapter:bool=False,
style_images:list=None,
ip_image =None,
):
print("++++++++++++++++ @@@@@@@@@@@@ +++++++++generate")
if freeu: # Freeu enable
#pipe.enable_freeu(s1=0.9, s2=0.2, b1=1.2, b2=1.4)
pipe.enable_freeu(s1=freeu_list[0],s2=freeu_list[1],b1=freeu_list[2],b2=freeu_list[3])
if ip_adapter:
if style_images==None:
print("Single_IP-Adapter")
ip_adapter_image=ip_image
else:
print("Multi_IP-Adapter")
ip_adapter_image=[style_images, ip_image ]
image = pipe(
image=image,
prompt=prompt,
mask_image=mask_image,
prompt_embeds=None,
pooled_prompt_embeds=None,
negative_prompt=negative_prompt,
num_inference_steps=num_inference_steps,
generator=generator,
height=height,
width= width,
guidance_scale=guidance_scale,
#strength=strength, # ip_adapterではエラーになります。ip_adapter==False:のときとの違いはここだけ
num_images_per_prompt=num_images_per_prompt,
guess_mode=guess_mode,
crops_coords_top_left=crops_coords_top_left,
controlnet_conditioning_scale=controlnet_conditioning_scale,
ip_adapter_image= ip_adapter_image,
style_images=style_images,
).images
elif ip_adapter==False:
print("++++++++++++++@@@@@@@@@@@@@@ embe=", embe)
image = pipe(
image=image,
prompt=prompt, # prompt is passed to OAI CLIP-ViT/L-14
prompt_2=prompt_2, # prompt_2 is passed to OpenCLIP-ViT/bigG-14
mask_image=mask_image,
prompt_embeds=conditioning,
pooled_prompt_embeds=pooled,
negative_prompt=negative_prompt,
negative_prompt_2=negative_prompt_2,
num_inference_steps=num_inference_steps,
strength=strength,
generator=generator,
height =height,
width = width,
num_images_per_prompt=num_images_per_prompt,
guess_mode=guess_mode,
output_type=output_type,
guidance_scale=guidance_scale,
crops_coords_top_left=crops_coords_top_left,
controlnet_conditioning_scale=controlnet_conditioning_scale,
).images
return image
#-----image generation controlnet
def generate_controlnet(self,
pipe,
image=None,
prompt:str=None,
negative_prompt=None,
num_inference_steps:int=4,
guidance_scale:float=1.0,
generator=None,
height:int = None,
width:int = None,
num_images_per_prompt:int = 1,
guess_mode:bool=None,
freeu:bool=False,
output_type:str="pil",
crops_coords_top_left=(0, 0),
controlnet_conditioning_scale=1.0,
):
if freeu: # Freeu enable
pipe.enable_freeu(s1=0.9, s2=0.2, b1=1.2, b2=1.4)
image = pipe(
prompt=prompt,
image=image,
negative_prompt=negative_prompt,
num_inference_steps=num_inference_steps,
generator=generator,
height =height,
width = width,
num_images_per_prompt=num_images_per_prompt,
guess_mode=guess_mode,
guidance_scale=guidance_scale,
crops_coords_top_left=crops_coords_top_left,
controlnet_conditioning_scale=controlnet_conditioning_scale,
).images
return image
#-----image generation t2a_adapter
def generate_t2a_adapter(self,
pipe,
image=None,
prompt:str=None,
negative_prompt=None,
num_inference_steps:int=4,
guidance_scale:float=1.0,
generator=None,
height:int = None,
width:int = None,
num_images_per_prompt:int = 1,
freeu:bool=False,
output_type:str="pil",
crops_coords_top_left=(0, 0),
adapter_conditioning_scale:float=None,
adapter_conditioning_factor:int=None,
):
if freeu: # Freeu enable
pipe.enable_freeu(s1=0.9, s2=0.2, b1=1.2, b2=1.4)
image = pipe(
prompt=prompt,
image=image,
negative_prompt=negative_prompt,
num_inference_steps=num_inference_steps,
generator=generator,
height =height,
width = width,
num_images_per_prompt=num_images_per_prompt,
guidance_scale=guidance_scale,
crops_coords_top_left=crops_coords_top_left,
adapter_conditioning_scale=adapter_conditioning_scale,
adapter_conditioning_factor=adapter_conditioning_factor,
).images
return image
