ComfiUI API调用随记

来进行知识接力了：

首先了解下ComfiUI的API
stable diffusion comfyui的api使用教程-CSDN博客

对于ComfiUI，接口比较简单。查询接口比较容易看明白。
对于发起prompt的请求，如果需要图片的，则需预先上传图片给ComfiUI，获取返回的文件名进行调用。
 

发起调用后，实际上是加入队列，获得一个任务ID，通过ws实时获取对应的执行状态，进度等数据。当未执行时，则查询任务获得图片。

AI的解释
 

这段代码是一个使用websockets API的Python示例，用于检测提示执行何时完成。一旦提示执行完成，它将使用/history端点下载图像。

首先，导入所需的库，如websocket（需要安装websocket-client），uuid，json，urllib.request和urllib.parse。设置服务器地址和客户端ID。

定义几个函数：

1. queue_prompt(prompt)：向服务器发送一个提示，并返回包含prompt_id的响应。
2. get_image(filename, subfolder, folder_type)：从服务器下载指定的图像数据。
3. get_history(prompt_id)：获取与给定prompt_id关联的历史记录。
4. get_images(ws, prompt)：通过websocket接收消息，直到提示执行完成。然后获取历史记录并下载所有相关图像。

在主程序中，首先定义了一个JSON字符串prompt_text，表示要使用的提示。然后解析这个字符串为Python字典，并根据需要修改其中的文本提示和种子值。

接下来，创建一个WebSocket连接到服务器，并调用get_images()函数来获取图像。最后，显示输出图像（这部分已注释掉）。

简而言之，这段代码的主要目的是使用websockets API与服务器通信，以执行提示并在提示完成后下载生成的图像。

AI加上注释的代码

# 导入所需库
import websocket # NOTE: 需要安装websocket-client (https://github.com/websocket-client/websocket-client)
import uuid
import json
import urllib.request
import urllib.parse# 设置服务器地址和客户端ID
server_address = "127.0.0.1:8188"
client_id = str(uuid.uuid4())# 定义向服务器发送提示的函数
def queue_prompt(prompt):p = {"prompt": prompt, "client_id": client_id}data = json.dumps(p).encode('utf-8')req =  urllib.request.Request("http://{}/prompt".format(server_address), data=data)return json.loads(urllib.request.urlopen(req).read())# 定义从服务器下载图像数据的函数
def get_image(filename, subfolder, folder_type):data = {"filename": filename, "subfolder": subfolder, "type": folder_type}url_values = urllib.parse.urlencode(data)with urllib.request.urlopen("http://{}/view?{}".format(server_address, url_values)) as response:return response.read()# 定义获取历史记录的函数
def get_history(prompt_id):with urllib.request.urlopen("http://{}/history/{}".format(server_address, prompt_id)) as response:return json.loads(response.read())# 定义通过WebSocket接收消息并下载图像的函数
def get_images(ws, prompt):prompt_id = queue_prompt(prompt)['prompt_id']output_images = {}while True:out = ws.recv()if isinstance(out, str):message = json.loads(out)if message['type'] == 'executing':data = message['data']if data['node'] is None and data['prompt_id'] == prompt_id:break # 执行完成else:continue # 预览是二进制数据history = get_history(prompt_id)[prompt_id]for o in history['outputs']:for node_id in history['outputs']:node_output = history['outputs'][node_id]if 'images' in node_output:images_output = []for image in node_output['images']:image_data = get_image(image['filename'], image['subfolder'], image['type'])images_output.append(image_data)output_images[node_id] = images_outputreturn output_images# 示例JSON字符串，表示要使用的提示
prompt_text = """
{"3": {"class_type": "KSampler","inputs": {"cfg": 8,"denoise": 1,"latent_image": ["5",0],"model": ["4",0],"negative": ["7",0],"positive": ["6",0],"sampler_name": "euler","scheduler": "normal","seed": 8566257,"steps": 20}},"4": {"class_type": "CheckpointLoaderSimple","inputs": {"ckpt_name": "v1-5-pruned-emaonly.ckpt"}},"5": {"class_type": "EmptyLatentImage","inputs": {"batch_size": 1,"height": 512,"width": 512}},"6": {"class_type": "CLIPTextEncode","inputs": {"clip": ["4",1],"text": "masterpiece best quality girl"}},"7": {"class_type": "CLIPTextEncode","inputs": {"clip": ["4",1],"text": "bad hands"}},"8": {"class_type": "VAEDecode","inputs": {"samples": ["3",0],"vae": ["4",2]}},"9": {"class_type": "SaveImage","inputs": {"filename_prefix": "ComfyUI","images": ["8",0]}}
}
"""# 将示例JSON字符串解析为Python字典，并根据需要修改其中的文本提示和种子值
prompt = json.loads(prompt_text)
prompt["6"]["inputs"]["text"] = "masterpiece best quality man"
prompt["3"]["inputs"]["seed"] = 5# 创建一个WebSocket连接到服务器
ws = websocket.WebSocket()
ws.connect("ws://{}/ws?clientId={}".format(server_address, client_id))# 调用get_images()函数来获取图像
images = get_images(ws, prompt)# 显示输出图像（这部分已注释掉）
#Commented out code to display the output images:# for node_id in images:
#     for image_data in images[node_id]:
#         from PIL import Image
#         import io
#         image = Image.open(io.BytesIO(image_data))
#         image.show()

代码尚未深入，就目前的知识进度做一下记录和分享