130 lines
3.7 KiB
Python
130 lines
3.7 KiB
Python
from server import PromptServer
|
|
from io import BytesIO
|
|
from PIL import Image
|
|
import numpy as np
|
|
import base64
|
|
import torch
|
|
import json
|
|
|
|
class ImageOutput:
|
|
def __init__(self):
|
|
pass
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"images": ("IMAGE", {"default": None, "forceInput": True}),
|
|
"Actions": ("STRING", {"default": None})}
|
|
}
|
|
|
|
RETURN_TYPES = ("IMAGE",)
|
|
RETURN_NAMES = ("IMAGE",)
|
|
|
|
FUNCTION = "Proc"
|
|
|
|
OUTPUT_NODE = True
|
|
|
|
CATEGORY = "Knodes"
|
|
|
|
def Proc(self, images, Actions = ""):
|
|
outs = []
|
|
|
|
for single_image in images:
|
|
img = np.asarray(single_image * 255., dtype=np.uint8)
|
|
img = Image.fromarray(img)
|
|
buffered = BytesIO()
|
|
img.save(buffered, format="PNG")
|
|
img_str = base64.b64encode(buffered.getvalue()).decode()
|
|
outs.append(img_str)
|
|
|
|
PromptServer.instance.send_sync("knodes", {"images": outs, "Actions": Actions})
|
|
|
|
return (images,)
|
|
|
|
class LoadImageBase64:
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {"required": {"image": ("STRING", {"multiline": False})}}
|
|
|
|
RETURN_TYPES = ("IMAGE", "MASK")
|
|
CATEGORY = "Knodes"
|
|
|
|
FUNCTION = "Proc"
|
|
|
|
def Proc(self, image):
|
|
imgdata = base64.b64decode(image)
|
|
img = Image.open(BytesIO(imgdata))
|
|
|
|
if "A" in img.getbands():
|
|
mask = np.array(img.getchannel("A")).astype(np.float32) / 255.0
|
|
mask = 1.0 - torch.from_numpy(mask)
|
|
else:
|
|
mask = torch.zeros((64, 64), dtype=torch.float32, device="cpu")
|
|
|
|
img = img.convert("RGB")
|
|
img = np.array(img).astype(np.float32) / 255.0
|
|
img = torch.from_numpy(img)[None,]
|
|
|
|
return (img, mask)
|
|
|
|
class LoadImagesBase64:
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {"required": {"strings": ("STRING", {"multiline": False})}}
|
|
|
|
RETURN_TYPES = ("IMAGE","MASK")
|
|
CATEGORY = "Knodes"
|
|
|
|
FUNCTION = "Proc"
|
|
|
|
def Proc(self, strings):
|
|
number_of_images = int(strings[:4].lstrip('0'), 16)
|
|
print("Number Of Images: " + str(number_of_images))
|
|
strings = strings[4:]
|
|
|
|
images = list()
|
|
|
|
for i in range(number_of_images):
|
|
length = int(strings[:8].lstrip('0'), 16)
|
|
print("Image #" + str(i) + " Length: " + str(length))
|
|
strings = strings[8:]
|
|
single_image = strings[:length]
|
|
strings = strings[length:]
|
|
images.append(single_image)
|
|
|
|
tensors = list()
|
|
masks = list()
|
|
|
|
for single_image in images:
|
|
imgdata = base64.b64decode(single_image)
|
|
img = Image.open(BytesIO(imgdata))
|
|
|
|
if "A" in img.getbands():
|
|
mask = np.array(img.getchannel("A")).astype(np.float32) / 255.0
|
|
mask = 1.0 - torch.from_numpy(mask)
|
|
masks.append(mask)
|
|
else:
|
|
mask = torch.zeros((64, 64), dtype=torch.float32, device="cpu")
|
|
masks.append(mask)
|
|
|
|
img = img.convert("RGB")
|
|
img = np.array(img).astype(np.float32) / 255.0
|
|
img = torch.from_numpy(img)[None,]
|
|
|
|
tensors.append(img)
|
|
|
|
return (torch.cat(tensors), torch.cat(masks))
|
|
|
|
NODE_CLASS_MAPPINGS = {
|
|
"Image(s) To Websocket (Base64)": ImageOutput,
|
|
"Load Image (Base64)": LoadImageBase64,
|
|
"Load Images (Base64)": LoadImagesBase64
|
|
}
|
|
|
|
NODE_DISPLAY_NAME_MAPPINGS = {
|
|
"ImageOutput": "Image(s) To Websocket (Base64)",
|
|
"LoadImageBase64": "Load Image (Base64)",
|
|
"LoadImageBase64": "Load Images (Base64)"
|
|
}
|