from IPython import display
with open(fname, 'rb') as fd:
b64 = base64.b64encode(fd.read()).decode('ascii')
return display.HTML(f'<img src="data:image/gif;base64,{b64}" />')
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:
def get_history(prompt_id):
with urllib.request.urlopen("http://{}/history/{}".format(server_address, prompt_id)) as response:
return json.loads(response.read())
def get_images(ws, prompt):
prompt_id = queue_prompt(prompt)['prompt_id']
print('prompt_id:{}'.format(prompt_id))
if isinstance(out, str):
message = json.loads(out)
if message['type'] == 'executing':
if data['node'] is None and data['prompt_id'] == prompt_id:
print('Execution completed')
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:
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:
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('Image retrieval completed')
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)
prompt_data["6"]["inputs"]["text"] = prompt
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 image_data in images[node_id]:
from datetime import datetime
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
GIF_LOCATION = "{}/{}_{}_{}.png".format(SageMaker_ComfyUI, idx, seed, timestamp)
os.makedirs(os.path.dirname(GIF_LOCATION), exist_ok=True)
print('GIF_LOCATION:' + GIF_LOCATION)
with open(GIF_LOCATION, "wb") as binary_file:
binary_file.write(image_data)
print("{} DONE!!!".format(GIF_LOCATION))
if __name__ == "__main__":
SageMaker_ComfyUI = WORKING_DIR
workflowfile = 'workflow_api.json'
COMFYUI_ENDPOINT = '14.**.**.**:***'
server_address = COMFYUI_ENDPOINT
client_id = str(uuid.uuid4())
"Anime girl with long pastel pink hair, large sapphire gradient eyes, fair skin, wearing a floral kimono - inspired outfit with floating sakura petals.",
"Anime girl with long flowing mint green hair, striking emerald gradient eyes, smooth porcelain skin, adorned in a celestial - themed yukata with tiny glowing stars and floating iridescent feathers."]
generate_clip(prompt, seed, workflowfile, idx)