ComfyUI for stable diffusion: API call script to run automated workflows

api_comfyui-img2img.py

# https://gist.github.com/audioscavenger/0471b389b88ce3a3998d510832affed9
# 1. install ComfyUI-Custom-Scripts: cd E:\GPT\ComfyUI\custom_nodes\ && git clone https://github.com/pythongosssss/ComfyUI-Custom-Scripts
# 2. enable dev mode options
# 3. save workflow in API format
# 4. python api_comfyui-img2img.py -w E:\GPT\ComfyUI\output\ComfyUI-workflow-recolor-api.json -i E:\GPT\ComfyUI\output\image-grey.jpg -o outputPrefix
# NOTE: Saving image outside the output folder is not allowed.

import getopt, sys, os
import json, urllib, random
from urllib import request, parse

#This is the ComfyUI api prompt format.

#If you want it for a specific workflow you can "enable dev mode options"
#in the settings of the UI (gear beside the "Queue Size: ") this will enable
#a button on the UI to save workflows in api format.

#keep in mind ComfyUI is pre alpha software so this format will change a bit.

# defaults
author = AudioscavengeR
version = 1.0.1
comfyUrl = "http://127.0.0.1:8188/prompt"
workflowFile = None
inputFile = None
outputPrefix = None

# Remove 1st argument from the
# list of command line arguments
argumentList = sys.argv[1:]

# Options
options = "hw:i:o:"
# Long options
long_options = ["help", "workflow", "input=", "output="]

try:
  # Parsing argument
  arguments, values = getopt.getopt(argumentList, options, long_options)
  
  # checking each argument
  for currentArgument, currentValue in arguments:
    if currentArgument in ("-h", "--help"):
      print (("usage: python %s --help") % (sys.argv[0]))
      print (("       python %s --workflow file.json [--input=input.jpg] [--output=output.png]") % (sys.argv[0]))
    elif currentArgument in ("-w", "--workflow"):
      workflowFile = currentValue
      print (("workflow   : %s") % (currentValue))
    elif currentArgument in ("-i", "--input"):
      inputFile = currentValue
      print (("input file : %s") % (currentValue))
    elif currentArgument in ("-o", "--output"):
      outputPrefix = os.path.basename(currentValue)
      print (("output file: %s") % (currentValue))
except getopt.error as err:
  # output error, and return with an error code
  print (str(err))

#this is the one for the default workflow
prompt_text_example_1 = """
{
  "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
      ]
    }
  }
}
"""

def jsonRecurseAndReplace(json, keyName=None, replaceTo=None, indent = '  '):
    for key in json.keys():
        if isinstance(json[key], dict):
            jsonRecurseAndReplace(json[key], keyName, replaceTo, indent+'  ')
        else:
            if keyName is None:
                print(indent+str(key)+' = '+str(json[key]))
            else:
                if str(key) == keyName:
                    if replaceTo is not None:
                        json[key] = replaceTo
                    print(indent+str(key)+' = '+str(json[key]))
                # else:

# inputFile='E:\GPT\ComfyUI\output\image-grey.jpg'
# outputPrefix='E:\GPT\ComfyUI\output\image-grey-recolor.jpg'
# jsonRecurseAndReplace(prompt)
# jsonRecurseAndReplace(prompt, 'choose file to upload', inputFile)
# jsonRecurseAndReplace(prompt, 'filename_prefix', outputPrefix)

def queue_prompt(prompt):
  p = {"prompt": prompt}
  data = json.dumps(p).encode('utf-8')
  # req =  urllib.Request(comfyUrl, data=data)
  req =  urllib.request.Request(comfyUrl, data=data)
  # urllib.request.urlopen(req)
  with urllib.request.urlopen(req) as response:
    # Do stuff with complete data == wait for a reply. Unfortunately Comfy does not reply anything
    answer = response.read()


if (outputPrefix is None and inputFile is not None):
  if (os.path.isfile(inputFile)):
    outputPrefix = ("%s-recolor" % os.path.splitext(os.path.basename(inputFile)))[0]

if (workflowFile is not None and os.path.isfile(workflowFile)):
  with open(workflowFile) as file:
    prompt = json.load(file)
    jsonRecurseAndReplace(prompt, 'choose file to upload', inputFile)
    jsonRecurseAndReplace(prompt, 'filename_prefix', outputPrefix)
else:
  prompt = json.loads(prompt_text_example_1)
  #set the text prompt for our positive CLIPTextEncode
  prompt["6"]["inputs"]["text"] = "masterpiece best quality man"
  #set the seed for our KSampler node
  prompt["3"]["inputs"]["seed"] = 5


# ddebug: print(prompt)
queue_prompt(prompt)

ComfyUI will not wait after you inject the POST, because of its queuing system.