登录

用Python调用ComfyUI API，一键生成图片攻略使用

1、comfyui设置

打开comfyui，记录下对应的端口，设置开发者模式

打开一条workflow工作流，这里以comfyui自带的工作流为例，保存为api格式

再次打开api格式工作流（注：这里一定再次点击查看是否能运行正常，因为有的节点可能在api格式中无法运作），确定可以在comfyui中正常启动就可以下一步了

2、python设置

新建一个python脚本，将以下内容粘贴进去

import json
import websocket  # NOTE: websocket-client (https://github.com/websocket-client/websocket-client)
import uuid
import urllib.request
import urllib.parse
import random
import pandas as pd
import os  # 添加在文件开头的导入部分

def check_dependencies():
    try:
        import pandas as pd
        import websocket
    except ImportError as e:
        print(f"错误：缺少必要的依赖包。请运行以下命令安装：")
        print("pip install pandas websocket-client")
        exit(1)

# 在主程序开始前检查依赖
check_dependencies()

# 定义一个函数来显示GIF图片
def show_gif(fname):
    # 在命令行环境中，只需打印文件保存的位置
    print(f"图片已保存到: {fname}")
    return None
# 定义一个函数向服务器队列发送提示信息
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']
    print('prompt')
    print(prompt)
    print('prompt_id:{}'.format(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:
                    print('执行完成')
                    break  # 执行完成
        else:
            continue  # 预览为二进制数据
    history = get_history(prompt_id)[prompt_id]
    print(history)
    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
            # 视频分支
            if 'videos' in node_output:
                videos_output = []
                for video in node_output['videos']:
                    video_data = get_image(video['filename'], video['subfolder'], video['type'])
                    videos_output.append(video_data)
                output_images[node_id] = videos_output
    print('获取图片完成')
    print(output_images)
    return output_images
# 解析工作流并获取图片
def parse_worflow(ws, prompt, seed, workflowfile):
    workflowfile = workflowfile
    print('workflowfile:'+workflowfile)
    with open(workflowfile, 'r', encoding="utf-8") as workflow_api_txt2gif_file:
        prompt_data = json.load(workflow_api_txt2gif_file)
        # 添加调试信息
        print("工作流文件内容:")
        print(json.dumps(prompt_data, indent=2))
        
        # 找到包含文本输入的节点
        for node_id, node_data in prompt_data.items():
            if "inputs" in node_data and "text" in node_data["inputs"]:
                print(f"找到文本输入节点: {node_id}")
                prompt_data[node_id]["inputs"]["text"] = prompt
                break
        
        return get_images(ws, prompt_data)
# 生成图像并显示
def generate_clip(prompt, seed, workflowfile, idx):
    print('seed:'+str(seed))
    ws = websocket.WebSocket()
    ws.connect("ws://{}/ws?clientId={}".format(server_address, client_id))
    images = parse_worflow(ws, prompt, seed, workflowfile)
    for node_id in images:
        for image_data in images[node_id]:
            from datetime import datetime
            # 获取当前时间，并格式化为 YYYYMMDDHHMMSS 的格式
            timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
            # 使用格式化的时间戳在文件名中
            GIF_LOCATION = "{}/{}_{}_{}.png".format(SageMaker_ComfyUI, idx, seed, timestamp)
            print('GIF_LOCATION:'+GIF_LOCATION)
            with open(GIF_LOCATION, "wb") as binary_file:
                # 写入二进制文件
                binary_file.write(image_data)
            show_gif(GIF_LOCATION)
            print("{} DONE!!!".format(GIF_LOCATION))

# 创建示例数据
data = {
    'prompt': [
        "一只可爱的猫",
        "一只奔跑的狗",
        "一朵红色的花"
    ]
}

# 创建 DataFrame 并保存为 CSV 文件
df = pd.DataFrame(data)
df.to_csv('prompt.csv', index=False)

# 修改读取函数
def read_prompts_from_csv(csv_file_path):
    df = pd.read_csv(csv_file_path)
    return df['prompt'].tolist()

# Execute the main function
if __name__ == "__main__":
    # 设置工作目录和项目相关的路径
    WORKING_DIR = 'output'
    SageMaker_ComfyUI = WORKING_DIR
    
    # 确保输出目录存在
    if not os.path.exists(WORKING_DIR):
        os.makedirs(WORKING_DIR)
        print(f"创建输出目录: {WORKING_DIR}")
    
    workflowfile = 'Remove_background_api.json'
    COMFYUI_ENDPOINT = '127.0.0.1:8188'
    server_address = COMFYUI_ENDPOINT
    client_id = str(uuid.uuid4())  # 生成一个唯一的客户端ID
    seed = 15465856
    csv_file_path = 'prompt.csv'  # 改用 CSV 文件
    prompts = read_prompts_from_csv(csv_file_path)
    idx = 1
    for prompt in prompts:
        generate_clip(prompt, seed, workflowfile, idx)
        idx += 1

替换相应的参数，比如：

  • workflowfile文件名
  • COMFYUI_ENDPOINT地址，

WORKING_DIR输出文件的目录是对应脚本文件的

接着打开存储工作流api信息的josn文件

找到我们要修改的参数，比如我要修改正向提示词参数

可以看到其对应的键值位置是[“6”][“inputs”][“text”]

就可以设置对应的参数，在函数中解析替换这个键值

最后还设置了一个文件’prompt.xlsx’，方便写入大量的prompt进行替换

这样就可以解放双手，开始自由出图了

关键步骤总结：

  1. ComfyUI 设置：
    • 确认 ComfyUI 的端口。
    • 打开一个工作流并保存为 API 格式，确保所有节点在 API 格式下正常运行。
  2. Python 脚本配置：
    • 利用 websocket-client 和 urllib 模块实现对 ComfyUI API 的调用。
    • 实现队列消息发送 (queue_prompt)、图片获取 (get_image)、历史记录解析 (get_history) 等功能。
    • 利用 parse_workflow 函数解析工作流 JSON 文件，动态修改其中的参数并生成图像。
  3. 批量生成支持：
    • 从 CSV 文件中读取 prompts 列表，实现批量生成和存储。
    • 支持设置正向提示词、种子值等参数。
  4. 提示与参数替换：
    • 根据 JSON 工作流中的键值位置，例如 ["6"]["inputs"]["text"]，修改提示词等配置。
    • 输出文件路径根据时间戳动态生成。

优化与注意事项：

  1. 错误处理：
    • 脚本中对 API 请求和 WebSocket 通信的异常处理较少，建议在 queue_prompt 和 ws.recv() 等函数中添加 try-except 块以捕获可能的网络错误。
  2. JSON 文件路径管理：
    • 可通过 os.path 模块确保 JSON 文件路径在不同环境下的兼容性。
  3. 改进用户交互：
    • 提供更加人性化的参数输入和输出，比如通过命令行参数或配置文件设置 COMFYUI_ENDPOINT 等参数。
  4. 多线程优化：
    • 当前生成流程是串行的，若有性能瓶颈，可以利用 Python 的 concurrent.futures 模块实现并行化处理，提高生成速度。
  5. 图片/视频二进制数据解析：
    • 针对 ws.recv() 返回的二进制数据，可以在处理时添加对其格式的验证，避免可能的解析错误。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注