StreamDiffusionを外部プログラムから利用する（2）FastAPIによるAPIの実装
StreamDiffusionを外部プログラムから利用する（2）FastAPIによるAPIの実装第1回に続き、サーバ編です。FastAPIによるサーバ実装はLLMでも手がけていて、難しくはありません。さらにDiffuserによる画像生成も基本的には画像フォーマットをpillow形式で行えば通信時のバイトデータの変換もpillowのみなので単純化出来ます。一方でStreamDiffusionでは画像生成が極めて高速で行えるため、通信によるオーバーヘッドが性能に大きな影響を受けることがわかりました。今回はasogサーバとしてFastAPIを使用したサーバでStreamDiffusinを動かす例です。通信時の入出力画像ファイルの形式としてpillowとOpenCVが選択出来るように実装しています。なお、次回は3回目としてTCP/IPによる高速通信の例を記事にする予定です。その後の試験で通信フォームをmedia_type="application/octet-stream"とすることでTCP/IPと同等の性能が出せるようになりました。該当箇所は太字で追記します。
POST/GETは遅い
プロトコルがTCP/IP上で動いているので、確実に遅いのですが、データボディーがjesonであることも大きいと思います。一方で、POST/GETはフロントエンドからのアクセスが容易であるために、有って損はないというところでしょうか。その後の試験で通信フォームをmedia_type="application/octet-stream"とすることでTCP/IPと同等の性能が出せるようになりました。
サーバ側コード
全体のコードは記事の最後に記載します。要点を説明します。
Stream作成用のパラメータ
TensorRT
コードの冒頭部分でパラメータの設定が出来ます。重要なパラメータのみを抜き出しています。TensorRTをTrueにセットすると、#--- RCFG の指定と#--- t_index_lisの指定に従って自動的にengine名称（ホルダ名）が作成されます。
model_path = "./models/Model/Counterfeit-V3.0_fix_fp16.safetensors"
lora_path="./models/LoRA/megu_sports_v02.safetensors"
lora_preload=True
lora_preload_weights=0.5
lora_load=False
lora_scale_weights=0.1
tensorrt=True
prompt="masterpiece, best quality, 1girl,"
negative_prompt=""
guidance_scale=1.2
delta=1.0
seed=1
#--- RCFG の指定
cfg_type = "none"
#cfg_type = "full"
#cfg_type = "self"
#cfg_type = "initialize"
#--- t_index_lisの指定
#index_list=[40]
#index_list=[32, 45]
#index_list=[38,40 ,42,45]
#index_list=[20,30,40]
#index_list=[40 ,42,45]
t_index_list=[41,42,44,45] #cam<<<<<<<<<<<<<<<<<<<< i2iのときは有効にすること
#t_index_list=[0, 16, 32, 45] #t2i <<<<<<<<<<<<<<<<<<<< t2iのときは有効にすること
img_size=[512,512]
#TensorRT engineフォルダー名 "engines_tindex"＋t_index_listの要素数＋CFGタイプ
tensorrt_engine="engines_tindex" + str(len(t_index_list))+"_"+ cfg_typeStreamの初期化
#--- モデルのロードから
# ---Warmup
まではStreamを準備するための初期設定です。
#-------------------- 初期設定--------------------------
#--- モデルのロード
pipe = StableDiffusionPipeline.from_single_file(
model_path).to(
device=torch.device("cuda"),
dtype=torch.float16,
)
...
...
...
# ---Warmup >= len(t_index_list) x frame_buffer_size
init_image= Image.new('RGB', img_size, (255, 255, 255))
for _ in range(len(t_index_list)):
stream(init_image)i2iエンドポイント
# ---画像生成 Run the stream infinitely i2i imput=pillow形式 遅い
@app.post("/stream_i2i/")
def stream_i2i(file: UploadFile = File(...), prompt: str = Form(None) , image_type: str = Form(None),out_type: str = Form(None)):画像形式の判断
i2iの場合は入力画像があるので入力画像に合わせた処理が必要です。
image_typで指定された形式の画像データを受け取り、OpenCV形式はpillow形式に変換してStreamへ送ります。
if image_type==None or image_type=="pil": #入力はpillow
image_data = file.file.read()
in_image = Image.open(BytesIO(image_data)) # バイナリデータをPIL形式に変換
elif image_type=="cv2":
contents = file.file.read()
nparr = np.frombuffer(contents, np.uint8)
cv2_image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)# バイナリデータをCv2形式に変換
image = cv2.cvtColor(cv2_image, cv2.COLOR_BGR2RGB)
in_image = Image.fromarray( image)インターラクティブプロンプトの処理
生成リクエストにプロンプトが含まれているとupdate_promptを呼び出して生成毎にプロンプトを変えることが出来ます。
print("+++++++++++ prompt ++++++++++++",prompt)
if prompt != None:
#動的にプロンプトを変える
stream.update_prompt(prompt)画像の生成と後処理
Streamで画像生成を行い、生成されたTensorデータからpillow形式で画像を抽出します。
#---画像生成 i2i
x_output = stream(in_image)
ximage=postprocess_image(x_output, output_type="pil")[0]レスポンスの画像形式の選択
pillowまたはOpenCV形式を選ぶことが出来ます。out_typeに"pil"が指定されたときはそのままバイトデータで返します。指定がないか、"cv2"が指定されれれば、pillow→OpenCV変換し、こちらもバイトデータへ変換して返してイます。
【#---media_type="application/octet-stream"の時---高速通信できる】を追加しています。
# 出力がpillow
if out_type=="pil":
#---media_type="image/png"の時
"""
#pillowイメージをバイト配列に変換
img_byte_arry = io.BytesIO()
ximage.save(img_byte_arry, format='PNG')
img_bytes= img_byte_arry.getvalue()
#---スポンスとしてバイト配列を返す
return Response(content=img_bytes, media_type="image/png")
"""
#---media_type="application/octet-stream"の時---高速通信できる
frame_data = pickle.dumps(cv2_image , 5) # tx_dataはpklデータ
return Response(content=frame_data, media_type="application/octet-stream")
# 出力がOpenVC
else:
#---pillowイメージをOpenVCに変換
new_image = np.array(ximage, dtype=np.uint8)
cv2_image = cv2.cvtColor(new_image, cv2.COLOR_RGB2BGR)
#---media_type="image/jpeg"の時
"""
# Opencv画像をバイナリ形式にエンコードする
_, encoded_img = cv2.imencode('.jpg', cv2_image)
#---スポンスとしてバイト配列を返す
return Response(content=encoded_img.tobytes(), media_type="image/jpeg")
"""
#---media_type="application/octet-stream"の時---高速通信できる
frame_data = pickle.dumps(cv2_image , 5) # tx_dataはpklデータ
return Response(content=frame_data, media_type="application/octet-stream")t2iエンドポイント
t2iは画像入力がないのでシンプルです。冒頭でプロンプトの有無を調べ、Streaｍで画像生成しています。今回は出力指定を入れていませんが、i2i同様の処理を追加すれば、pillowまたはOpenCV形式で返すことが出来ます。
class generate_t2i(BaseModel):
prompt: str=None
@app.post("/stream_t2i/")
def stream_t2i(request:generate_t2i):
prompt=request.prompt
if prompt !=None:
#動的にプロンプトを変える
stream.update_prompt(prompt)
print("prompt=",prompt)
#画像生成 t2i
x_output = stream.txt2img()
ximage=postprocess_image(x_output, output_type="pil")[0]
#---pillowイメージをバイト配列に変換
img_byte_arry = io.BytesIO()
ximage.save(img_byte_arry, format='PNG')
img_bytes= img_byte_arry.getvalue()
#---レスポンスとしてバイト配列を返す
return Response(content=img_bytes, media_type="image/png")t2iテストプログラム
エンドポイントのurlを設定します。
url = 'http://0.0.0.0:8000/stream_t2i/'プロンプトの準備
インターラクティブを試すためにプロンプトをリスト化して順次取り出して画像生成リクエストを送信しています。
#>>>動的プロンプトの初期プロンプト
prompt = "masterpiece, best quality, 1girl,"
prompt_list=[
"1girl","long hair,","white shirt,","serafuku,","brown hair,","looking at viewer,","blush,","smile,", "bangs,","blue eyes,","simple background,", "t-shirt,",\
"white background,","walk a head,","white background,","walk a head,","white background,","walk a head,","white background,","walk a head,","white background,"]
通信部分
プロンプトの回数だけループするようになっています。
for n in range(len(prompt_list)):
start_time=time.time()
prompt=prompt+ prompt_list[n]
# リクエストを送信
print("prompt=",prompt)
data = {"prompt": prompt}
response = requests.post(url, json=data )
# レスポンスが成功であることを確認
if response.status_code == 200:
print("response.status_code== 200")
# レスポンスのコンテンツ（画像データ）をバイナリ形式で取得
image_data = response.content
# イメージをバイト配列に変換
try:
image = Image.open(BytesIO(image_data)) # バイナリデータをPIL形式に変換
#スレッドで表示
th_img = image #生成画像を準備
th_img_flag=True #画像準備フラグセット
except:
print("error")表示部
関数化のときと同様にスレッディングで表示させています。
# -----表示のスレッド化準備（必要に応じて）
global th_img , th_img_flag
#>>> 表示スレッドの定義
import threading
def disp_th():
global th_img , th_img_flag
while True:
if th_img_flag==True:
imgCV_RGB = np.array(th_img, dtype=np.uint8)
th_img = np.array(imgCV_RGB)[:, :, ::-1]
cv2.imshow("image", th_img)
cv2.waitKey(1)
th_img_flag=False
time.sleep(0.01)i2iテストコード
i2iのテストでは過去記事にもあるようなWebCamからの画像入力をStreamDiffusionでアニメ風に変換します。
サーバエンドポイントの設定
url = 'http://0.0.0.0:8000/stream_i2i/'カメラの初期化
capture.pyで画像の取り込みやリフォームを行っています。カメラ入力をそのまま使う場合はコメントアウトした部分を使えば大丈夫です。
#capture.pyを使わないとき
"""
cap = cv2.VideoCapture(0)
# キャプチャがオープンしていることを確認
if not cap.isOpened():
print("カメラを開けません")
status=False
"""
# Webカメラの初期化キャプチャを開始
status=init_cam()
# キャプチャがオープンしていることを確認
if status==False:
print("カメラを開けません")リクエスト送信部
3種類のコードを準備しています。
pilloイメージでリクエスト 返答はpillow
# pilloイメージをバイト配列に変換
img_byte_arry = io.BytesIO()
pil_image.save(img_byte_arry, format='PNG')
img_byte_arry= img_byte_arry.getvalue()
files = {"file": ("img.png", img_byte_arry, "image/png")}
data = {'prompt': "masterpiece, best quality, 1girl",
"image_type":"pil", #pil or cv2 画像形式に合わせる
"out_type":"pil"}
response = requests.post(url, files=files, data=data) #リクエスト送信Opencvイメージでリクエスト 返答はpillow
# Opencv画像をバイナリ形式にエンコードする
_, img_encoded = cv2.imencode('.jpg', cv2_image)
files = {"file": ("img.jpg", img_encoded, "image/jpg")}
data = {"prompt": "masterpiece, best quality, 1girl",
"image_type" :"cv2", #pil or cv2 画像形式に合わせる
"out_type":"pil"}
# リクエストを送信
response = requests.post(url, files=files, data=data) #リクエスト送信Opencvイメージでリクエスト 返答はOpencv
ここでも"application/octet-stream"の例を追記しています。上記2パターンも同様に利用できます。
# Opencvイメージでリクエスト 返答はOpenCV
# Opencv画像をバイナリ形式にエンコードする
_, img_encoded = cv2.imencode('.jpg', cv2_image)
files = {"file": ("img.jpg", img_encoded, "image/jpg")}
#files = {"file": ("img.jpg", img_encoded, "application/octet-stream")}
data = {"prompt": "masterpiece, best quality, 1girl",
"image_type" :"cv2"} #pil or cv2 画像形式に合わせる
# リクエストを送信
response = requests.post(url, files=files, data=data) #リクエスト送信
受信部の処理
こちらもpillowで受ける場合とOpenCVで受ける場合の2種類を準備しています。処理の最後に作成される画像はOpenCV形式です。
OpenCVで受けるときの例に
#--- media_type="application/octet-stream"の時
image =(pickle.loads(image_data))#元の形式にpickle.loadsで復元
を追加しています。pillowで受けるときもt2i同様にこの処理を用いると早くなリます。
pillowで受ける時
# レスポンスが成功であることを確認
if response.status_code == 200:
print("response.status_code== 200")
# レスポンスのコンテンツ（画像データ）をバイナリ形式で取得
image_data = response.content
#--- media_type="application/octet-stream"の時
image =(pickle.loads(image_data))#元の形式にpickle.loadsで復元
#---media_type="image/jpeg"の時
# イメージをバイト配列に変換
#image = Image.open(BytesIO(image_data)) # バイナリデータをPIL形式に変換opencvで受ける時
#--- media_type="application/octet-stream"の時
image =(pickle.loads(image_data))#元の形式にpickle.loadsで復元
を追加しています。
#opencvで受ける時
if response.status_code == 200:
# レスポンスから画像データを読み込む
image_data = np.frombuffer(response.content, np.uint8)
#---media_type="image/jpeg"の時
# バイナリデータからOpenCVカラー画像に変換
#image = cv2.imdecode(image_data, cv2.IMREAD_COLOR) # バイナリデータをOpenCVカラー画像に変換
#--- media_type="application/octet-stream"の時
image =(pickle.loads(image_data))#元の形式にpickle.loadsで復元 表示部
表示は生成データがOpenCVのケースもあるので、全てOpenCVとし、ループ内表示、スレッディング表示のいずれかが選択できます。
#表示をループ内で行う時
#cv2.imshow("image", image)
#cv2.waitKey(1)
#スレッドで表示
th_img = image #生成画像を準備
th_img_flag=True #画像準備フラグセットパフォーマンスの比較
クライアント側で1ループ回る場合の速度
t2i
インターラクティブプロンプト 有り 12.2fps
インターラクティブプロンプト 無し 13.68fps
media_type="application/octet-stream"の時
インターラクティブプロンプト 有り 27.9fps
インターラクティブプロンプト 無し 29.6fps
i2i
pillowイメージでリクエスト 返答はpillow 5.32fps
Opencvイメージでリクエスト 返答はpillow 10.15fps
Opencvイメージでリクエスト 返答はOpencv 16.66fps
media_type="application/octet-stream"の時
プロンプト 有り 送信 "image/jpg" 20.47fps
プロンプト 有り 送信 "application/octet-stream" 20.57fps
プロンプト 無し 送信 "image/jpg" 22.11fps
プロンプト 無し 送信 "application/octet-stream" 22.34fps
まとめ
今回はFastAPIによるサーバでStreamDiffuisonを試しました。データ量の少ない場合や画像数が少ないときの通信は何も気にすることはありませんでしたが、高速生成するStreamDiffuisonではプロトコルや画像の形式がパフォーマンスに大きく影響することがわかります。また、TensorRTを有効にする場合、t2iとi2iのt_index_listの設定が異なることから共通化出来ていません。次回は更にパフォーマンスが上がるTCP/IPによる実装と、t2i/i2i共通化も試したいと思います。
FastAPIサーバ全体のコード
【#---media_type="application/octet-stream"の時---高速通信できる】を追加しています。
import torch
from diffusers import AutoencoderTiny, StableDiffusionPipeline,StableDiffusionImg2ImgPipeline
from diffusers.utils import load_image
from streamdiffusion import StreamDiffusion
from streamdiffusion.image_utils import postprocess_image
import numpy as np
import time
import cv2
from PIL import Image
import pickle
model_path = "./models/Model/Counterfeit-V3.0_fix_fp16.safetensors"
lora_path="./models/LoRA/megu_sports_v02.safetensors"
lora_preload=True
lora_preload_weights=0.5
lora_load=False
lora_scale_weights=0.1
tensorrt=True
prompt="masterpiece, best quality, 1girl,"
negative_prompt=""
guidance_scale=1.2
delta=1.0
seed=1
#--- RCFG の指定
cfg_type = "none"
#cfg_type = "full"
#cfg_type = "self"
#cfg_type = "initialize"
#--- t_index_lisの指定
#t_index_list=[40]
#t_index_list=[32, 45]
#t_index_list=[38,40 ,42,45]
#t_index_list=[20,30,40]
#t_index_list=[40 ,42,45]
#t_index_list=[41,42,44,45] #cam<<<<<<<<<<<<<<<<<<<< i2iのときは有効にすること
t_index_list=[0, 16, 32, 45] #t2i <<<<<<<<<<<<<<<<<<<< t2iのときは有効にすること
img_size=[512,512]
#TensorRT engineフォルダー名 "engines_tindex"＋t_index_listの要素数＋CFGタイプ
tensorrt_engine="engines_tindex" + str(len(t_index_list))+"_"+ cfg_type
# =================================== 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()
#-------------------- 初期設定--------------------------
#--- モデルのロード
pipe = StableDiffusionPipeline.from_single_file(
model_path).to(
device=torch.device("cuda"),
dtype=torch.float16,
)
#--- 非公式 独自LoRAのロード
if lora_preload==True:
print("lora_path=",lora_path)
pipe.load_lora_weights("latent-consistency/lcm-lora-sdv1-5", adapter_name="lcm") #Stable Diffusion 1.5 のLCM LoRA
pipe.load_lora_weights(lora_path, adapter_name="papercut")
pipe.set_adapters(["lcm", "papercut"], adapter_weights=[1.0, lora_preload_weights])
# ---Wrap the pipeline in StreamDiffusion
stream = StreamDiffusion(
pipe,
t_index_list=t_index_list,
torch_dtype=torch.float16,
cfg_type=cfg_type,
width =img_size[0], #height/width TensorRT有効のときはサイズに注意 512x512のみ
height = img_size[1],
)
# ---IIf the loaded model is not LCM, merge LCM
stream.load_lcm_lora()
stream.fuse_lora()
#---公式 独自LoRAのロード
if lora_load & (lora_path != "none"):
stream.load_lora(lora_path)
stream.fuse_lora(lora_scale=lora_scale_weights)
# ---Use Tiny VAE for further acceleration
stream.vae = AutoencoderTiny.from_pretrained("madebyollin/taesd").to(device=pipe.device, dtype=pipe.dtype)
# --- Enable acceleration いずれかを有効にする
if tensorrt==False:
pipe.enable_xformers_memory_efficient_attention()
else: # Enable acceleration with TensroRT
from streamdiffusion.acceleration.tensorrt import accelerate_with_tensorrt
stream = accelerate_with_tensorrt(stream, tensorrt_engine, max_batch_size=4,) #Step=3
# ---事前計算
stream.prepare(
prompt,
negative_prompt=negative_prompt,
guidance_scale = guidance_scale,
delta=delta,
seed=seed,
)
# ---Warmup >= len(t_index_list) x frame_buffer_size
init_image= Image.new('RGB', img_size, (255, 255, 255))
for _ in range(len(t_index_list)):
stream(init_image)
#-------------------- FastAPI エンドポイント--------------------------
# ---画像生成 Run the stream infinitely i2i imput=pillow形式 遅い
@app.post("/stream_i2i/")
def stream_i2i(file: UploadFile = File(...), prompt: str = Form(None) , image_type: str = Form(None),out_type: str = Form(None)):
if image_type==None or image_type=="pil": #入力はpillow
image_data = file.file.read()
in_image = Image.open(BytesIO(image_data)) # バイナリデータをPIL形式に変換
elif image_type=="cv2":
contents = file.file.read()
nparr = np.frombuffer(contents, np.uint8)
cv2_image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)# バイナリデータをCv2形式に変換
image = cv2.cvtColor(cv2_image, cv2.COLOR_BGR2RGB)
in_image = Image.fromarray( image)
print("+++++++++++ prompt ++++++++++++",prompt)
if prompt != None:
#動的にプロンプトを変える
stream.update_prompt(prompt)
#---画像生成 i2i
x_output = stream(in_image)
ximage=postprocess_image(x_output, output_type="pil")[0]
# 出力がpillow
if out_type=="pil":
#---media_type="image/png"の時
"""
#pillowイメージをバイト配列に変換 img_byte_arry = io.BytesIO()
ximage.save(img_byte_arry, format='PNG')
img_bytes= img_byte_arry.getvalue()
#---スポンスとしてバイト配列を返す
return Response(content=img_bytes, media_type="image/png")
"""
#---media_type="application/octet-stream"の時---高速通信できる
frame_data = pickle.dumps(cv2_image , 5) # tx_dataはpklデータ
return Response(content=frame_data, media_type="application/octet-stream")
# 出力がOpenVC
else:
#---pillowイメージをOpenVCに変換
new_image = np.array(ximage, dtype=np.uint8)
cv2_image = cv2.cvtColor(new_image, cv2.COLOR_RGB2BGR)
#---media_type="image/jpeg"の時
"""
# Opencv画像をバイナリ形式にエンコードする
_, encoded_img = cv2.imencode('.jpg', cv2_image)
#---スポンスとしてバイト配列を返す
return Response(content=encoded_img.tobytes(), media_type="image/jpeg")
"""
#---media_type="application/octet-stream"の時---高速通信できる
frame_data = pickle.dumps(cv2_image , 5) # tx_dataはpklデータ
return Response(content=frame_data, media_type="application/octet-stream")
class generate_t2i(BaseModel):
prompt: str=None
@app.post("/stream_t2i/")
def stream_t2i(request:generate_t2i):
prompt=request.prompt
if prompt !=None:
#動的にプロンプトを変える
stream.update_prompt(prompt)
print("prompt=",prompt)
#画像生成 t2i
x_output = stream.txt2img()
ximage=postprocess_image(x_output, output_type="pil")[0]
#media_type="image/png"の時
"""
#---pillowイメージをバイト配列に変換
img_byte_arry = io.BytesIO()
ximage.save(img_byte_arry, format='PNG')
img_bytes= img_byte_arry.getvalue()
#---スポンスとしてバイト配列を返す
return Response(content=img_bytes, media_type="image/png")
"""
#media_type="application/octet-stream"の時---高速通信できる
frame_data = pickle.dumps(ximage, 5) # tx_dataはpklデータ
return Response(content=frame_data, media_type="application/octet-stream")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
t2iテストプログラム全体
import os
from PIL import Image
import cv2
import time
from time import sleep
import numpy as np
import requests
import json
import io
from io import BytesIO
url = 'http://0.0.0.0:8000/stream_t2i/'
# -----表示のスレッド化準備（必要に応じて）
global th_img , th_img_flag
#>>> 表示スレッドの定義
import threading
def disp_th():
global th_img , th_img_flag
while True:
if th_img_flag==True:
imgCV_RGB = np.array(th_img, dtype=np.uint8)
th_img = np.array(imgCV_RGB)[:, :, ::-1]
cv2.imshow("image", th_img)
cv2.waitKey(1)
th_img_flag=False
time.sleep(0.01)
# -----プログラムの開始
def main():
global th_img , th_img_flag #表示スレッド用共有データとセマフォ
#-----表示スレッドを開始
th_img_flag=False
thread = threading.Thread(target=disp_th, name='disp_th',daemon = True)
thread.start()
#>>>動的プロンプトの初期プロンプト
prompt = "masterpiece, best quality, 1girl,"
prompt_list=[
"1girl","long hair,","white shirt,","serafuku,","brown hair,","looking at viewer,","blush,","smile,", "bangs,","blue eyes,","simple background,", "t-shirt,",\
"white background,","walk a head,","white background,","walk a head,","white background,","walk a head,","white background,","walk a head,","white background,"]
count=len(prompt_list)
total_time=0
for n in range(len(prompt_list)):
start_time=time.time()
prompt=prompt+ prompt_list[n]
# リクエストを送信
print("prompt=",prompt)
#data = {"prompt": prompt}
data = {"prompt": None}
response = requests.post(url, json=data )
# レスポンスが成功であることを確認
if response.status_code == 200:
print("response.status_code== 200")
# レスポンスのコンテンツ（画像データ）をバイナリ形式で取得
image_data = response.content
# イメージをバイト配列に変換
try:
image = Image.open(BytesIO(image_data)) # バイナリデータをPIL形式に変換
#スレッドで表示
th_img = image #生成画像を準備
th_img_flag=True #画像準備フラグセット
except:
print("error")
#生成時間とフレームレートの表示
end_time=time.time()
print("生成時間",end_time- start_time)
print("i-fps",1/(end_time- start_time))
total_time=total_time+(end_time- start_time)
print("avr-time:",total_time/count)
print("avr-fps :",1/(total_time/count))
if __name__ == '__main__':
main()
i2iテストプログラム全体
import os
from PIL import Image
import cv2
import time
from time import sleep
import numpy as np
from streamdiffusion.image_utils import postprocess_image
import requests
import io
from io import BytesIO
from capture import init_cam,cap_img_pil,cap_img_pil_t,cap_close
url = 'http://0.0.0.0:8000/stream_i2i/'
# -----表示のスレッド化準備（必要に応じて）
global th_img , th_img_flag
#>>> 表示スレッドの定義
import threading
def disp_th():
global th_img , th_img_flag
while True:
if th_img_flag==True:
cv2.imshow("image", th_img)
cv2.waitKey(1)
th_img_flag=False
time.sleep(0.01)
# -----プログラムの開始
def main():
global th_img , th_img_flag #表示スレッド用共有データとセマフォ
#capture.pyを使わないとき
"""
cap = cv2.VideoCapture(0)
# キャプチャがオープンしていることを確認
if not cap.isOpened():
print("カメラを開けません")
status=False
"""
# Webカメラの初期化キャプチャを開始
status=init_cam()
# キャプチャがオープンしていることを確認
if status==False:
print("カメラを開けません")
#-----表示スレッドを開始
th_img_flag=False
thread = threading.Thread(target=disp_th, name='disp_th',daemon = True)
thread.start()
count=20
total_time=0
for n in range(count):
start_time=time.time()
#カメラ入力 cap_img_pil()はpilloとOpenCV形式が得られる
#ret, cv2_image = cap.read() #capture.pyを使わず、OpenCV形式で送る場合
pil_image , cv2_image = cap_img_pil()
# pilloイメージでリクエスト 返答はpillow
"""
# pilloイメージをバイト配列に変換
img_byte_arry = io.BytesIO()
pil_image.save(img_byte_arry, format='PNG')
img_byte_arry= img_byte_arry.getvalue()
files = {"file": ("img.png", img_byte_arry, "image/png")}
data = {'prompt': "masterpiece, best quality, 1girl",
"image_type":"pil", #pil or cv2 画像形式に合わせる
"out_type":"pil"}
response = requests.post(url, files=files, data=data) #リクエスト送信
"""
# Opencvイメージでリクエスト 返答はpillow
"""
# Opencv画像をバイナリ形式にエンコードする
_, img_encoded = cv2.imencode('.jpg', cv2_image)
files = {"file": ("img.jpg", img_encoded, "image/jpg")}
data = {"prompt": "masterpiece, best quality, 1girl",
"image_type" :"cv2", #pil or cv2 画像形式に合わせる
"out_type":"pil"}
# リクエストを送信
response = requests.post(url, files=files, data=data) #リクエスト送信
"""
# Opencvイメージでリクエスト 返答はOpencv
# Opencv画像をバイナリ形式にエンコードする
_, img_encoded = cv2.imencode('.jpg', cv2_image)
files = {"file": ("img.jpg", img_encoded, "image/jpg")}
data = {"prompt": "masterpiece, best quality, 1girl",
"image_type" :"cv2"} #pil or cv2 画像形式に合わせる
# リクエストを送信
response = requests.post(url, files=files, data=data) #リクエスト送信
#pillowで受ける時
"""
# レスポンスが成功であることを確認
if response.status_code == 200:
print("response.status_code== 200")
# レスポンスのコンテンツ（画像データ）をバイナリ形式で取得
image_data = response.content
# イメージをバイト配列に変換
image = Image.open(BytesIO(image_data)) # バイナリデータをPIL形式に変換 imageはPIL形式
imgCV_RGB = np.array(mage, dtype=np.uint8)# Opencvイメージに変換
image = np.array(imgCV_RGB)[:, :, ::-1]
"""
#opencvで受ける時
if response.status_code == 200:
# レスポンスから画像データを読み込む
img_data = np.frombuffer(response.content, np.uint8)
# バイナリデータからOpenCVカラー画像に変換
image = cv2.imdecode(img_data, cv2.IMREAD_COLOR) # バイナリデータをOpenCVカラー画像に変換
#表示をループ内で行う時
#cv2.imshow("image", image)
#cv2.waitKey(1)
#スレッドで表示
th_img = image #生成画像を準備
th_img_flag=True #画像準備フラグセット
#生成時間とフレームレートの表示
end_time=time.time()
print("生成時間",end_time- start_time)
print("i-fps",1/(end_time- start_time))
total_time=total_time+(end_time- start_time)
cap_close()
print("avr-time:",total_time/count)
print("avr-fps :",1/(total_time/count))
if __name__ == '__main__':
main()
capture.py
import os
from PIL import Image
import cv2
import time
def init_cam():
global cap
# Webカメラのキャプチャを開始
cap = cv2.VideoCapture(0)
# キャプチャがオープンしていることを確認
if not cap.isOpened():
print("カメラを開けません")
status=False
return status
ret, frame = cap.read()
print("caputure init")
#cv2.imshow('Frame',frame )
status=True
return status
def cap_img():
global cap
# カメラからフレームを読み込む
ret, frame = cap.read()
# フレームの表示と返送
#cv2.imshow('Frame', frame)
return frame
#get pil and cv2イメージ
def cap_img_pil():
img_cv2=cap_img()
# cv2-> PIL
new_image = cv2.cvtColor(img_cv2, cv2.COLOR_BGR2RGB)
pil_image= Image.fromarray(new_image)
#サイズ調整
w,h = pil_image.size
#正方形
img_new = Image.new('RGB', [h,h], (255, 255, 255))
img_new .paste(pil_image ,(-int((w-h)/2),0))
#縦長
#pil_image=pil_image.resize((768,768))
#img_new = Image.new('RGB', [512,768], (255, 255, 255))
#img_new .paste(pil_image ,(-100,0))
return img_new , img_cv2
#get pil and cv2イメージ
def cap_img_pil_t(interval , queue):#interval=mS
print("Start CAM")
while True:
start_time = time.time()
print("CAM time",start_time)
img_cv2=cap_img()
new_image = cv2.cvtColor(img_cv2, cv2.COLOR_BGR2RGB)
pil_image= Image.fromarray(new_image)
w,h = pil_image.size
img_new = Image.new('RGB', [h,h], (255, 255, 255))
img_new .paste(pil_image ,(-int((w-h)/2),0))
#img_new .paste(pil_image ,(-50,0))
#return img_new , img_cv2
queue_data=[img_new,img_cv2]
print("queue.put")
queue.put(queue_data)
# 次の実行までの待機時間を計算
elapsed_time = time.time() - start_time
time_to_wait = max(0, interval - elapsed_time) # interval - 実行にかかった時間
time.sleep(time_to_wait)
def cap_close():
cap.release()