Diffusersで画像サーバ構築 V1.1
Diffusersが0.26.0にバージョンアップし、注目のIP-Adapetrにマルチアダプタ対応が導入されたので、自作DiffusersPipleineManagerクラスとFastAPIによるAPIサーバもv1.1 へアップデートしました。同じモデル、同じパラメータで生成される画風が0.26.0では変わってしまってしまいましたが、マルチIP-Adapetrが使えるので、環境を使い分けても良さそうです。
環境
Python=3.10
CUDA=12.2
Torch=2.22
transformers
diffusers 0.25.0 and/or 0.26.0
compel
numpy
cv2
PIL
GPU 中程度の性能が必要
DiffusersPipleineManagerクラスv1.1
DiffusersPipleineManagerをV1.1へ機能upしました。
１）マルチIP-Adaper機能
t2i/i2iに対してIP-Adaper及びマルチIP-Adaperを適用
２）IP-Adaper機能にinpaintを追加
３）Freeu有効時にパラメータを渡す機能
デフォルト：s1=0.9, s2=0.2, b1=1.2, b2=1.4
s1、s2、b1、b2をクライアント側から指定できます。
４）SEEDの指定方法変更
seed無し→ auto_seed有効→seedx自動生成 torch.manual_seed(seedx )
auto_seed無効→ generator指定なし(pipelineがseed生成)
seed有り→ generator = torch.manual_seed(seed )
auto_seedは無視される
マルチIP-Adapter
0.26.0で導入された主要機能です。複数のIP-Adapterを用いて10枚程度の画像とfaceイメージでLoRAのような効果が期待できます。
refarenceイメージ 修正をかけたい元画像
faceイメージ refarenceイメージの一部に反映させたい画像（顔など）
stypeイメージ 画風（雰囲気）を定義する10枚程度の画像
SDXL版のウエイトはまだ少数です。以下参照ください。
サーバのコード
デフォルトlola_modelの定義
pipelineにbindされるのでクライアント側では指定できません。サーバ側で事前にbindしたいLoRAを指定します。List形式でモデル名とscaleを設定
lola_models=[
["./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]]モデルの指定
モデルも事前にサーバで指定します。DiffusersPipleineManagerはSDXL専用なので注意してください。safetenser形式のchkekpointは事前にダンロードする必要が有ります。pipline作成のコードで.from_single_fileをfrom_pretrainedに修正するとHugingFaceのモデルは自動ダウンロド出来ます。HugingFaceに無いモデルは事前にダンロードが必要です。
#----- model定義
model="./models/animagine-xl-2.0/animagine-xl-2.0.safetensors"
model_t2i="stabilityai/stable-diffusion-xl-base-1.0"モデルのロード
作成したい生成方法ごとにpipelineへのロードが必要です。t2iを基本に、i2iとinpaintはt2iで生成したpipelineを引き継いで指定できます。VRAM使用量は増えません。以下、t2i/i2i/inpaintをデフォルトロードし、他のpipelineは必要に応じてロードします。16G-VRAMですと1種類しかロードできません。
#----- pipelineのロード "t2i"・"i2i_pipe"・"inpaint_pipe"はデフォルトでロードでOK、他はVRAMの容量次第
#pipe_t2i=Df.pipeline_init("t2i",model=model, lola_models=lola_models)
#pipe_i2i=Df.pipeline_init("i2i_pipe",pipeline=pipe_t2i)#pipelineの使い回し
#pipe_inpaint=Df.pipeline_init("inpaint_pipe",pipeline=pipe_t2i)#pipelineの使い回し
#Control Net pipelineの設定
#pipe_canny=Df.pipeline_init("canny",model=model) #Canny ContorolNET用pipeの新規定義
#pipe_openpose=Df.pipeline_init("openpose",model=model) #openpose ContorolNET用pipeの新規定義
#pipe_openpose=Df.pipeline_init("openpose",model=model, lola_models=lola_models) #lola_models指定
#pipe_depth = Df.pipeline_init("depth",model=model)
#pipe_t2i_adapter=Df.pipeline_init("t2i_adapter_canny",model=model_t2i, lora_en=False)#t2i_adapter用
#IP-Adapter pipeline 設定
#mode = t2i_ip/i2i_ip/inpaint_ip
#mul_ip=True or single_ip=True
#pipe_t2i_mip=Df.pipeline_init("t2i_ip",model=model, lcm_lora_en=True, mul_ip=True, ip_scal1=0.9,ip_scal2=0.5)
#pipe_i2i_mip=Df.pipeline_init("i2i_ip",model=model, lcm_lora_en=True, mul_ip=True, ip_scal1=0.9,ip_scal2=0.5)
#pipe_t2i_sip=Df.pipeline_init("t2i_ip",model=model, lcm_lora_en=True, single_ip=True,ip_scal1=0.6)
#pipe_i2i_sip=Df.pipeline_init("i2i_ip",model=model, lcm_lora_en=True, single_ip=True,ip_scal1=0.6)
pipe_inpaint_sip=Df.pipeline_init("inpaint_ip",model=model, lcm_lora_en=True, single_ip=True,ip_scal1=0.6)pipelineの指定
サーバ-クライント間ではpipelineをやり取りできません。クライントからは使いたいpipelineを指定して、サーバ側で実施のpipelineを選択します。
pipline_sel(pipeline)ではクライアントから受け取ったpipeline名から利用する生成に利用するpipelineを選んでいます。いずれにも該当がなければ自動的にt2iが選択されます。
#--------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_t2iControleNETで使用する画像変換
クライアントからは以下の3種類
get_canny
get_openpose
get_depth_map
を呼び出すことが出来ます。
画像の生成
クライアントからパラメータや画像データを受け取りDiffusersPipleineManagerのgenerateクラスへ渡して画像を生成します。
@app.post("/generate/")エンドポイントでは前準備として
１）pipelineの選択
２）プロンプトエンべデッディングの処理（デフォルトは無し）
３）SEEDの生成
４）イメージデータの復元
i2iのレファレンス
inpaintのレファレンスとマスク
IP-AdapterのレファレンスとStyle画像群
５）Freeuの設定 デフォルトで標準値が設定されます
６）crops_coords_top_leftの設定
を実行し、pipelineへ渡して画像を生成します。
t2i_adapterとそれ以外の２種類を使わけています。
生成パラメータ
各パラメータの意味はDiffusersPipleineManagerの記事を参照してください
クライアントから指定できる変数
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), negative_prompt: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 =Form(None), style_images:UploadFile =File(None), ip_image:UploadFile =File(None),
実際にpipelineへ渡される変数
pipe, image=image, prompt=prompt , mask_image=mask_image, embe=embe, conditioning=conditioning ,#prompt_embeds=conditioning, pooled=pooled ,#pooled_prompt_embeds=pooled, negative_prompt = negative_prompt, 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=True, 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,
生成画像の返送
DiffusersPipleineManagerで生成された画像はクラインアントへ返送されます。media_type="application/octet-stream"で通信オーバヘッドを軽減しています。
#−−−−−生成画像を返信
images_data = pickle.dumps(image_list, 5) # tx_dataはpklデータ
return Response(content= images_data, media_type="application/octet-stream")基本テスト
i2i/t2i/inpaintのテスト
以下でテストを選択します。複数選択も可能です。
t2i=True
i2i=False
inpaint=False
import requests
from PIL import Image
from io import BytesIO
import pickle
import time
t2i=True
i2i=False
inpaint=False
url = 'http://0.0.0.0:8000/generate/'
# API サーバrequest モジュール
def resuest_imag(data,files=None):
response = requests.post(url, data=data ,files=files) # POSTリクエストを送信
print(response)
# レスポンスを表示
if response.status_code == 200:
print("get data OK")
image_data = response.content
image =(pickle.loads(image_data))#元の形式にpickle.loadsで復元
else:
image=[]
print("リクエストが失敗しました。ステータスコード:", response.status_code)
return image
# t2iのテスト pipe_t2i のロードが必要
if t2i:
count=5 #生成回数
for i in range(count):
start_time = time.time()
data= {
"pipeline":"t2i",
"prompt":"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",
"negative_prompt":"monochrome, lowres, bad anatomy, worst quality, low quality",
"freeu":True,
"freeu_list":[0.9, 0.2,1.2,1.4],
"num_inference_steps":4,
"guidance_scale":1.0,
#"auto_seed":False,
"seed":i,
"height": 512,
"width": 512,
}
image = resuest_imag(data)
print("Generation time = ",(time.time() - start_time)*1000,"mS")
image[0].show()
# t2iのテスト pipe_t2i のロードが必要
if i2i:
count=5 #生成回数
for i in range(count):
start_time = time.time()
image_file_path="MIKU.png"
file_data = open(image_file_path, "rb").read()
files={"image": ("img.png", BytesIO(file_data), "image/png"), }
data= {
"pipeline":"pipe_i2i",
"prompt":"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",
"negative_prompt":"monochrome, lowres, bad anatomy, worst quality, low quality",
"freeu":True,
"freeu_list":[0.9, 0.2,1.2,1.4],
"num_inference_steps":8,
"guidance_scale":1.0,
#"auto_seed":False,
"strength":0.8, #小さすぎるとエラーになります
"seed":i,
}
image = resuest_imag(data,files)
print("Generation time = ",(time.time() - start_time)*1000,"mS")
image[0].show()
# inpaintのテスト pipe_t2i のロードが必要
if inpaint:
count=5 #生成回数
for i in range(count):
start_time = time.time()
image_file_path="image_lcm_lora_3.png"
file_data = open(image_file_path, "rb").read()
mask_image_file_path="mask_body.jpg"
mask_file_data = open(mask_image_file_path, "rb").read()
files={"image": ("img.png", BytesIO(file_data), "image/png"),
"mask_image":("img.png", BytesIO(mask_file_data), "image/png"),}
data= {
"pipeline":"pipe_inpaint",
"prompt": "masterpiece, best quality, 1girl, solo, long hair, red t-shirt, simple background,white background",
"freeu":True,
"freeu_list":[0.9, 0.2,1.2,1.4],
"num_inference_steps":8,
"guidance_scale":1.0,
"height": 512,
"width": 512,
#"auto_seed":False,
"strength":0.95, #大きくないとプロンプトが反映されにくい
"seed":i,
}
image = resuest_imag(data,files)
print("Generation time = ",(time.time() - start_time)*1000,"mS")
image[0].show()
IP- Adapterテスト
シングルのIP- AdapterテストとマルチIP- Adapterテストがi2i/t2i/inpaintに対して実行できます。以下から選択しますが、各IP- Adapterテストを実行するためにはサーバ側で対応するpiupelineを定義して置く必要があります。
t2i_sip=False
t2i_mip=False
i2i_sip=False
i2i_mip=False
inpaint_sip=True
import requests
from PIL import Image
from io import BytesIO
import pickle
t2i_sip=False
t2i_mip=False
i2i_sip=False
i2i_mip=False
inpaint_sip=True
url = 'http://0.0.0.0:8000/generate/'
# API サーバrequest モジュール
def resuest_imag(data,files=None):
response = requests.post(url, data=data ,files=files) # POSTリクエストを送信
print(response)
# レスポンスを表示
if response.status_code == 200:
print("get data OK")
image_data = response.content
image =(pickle.loads(image_data))#元の形式にpickle.loadsで復元
else:
image=[]
print("リクエストが失敗しました。ステータスコード:", response.status_code)
return image
# t2iのIP-Adapterのテスト pipe_t2i_sip のロードが必要
if t2i_sip:
count=5 #生成回数
for i in range(count):
image_file_path="img_gen0iPkha .jpg"
face_image= Image.open(image_file_path)
face_image .show()
file_data = open(image_file_path, "rb").read()
files={"ip_image": ("img.png", BytesIO(file_data), "image/png"),}
data= {
"pipeline":"t2i_sip",
"prompt":"masterpiece, best quality, 1girl, solo, flower, long hair, outdoors, letterboxed",
"negative_prompt":"monochrome, lowres, bad anatomy, worst quality, low quality",
"freeu":True ,
"freeu_list":[0.9, 0.2,1.2,1.4],
"height":768,
"width": 768,
"num_inference_steps":8,
"guidance_scale":1.0,
"embe":False,
"ip_adapter":True,
"seed":i,
}
image = resuest_imag(data,files)
image[0].show()
# t2iのマルチIP-Adapterのテスト pipe_t2i_mip のロードが必要
if t2i_mip:
image_file_path="img_gen5PlAk9.png"
ip_image= Image.open(image_file_path)
ip_image .show()
file_data = open(image_file_path, "rb").read()
style_folder = "./style_images"
image_list = [Image.open(f"{style_folder}/img{i}.png") for i in range(10)]
files = {"ip_image": ("img.png", BytesIO(file_data), "image/png"),
"style_images":("data.dat",BytesIO(pickle.dumps(image_list )),"data/data"),# image_list をpikleでバイト化し、ファイルデータとして送る
}
count=1 #生成回数
for i in range(count):
data={
"pipeline":"t2i_mip",
"prompt":"masterpiece, best quality, 1girl, solo,looking at viewer,blush,smile ,bangs,blue eyes",
"negative_prompt":"monochrome, lowres, bad anatomy, worst quality, low quality",
"freeu":True ,
"freeu_list":[0.9, 0.2,1.2,1.4],
"num_inference_steps":4, #lcm_lora_en=Falseのときは50
"guidance_scale":1.0, #lcm_lora_en=Falseのときは6.0
"embe":False,
"ip_adapter":True,
"ip_image": ip_image, #face_image "height": 512,
"width": 512,
}
image = resuest_imag(data,files)
image[0].show()
# i2iのIP-Adapterのテスト pipe_t2i_sip のロードが必要
if i2i_sip:
count=5 #生成回数
for i in range(count):
image_file_path="img_gen0iPkha .jpg"
face_image= Image.open(image_file_path)
face_image .show()
ip_file_data = open(image_file_path, "rb").read()
image_file_path="MIKU.png"
file_data = open(image_file_path, "rb").read()
files={"ip_image": ("img.png", BytesIO(ip_file_data), "image/png"),
"image": ("img.png", BytesIO(file_data), "image/png"),
}
data= {
"pipeline":"i2i_sip",
"prompt":"masterpiece, best quality, 1girl, solo, flower, long hair, outdoors, letterboxed",
"negative_prompt":"monochrome, lowres, bad anatomy, worst quality, low quality",
"freeu":True ,
"freeu_list":[0.9, 0.2,1.2,1.4],
"height":768,
"width": 768,
"num_inference_steps":8,
"guidance_scale":1.0,
"embe":False,
"ip_adapter":True,
"seed":i,
}
image = resuest_imag(data,files)
image[0].show()
# i2iのマルチIP-Adapterのテスト pipe_i2i_mip のロードが必要
if i2i_mip:
count=5 #生成回数
for i in range(count):
image_file_path="img_gen0iPkha .jpg"
face_image= Image.open(image_file_path)
face_image .show()
ip_file_data = open(image_file_path, "rb").read()
image_file_path="MIKU.png"
file_data = open(image_file_path, "rb").read()
style_folder = "./style_images"
image_list = [Image.open(f"{style_folder}/img{i}.png") for i in range(10)]
files={"ip_image": ("img.png", BytesIO(ip_file_data), "image/png"),
"image": ("img.png", BytesIO(file_data), "image/png"),
"style_images":("data.dat",BytesIO(pickle.dumps(image_list )),"data/data"),# image_list をpikleでバイト化し、ファイルデータとして送る
}
data= {
"pipeline":"i2i_mip",
"prompt":"masterpiece, best quality, 1girl, solo, flower, long hair, outdoors, letterboxed",
"negative_prompt":"monochrome, lowres, bad anatomy, worst quality, low quality",
"freeu":True ,
"freeu_list":[0.9, 0.2,1.2,1.4],
"height":768,
"width": 768,
"num_inference_steps":8,
"guidance_scale":1.0,
"embe":False,
"ip_adapter":True,
"seed":i,
}
image = resuest_imag(data,files)
image[0].show()
# inpaintのIP-Adapterのテスト pipe_t2i_sip のロードが必要
if inpaint_sip:
count=5 #生成回数
for i in range(count):
image_file_path="image_lcm_lora_3.png"
face_image= Image.open(image_file_path)
#face_image .show()
ip_file_data = open(image_file_path, "rb").read()
image_file_path="MIKU.png"
mask_file_path="mask_body.jpg"
mask_img = open(mask_file_path, "rb").read()
file_data = open(image_file_path, "rb").read()
files={"ip_image": ("img.png", BytesIO(ip_file_data), "image/png"),
"image": ("img.png", BytesIO(file_data), "image/png"),
"mask_image": ("mask_image.png", BytesIO(mask_img), "image/png"),
}
data= {
"pipeline":"inpaint_sip",
"prompt":"masterpiece, best quality, 1girl, solo, long hair, red t-shirt,simple background,white background",
"negative_prompt":"monochrome, lowres, bad anatomy, worst quality, low quality",
"freeu":True ,
"freeu_list":[0.9, 0.2,1.2,1.4],
"height":512,
"width": 512,
"num_inference_steps":8,
"guidance_scale":1.0,
"embe":False,
"ip_adapter":True,
"seed":i,
}
image = resuest_imag(data,files)
image[0].show()
ControlNETテスト
V1.0と同じ仕様です。以下の記事を参考にしてください。
生成時間の目安
GPU：A4000（ampaleアーキテクチャ、VRAM16G）中クラスのGPUです
CPU：i5-12400F
SDLXモデル animagine-xl-2.0.safetensors
t2iの時
Step=4,Guidance?scale=1.0（無効）
512x512 0.55秒/image
1024x1024 1.7秒/image
まとめ
SDXL専用の画像生成サーバを構築しました。SDXLでLCM対応なので非常に綺麗な画像が短い時間で生成出来ます。ControleNETやIP-Adapterのpipelineの定義を個別に行わなければならない点は問題でもありますが、基本的なt1i/i2i/inpaitは満足出来る品質と生成速度が得られました。画像生成サーバを動かしてアプリから生成リクエストを送る使い方は完全な並列処理ができて生成待ち時間も有効に他のタスクに活用できるので、アプリの開発に余裕が出るはずです。Diffusersはバージョンアップも早く、次々と機能が追加されます。実際に0.25.0から0.26.0 へは2ヶ月程度でバージョンアップしています。
画風が変わってしまう件は理由が明確ではないですが、時間があれば原因を調べたいと思います。
付録：サーバ全コード
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 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
app = FastAPI()
#------ DiffusersPipelineManagerの定義と初期化
Df=DiffusersPipelineManager()
#------ 独自LoRAの定義
lola_models=[
["./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]]
#----- model定義
model="./models/animagine-xl-2.0/animagine-xl-2.0.safetensors"
model_t2i="stabilityai/stable-diffusion-xl-base-1.0"
#----- pipelineのロード "t2i"・"i2i_pipe"・"inpaint_pipe"はデフォルトでロードでOK、他はVRAMの容量次第
pipe_t2i=Df.pipeline_init("t2i",model=model, lola_models=lola_models)
pipe_i2i=Df.pipeline_init("i2i_pipe",pipeline=pipe_t2i)#pipelineの使い回し
pipe_inpaint=Df.pipeline_init("inpaint_pipe",pipeline=pipe_t2i)#pipelineの使い回し
#Control Net pipelineの設定
pipe_canny=Df.pipeline_init("canny",model=model) #Canny ContorolNET用pipeの新規定義
#pipe_openpose=Df.pipeline_init("openpose",model=model) #openpose ContorolNET用pipeの新規定義
#pipe_openpose=Df.pipeline_init("openpose",model=model, lola_models=lola_models) #lola_models指定
#pipe_depth = Df.pipeline_init("depth",model=model)
#pipe_t2i_adapter=Df.pipeline_init("t2i_adapter_canny",model=model_t2i, lora_en=False)#t2i_adapter用
#IP-Adapter pipeline 設定
#mode = t2i_ip/i2i_ip/inpaint_ip
#mul_ip=True or single_ip=True
#pipe_t2i_mip=Df.pipeline_init("t2i_ip",model=model, lcm_lora_en=True, mul_ip=True, ip_scal1=0.9,ip_scal2=0.5)
#pipe_i2i_mip=Df.pipeline_init("i2i_ip",model=model, lcm_lora_en=True, mul_ip=True, ip_scal1=0.9,ip_scal2=0.5)
#pipe_t2i_sip=Df.pipeline_init("t2i_ip",model=model, lcm_lora_en=True, single_ip=True,ip_scal1=0.6)
#pipe_i2i_sip=Df.pipeline_init("i2i_ip",model=model, lcm_lora_en=True, single_ip=True,ip_scal1=0.6)
# 利用できる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
else:
return pipe_t2i
#-------------------- FastAPI エンドポイント--------------------------
@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),
negative_prompt: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=Form(None),
style_images:UploadFile =File(None),
ip_image:UploadFile =File(None),
):
#−−−−−pipeの設定
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])
#−−−−−こから画像の生成
# t2i_adapter専用
if pipeline=="t2i_adapter":
image_list = Df.generate_t2a_adapter(
pipe,
image=image,
prompt=prompt ,
negative_prompt = negative_prompt,
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,
)
# t2i/i2i/inpaint/canny/openpose/depth
else:
image_list = Df.generate(
pipe,
image=image,
prompt=prompt ,
mask_image=mask_image,
embe=embe,
conditioning=conditioning ,#prompt_embeds=conditioning,
pooled=pooled ,#pooled_prompt_embeds=pooled,
negative_prompt = negative_prompt,
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=True,
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,
)
#−−−−−生成画像を返信
images_data = pickle.dumps(image_list, 5) # tx_dataはpklデータ
return Response(content= images_data, media_type="application/octet-stream")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)