|
from typing import Dict, List, Any |
|
from diffusers import StableDiffusionPipeline |
|
import torch |
|
from PIL import Image |
|
import numpy as np |
|
|
|
class EndpointHandler(): |
|
def __init__(self, path=""): |
|
|
|
|
|
|
|
model_id = "selectmixer/SDXL_NIJI_SIX" |
|
self.pipe = StableDiffusionPipeline.from_pretrained(model_id) |
|
self.pipe.to("cuda") |
|
|
|
self.pipe.load_lora_weights("selectmixer/LORA_NIJI_DLCV6_SDXL", weight_name="SDXL_Niji_V6_DLC_LoRa_V2.safetensors", adapter_name="test") |
|
|
|
def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]: |
|
""" |
|
data args: |
|
inputs (:obj: `str` | `PIL.Image` | `np.array`) |
|
kwargs |
|
Return: |
|
A :obj:`list` | `dict`: will be serialized and returned |
|
""" |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
inputs = data.get("inputs") |
|
kwargs = data.get("kwargs", {}) |
|
|
|
|
|
if isinstance(inputs, str): |
|
|
|
prompt = inputs |
|
elif isinstance(inputs, Image.Image): |
|
|
|
prompt = self.pipe.feature_extractor(images=inputs, return_tensors="pt").to("cuda") |
|
elif isinstance(inputs, np.ndarray): |
|
|
|
image = Image.fromarray(inputs) |
|
prompt = self.pipe.feature_extractor(images=image, return_tensors="pt").to("cuda") |
|
else: |
|
raise ValueError("Unsupported input type") |
|
|
|
lora_scale = kwargs.get("lora_scale", 0.5) |
|
|
|
|
|
results = self.pipe(prompt, cross_attention_kwargs={"scale":lora_scale}, **kwargs) |
|
|
|
|
|
output = [{"generated_image": result} for result in results.images] |
|
|
|
return output |
|
|