当前位置：首页 > article >正文

ComfiUI API调用随记

article 2024/12/30 1:31:58

来进行知识接力了：

首先了解下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。

定义几个函数：

queue_prompt(prompt)：向服务器发送一个提示，并返回包含prompt_id的响应。
get_image(filename, subfolder, folder_type)：从服务器下载指定的图像数据。
get_history(prompt_id)：获取与给定prompt_id关联的历史记录。
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_output

    return 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()

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

查看全文

http://www.kler.cn/a/157134.html