from PIL import Image from PIL import ImageFilter import cv2 import numpy as np import scipy import scipy.signal from scipy.spatial import cKDTree import os from perlin2d import * patch_match_compiled = True try: from PyPatchMatch import patch_match except Exception as e: try: import patch_match except Exception as e: patch_match_compiled = False try: patch_match except NameError: print("patch_match compiling failed, will fall back to edge_pad") patch_match_compiled = False def edge_pad(img, mask, mode=1): if mode == 0: nmask = mask.copy() nmask[nmask > 0] = 1 res0 = 1 - nmask res1 = nmask p0 = np.stack(res0.nonzero(), axis=0).transpose() p1 = np.stack(res1.nonzero(), axis=0).transpose() min_dists, min_dist_idx = cKDTree(p1).query(p0, 1) loc = p1[min_dist_idx] for (a, b), (c, d) in zip(p0, loc): img[a, b] = img[c, d] elif mode == 1: record = {} kernel = [[1] * 3 for _ in range(3)] nmask = mask.copy() nmask[nmask > 0] = 1 res = scipy.signal.convolve2d( nmask, kernel, mode="same", boundary="fill", fillvalue=1 ) res[nmask < 1] = 0 res[res == 9] = 0 res[res > 0] = 1 ylst, xlst = res.nonzero() queue = [(y, x) for y, x in zip(ylst, xlst)] # bfs here cnt = res.astype(np.float32) acc = img.astype(np.float32) step = 1 h = acc.shape[0] w = acc.shape[1] offset = [(1, 0), (-1, 0), (0, 1), (0, -1)] while queue: target = [] for y, x in queue: val = acc[y][x] for yo, xo in offset: yn = y + yo xn = x + xo if 0 <= yn < h and 0 <= xn < w and nmask[yn][xn] < 1: if record.get((yn, xn), step) == step: acc[yn][xn] = acc[yn][xn] * cnt[yn][xn] + val cnt[yn][xn] += 1 acc[yn][xn] /= cnt[yn][xn] if (yn, xn) not in record: record[(yn, xn)] = step target.append((yn, xn)) step += 1 queue = target img = acc.astype(np.uint8) else: nmask = mask.copy() ylst, xlst = nmask.nonzero() yt, xt = ylst.min(), xlst.min() yb, xb = ylst.max(), xlst.max() content = img[yt : yb + 1, xt : xb + 1] img = np.pad( content, ((yt, mask.shape[0] - yb - 1), (xt, mask.shape[1] - xb - 1), (0, 0)), mode="edge", ) return img, mask def perlin_noise(img, mask): lin = np.linspace(0, 5, mask.shape[0], endpoint=False) x, y = np.meshgrid(lin, lin) avg = img.mean(axis=0).mean(axis=0) # noise=[((perlin(x, y)+1)*128+avg[i]).astype(np.uint8) for i in range(3)] noise = [((perlin(x, y) + 1) * 0.5 * 255).astype(np.uint8) for i in range(3)] noise = np.stack(noise, axis=-1) # mask=skimage.measure.block_reduce(mask,(8,8),np.min) # mask=mask.repeat(8, axis=0).repeat(8, axis=1) # mask_image=Image.fromarray(mask) # mask_image=mask_image.filter(ImageFilter.GaussianBlur(radius = 4)) # mask=np.array(mask_image) nmask = mask.copy() # nmask=nmask/255.0 nmask[mask > 0] = 1 img = nmask[:, :, np.newaxis] * img + (1 - nmask[:, :, np.newaxis]) * noise # img=img.astype(np.uint8) return img, mask def gaussian_noise(img, mask): noise = np.random.randn(mask.shape[0], mask.shape[1], 3) noise = (noise + 1) / 2 * 255 noise = noise.astype(np.uint8) nmask = mask.copy() nmask[mask > 0] = 1 img = nmask[:, :, np.newaxis] * img + (1 - nmask[:, :, np.newaxis]) * noise return img, mask def cv2_telea(img, mask): ret = cv2.inpaint(img, 255 - mask, 5, cv2.INPAINT_TELEA) return ret, mask def cv2_ns(img, mask): ret = cv2.inpaint(img, 255 - mask, 5, cv2.INPAINT_NS) return ret, mask def patch_match_func(img, mask): ret = patch_match.inpaint(img, mask=255 - mask, patch_size=3) return ret, mask def mean_fill(img, mask): avg = img.mean(axis=0).mean(axis=0) img[mask < 1] = avg return img, mask """ Apache-2.0 license https://github.com/hafriedlander/stable-diffusion-grpcserver/blob/main/sdgrpcserver/services/generate.py https://github.com/parlance-zz/g-diffuser-bot/tree/g-diffuser-bot-beta2 _handleImageAdjustment """ try: from sd_grpcserver.sdgrpcserver import images import torch from math import sqrt def handleImageAdjustment(array, adjustments): tensor = images.fromPIL(Image.fromarray(array)) for adjustment in adjustments: which = adjustment[0] if which == "blur": sigma = adjustment[1] direction = adjustment[2] if direction == "DOWN" or direction == "UP": orig = tensor repeatCount=256 sigma /= sqrt(repeatCount) for _ in range(repeatCount): tensor = images.gaussianblur(tensor, sigma) if direction == "DOWN": tensor = torch.minimum(tensor, orig) else: tensor = torch.maximum(tensor, orig) else: tensor = images.gaussianblur(tensor, adjustment.blur.sigma) elif which == "invert": tensor = images.invert(tensor) elif which == "levels": tensor = images.levels(tensor, adjustment[1], adjustment[2], adjustment[3], adjustment[4]) elif which == "channels": tensor = images.channelmap(tensor, [adjustment.channels.r, adjustment.channels.g, adjustment.channels.b, adjustment.channels.a]) elif which == "rescale": self.unimp("Rescale") elif which == "crop": tensor = images.crop(tensor, adjustment.crop.top, adjustment.crop.left, adjustment.crop.height, adjustment.crop.width) return np.array(images.toPIL(tensor)[0]) except: pass def g_diffuser(img,mask): adjustments=[["blur",32,"UP"],["level",0,0.05,0,1]] mask=handleImageAdjustment(mask,adjustments) out_mask=handleImageAdjustment(mask,adjustments) return img, mask, out_mask def dummy_fill(img,mask): return img,mask functbl = { "gaussian": gaussian_noise, "perlin": perlin_noise, "edge_pad": edge_pad, "patchmatch": patch_match_func if patch_match_compiled else edge_pad, "cv2_ns": cv2_ns, "cv2_telea": cv2_telea, "g_diffuser": g_diffuser, "g_diffuser_lib": dummy_fill, } try: from postprocess import PhotometricCorrection correction_func = PhotometricCorrection() except Exception as e: print(e, "so PhotometricCorrection is disabled") class DummyCorrection: def __init__(self): self.backend="" pass def run(self,a,b,**kwargs): return b correction_func=DummyCorrection() if "taichi" in correction_func.backend: import sys import io import base64 from PIL import Image def base64_to_pil(base64_str): data = base64.b64decode(str(base64_str)) pil = Image.open(io.BytesIO(data)) return pil def pil_to_base64(out_pil): out_buffer = io.BytesIO() out_pil.save(out_buffer, format="PNG") out_buffer.seek(0) base64_bytes = base64.b64encode(out_buffer.read()) base64_str = base64_bytes.decode("ascii") return base64_str from subprocess import Popen, PIPE, STDOUT class SubprocessCorrection: def __init__(self): self.backend=correction_func.backend self.child= Popen(["python", "postprocess.py"], stdin=PIPE, stdout=PIPE, stderr=STDOUT) def run(self,img_input,img_inpainted,mode): if mode=="disabled": return img_inpainted base64_str_input = pil_to_base64(img_input) base64_str_inpainted = pil_to_base64(img_inpainted) try: if self.child.poll(): self.child= Popen(["python", "postprocess.py"], stdin=PIPE, stdout=PIPE, stderr=STDOUT) self.child.stdin.write(f"{base64_str_input},{base64_str_inpainted},{mode}\n".encode()) self.child.stdin.flush() out = self.child.stdout.readline() base64_str=out.decode().strip() while base64_str and base64_str[0]=="[": print(base64_str) out = self.child.stdout.readline() base64_str=out.decode().strip() ret=base64_to_pil(base64_str) except: print("[PIE] not working, photometric correction is disabled") ret=img_inpainted return ret correction_func = SubprocessCorrection()