144 lines
4.7 KiB
Python
144 lines
4.7 KiB
Python
import requests
|
|
import base64
|
|
import io
|
|
import numpy
|
|
import PIL
|
|
import requests
|
|
import torch
|
|
from collections.abc import Iterable
|
|
from typing import List
|
|
|
|
|
|
def imageurl2tensor(image_urls: List[dict]):
|
|
if not image_urls:
|
|
return torch.zeros((1, 3, 1, 1)) # 返回一个1x3x1x1的黑色tensor
|
|
|
|
print("INFO:", "output image_urls:", image_urls)
|
|
images = []
|
|
for url_dict in image_urls:
|
|
try:
|
|
image_data = fetch_image(url_dict.get("url"))
|
|
image = decode_image(image_data)
|
|
except:
|
|
continue
|
|
images.append(image)
|
|
print(
|
|
"INFO:", f"{len(images)} of {len(image_urls)} output images loaded successfully.")
|
|
if len(images) != len(image_urls):
|
|
print("WARN:", f"{len(image_urls) - len(images)} of {len(image_urls)} output image urls cannot be loaded in GUI.\nPlease check console for image_urls manually.")
|
|
return images2tensor(images)
|
|
|
|
|
|
def fetch_image(url, stream=True):
|
|
return requests.get(url, stream=stream).content
|
|
|
|
|
|
def tensor2images(tensor):
|
|
np_imgs = numpy.clip(tensor.cpu().numpy() * 255.0,
|
|
0.0, 255.0).astype(numpy.uint8)
|
|
return [PIL.Image.fromarray(np_img) for np_img in np_imgs]
|
|
|
|
|
|
def images2tensor(images):
|
|
if isinstance(images, Iterable):
|
|
return torch.stack([torch.from_numpy(numpy.array(image)).float() / 255.0 for image in images])
|
|
return torch.from_numpy(numpy.array(images)).unsqueeze(0).float() / 255.0
|
|
|
|
|
|
def decode_image(data_bytes, rtn_mask=False):
|
|
with io.BytesIO(data_bytes) as bytes_io:
|
|
img = PIL.Image.open(bytes_io)
|
|
if not rtn_mask:
|
|
img = img.convert('RGB')
|
|
elif 'A' in img.getbands():
|
|
img = img.getchannel('A')
|
|
else:
|
|
img = None
|
|
return img
|
|
|
|
|
|
def encode_image(img, mask=None):
|
|
format = "JPEG"
|
|
if mask is not None:
|
|
format = "PNG"
|
|
img = img.copy()
|
|
img.putalpha(mask)
|
|
with io.BytesIO() as bytes_io:
|
|
img.save(bytes_io, format=format)
|
|
data_bytes = bytes_io.getvalue()
|
|
return data_bytes, format
|
|
|
|
|
|
def decorate_base64(base64, format="JPEG"):
|
|
return f"data:image/{format};base64,{base64}"
|
|
|
|
|
|
def image_to_base64(image):
|
|
if image is None:
|
|
return None
|
|
data_bytes, format = encode_image(tensor2images(image)[0])
|
|
return decorate_base64(base64.b64encode(data_bytes).decode("utf-8"), format=format)
|
|
|
|
|
|
def image_to_base64s(tensor):
|
|
if tensor is None:
|
|
return None
|
|
images = tensor2images(tensor)
|
|
data_bytes_list = [encode_image(image) for image in images]
|
|
return [decorate_base64(base64.b64encode(data_bytes).decode("utf-8"), format=format) for data_bytes, format in data_bytes_list]
|
|
# return [base64.b64encode(encode_image(image)).decode("utf-8") for image in images]
|
|
|
|
|
|
def check_lora_path(path):
|
|
"""Checks if the LoRA path is valid."""
|
|
if path.startswith('http://') or path.startswith('https://'):
|
|
return path
|
|
elif '/' in path and not path.startswith('/'):
|
|
# Ensure format is 'username/model-name'
|
|
parts = path.split('/')
|
|
if len(parts) == 2 and all(part.strip() for part in parts):
|
|
return path
|
|
raise ValueError(
|
|
"Invalid LoRA path format. It should be either a full URL or in the format 'username/model-name'.")
|
|
|
|
|
|
def normalization_loras(loras, scale_max, scale_default):
|
|
"""Normalizes and validates a list of LoRA dictionaries."""
|
|
_loras = []
|
|
if not loras:
|
|
return _loras
|
|
for lora in loras:
|
|
if "path" in lora:
|
|
lora_path = lora["path"]
|
|
if lora_path:
|
|
lora_path = lora_path.strip()
|
|
if lora_path:
|
|
lora_scale = lora.get("scale", scale_default)
|
|
if lora_scale < 0 or lora_scale > scale_max:
|
|
raise ValueError(
|
|
f"Invalid {lora_path} LoRA scale. It should be between 0 and {scale_max}.")
|
|
_loras.append({"path": check_lora_path(
|
|
lora_path), "scale": lora_scale})
|
|
|
|
return _loras
|
|
|
|
|
|
class BaseRequest:
|
|
"""Base class for all API request objects."""
|
|
|
|
def build_payload(self):
|
|
"""Builds the request payload dictionary."""
|
|
raise NotImplementedError("Subclasses must implement build_payload")
|
|
|
|
# def get_api_path(self):
|
|
# """Gets the API path for the request."""
|
|
# raise NotImplementedError("Subclasses must implement path")
|
|
|
|
def _remove_empty_fields(self, payload):
|
|
"""Removes None, empty string, and empty dict values from payload."""
|
|
return {k: v for k, v in payload.items() if v is not None and (v != "" and v != {})}
|
|
|
|
def field_order(self):
|
|
"""Get field order"""
|
|
raise NotImplementedError("Subclasses must implement field order")
|