Hand_Written / app.py
keysun89's picture
Update app.py
36404e7 verified
Raw
History Blame Contribute Delete
23.5 kB
# import spaces
import os
import math
import random
import gradio as gr
import torch
from torch import nn
from torch.nn import functional as F
from PIL import Image, ImageDraw, ImageFont
import torchvision.transforms as transforms
from torch.amp import autocast
from diffusers import AutoencoderKL
from huggingface_hub import hf_hub_download
from indic_transliteration import sanscript
from indic_transliteration.sanscript import transliterate
from torchvision.models import mobilenet_v2, MobileNet_V2_Weights
# CONFIG
CKPT_REPO_ID = "keysun89/HW_Hindi_Model"
CKPT_FILENAME = "ckpt_epoch_434.pt"
VAE_PATH = "runwayml/stable-diffusion-v1-5"
FONT_PATH = "NotoSansDevanagari-Regular.ttf"
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
TIMESTEPS = 1000
betas = torch.linspace(1e-4, 0.02, TIMESTEPS, device=device)
alphas = 1.0 - betas
alpha_bars = torch.cumprod(alphas, dim=0)
class Resblock(nn.Module):
def __init__(self, in_channels, out_channels):
super().__init__()
groups_in = min(32, in_channels) if in_channels >= 8 else in_channels
groups_out = min(32, out_channels) if out_channels >= 8 else out_channels
self.groupnorm_1 = nn.GroupNorm(groups_in, in_channels)
self.conv1 = nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1)
self.groupnorm_2 = nn.GroupNorm(groups_out, out_channels)
self.conv2 = nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1)
self.conv3 = nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1)
self.residual_layer = (nn.Identity() if in_channels == out_channels
else nn.Conv2d(in_channels, out_channels, kernel_size=1, padding=0))
def forward(self, x):
residual = x
x = self.groupnorm_1(x); x = F.silu(x); x = self.conv1(x)
x = self.groupnorm_2(x); x = F.silu(x); x = self.conv2(x)
x = self.conv3(x)
return x + self.residual_layer(residual)
class SelfAttention(nn.Module):
def __init__(self, channels, n_heads=8):
super().__init__()
self.n_heads = n_heads
self.d_head = channels // n_heads
self.qkv = nn.Linear(channels, channels * 3)
self.proj = nn.Linear(channels, channels)
def forward(self, x):
b, c, h, w = x.shape
x_flat = x.view(b, c, h * w).transpose(1, 2)
qkv = self.qkv(x_flat)
q, k, v = qkv.chunk(3, dim=-1)
q = q.view(b, -1, self.n_heads, self.d_head).transpose(1, 2)
k = k.view(b, -1, self.n_heads, self.d_head).transpose(1, 2)
v = v.view(b, -1, self.n_heads, self.d_head).transpose(1, 2)
attn = torch.softmax(q @ k.transpose(-1, -2) / math.sqrt(self.d_head), dim=-1)
out = attn @ v
out = out.transpose(1, 2).reshape(b, h * w, c)
out = self.proj(out)
return out.transpose(1, 2).reshape(b, c, h, w)
class SelfAttentionBlock(nn.Module):
def __init__(self, channels):
super().__init__()
groups = min(32, channels) if channels >= 8 else channels
self.norm = nn.GroupNorm(groups, channels)
self.attn = SelfAttention(channels)
def forward(self, x):
return x + self.attn(self.norm(x))
class StyleProxyHead(nn.Module):
"""Kept only for checkpoint state_dict compatibility (strict=False loading);
not used in the inference forward path."""
def __init__(self, in_channels=64, embed_dim=512, mode="column", mask_ratio=0.5):
super().__init__()
self.mode = mode
self.mask_ratio = mask_ratio
self.pool = nn.AdaptiveAvgPool2d((1, 1))
self.fc = nn.Linear(in_channels, embed_dim)
def forward(self, feat_map):
x = self.pool(feat_map).flatten(1)
x = self.fc(x)
return F.normalize(x, p=2, dim=1)
class Ver_Style(nn.Module):
def __init__(self):
super().__init__()
self.conv1 = nn.Conv2d(512, 256, kernel_size=3, padding=1)
self.res1 = Resblock(256, 256)
self.res2 = Resblock(256, 64)
self.attn1 = SelfAttentionBlock(64)
self.attn2 = SelfAttentionBlock(64)
self.attn3 = SelfAttentionBlock(64)
self.res3 = Resblock(64, 64)
def forward(self, x):
x = self.conv1(x); x = self.res1(x); x = self.res2(x)
x = self.attn1(x); x = self.attn2(x); x = self.attn3(x)
return self.res3(x)
class Hor_Style(nn.Module):
def __init__(self):
super().__init__()
self.conv1 = nn.Conv2d(512, 256, kernel_size=3, padding=1)
self.res1 = Resblock(256, 256)
self.res2 = Resblock(256, 64)
self.attn1 = SelfAttentionBlock(64)
self.attn2 = SelfAttentionBlock(64)
self.attn3 = SelfAttentionBlock(64)
self.res3 = Resblock(64, 64)
def forward(self, x):
x = self.conv1(x); x = self.res1(x); x = self.res2(x)
x = self.attn1(x); x = self.attn2(x); x = self.attn3(x)
return self.res3(x)
class MobileNetStride8Backbone(nn.Module):
def __init__(self, out_channels=512):
super().__init__()
mnet = mobilenet_v2(weights=MobileNet_V2_Weights.DEFAULT)
self.features = mnet.features[:7]
self.project = nn.Conv2d(32, out_channels, kernel_size=1)
def forward(self, x):
return self.project(self.features(x))
class StyleEncoder(nn.Module):
def __init__(self, embed_dim=512, mask_ratio=0.5):
super().__init__()
self.backbone = MobileNetStride8Backbone(out_channels=512)
self.ver = Ver_Style()
self.hor = Hor_Style()
self.ver_proxy_head = StyleProxyHead(64, embed_dim, mode="column", mask_ratio=mask_ratio)
self.hor_proxy_head = StyleProxyHead(64, embed_dim, mode="row", mask_ratio=mask_ratio)
self.pool = nn.AdaptiveAvgPool2d((1, 1))
self.fc = nn.Linear(512, embed_dim)
def forward(self, x):
feat = self.backbone(x)
ver_map = self.ver(feat)
hor_map = self.hor(feat)
ver_emb = self.ver_proxy_head(ver_map)
hor_emb = self.hor_proxy_head(hor_map)
global_emb = self.pool(feat).flatten(1)
global_emb = F.normalize(self.fc(global_emb), p=2, dim=1)
return ver_map, hor_map, ver_emb, hor_emb, global_emb
class UnifontTextEncoder(nn.Module):
def __init__(self, font_path=FONT_PATH, image_size=(64, 1024)):
super().__init__()
self.font_path = font_path
self.image_size = image_size
self.transform = transforms.Compose([
transforms.Resize(image_size),
transforms.Grayscale(num_output_channels=3),
transforms.ToTensor(),
transforms.Normalize([0.5] * 3, [0.5] * 3),
])
try:
self.font = ImageFont.truetype(font_path, 28)
except Exception:
self.font = ImageFont.load_default()
def render_text(self, text):
H, W = self.image_size
img = Image.new('RGB', (W, H), (255, 255, 255))
draw = ImageDraw.Draw(img)
font_size, margin, min_font_size = 28, 20, 8
font = self._load_font(font_size)
while font_size > min_font_size:
bbox = draw.textbbox((0, 0), text, font=font)
if (bbox[2] - bbox[0]) <= (W - margin):
break
font_size -= 2
font = self._load_font(font_size)
bbox = draw.textbbox((0, 0), text, font=font)
text_h, text_w = bbox[3] - bbox[1], bbox[2] - bbox[0]
x = max(0, (W - text_w) // 2 - bbox[0])
y = max(0, (H - text_h) // 2 - bbox[1])
draw.text((x, y), text, font=font, fill=(0, 0, 0))
return img
def _load_font(self, font_size):
try:
return ImageFont.truetype(self.font_path, font_size, layout_engine=ImageFont.Layout.RAQM)
except Exception:
try:
return ImageFont.truetype(self.font_path, font_size)
except Exception:
return self.font
def forward(self, texts, device=None):
imgs = torch.stack([self.transform(self.render_text(t)) for t in texts])
return imgs.to(device) if device is not None else imgs
class ContentEncoder(nn.Module):
def __init__(self, font_path=FONT_PATH):
super().__init__()
self.unifont = UnifontTextEncoder(font_path)
mnet = mobilenet_v2(weights=MobileNet_V2_Weights.DEFAULT)
self.backbone = mnet.features[:7]
self.proj = nn.Conv2d(32, 64, kernel_size=1)
self.attn_head = nn.Sequential(
SelfAttentionBlock(64), SelfAttentionBlock(64), SelfAttentionBlock(64),
Resblock(64, 64),
)
self.pos_embed = nn.Parameter(torch.zeros(1, 64, 1, 128))
def forward(self, text):
dev = next(self.parameters()).device
x = self.unifont(text, device=dev)
x = self.backbone(x)
x = self.proj(x)
x = x + self.pos_embed
return self.attn_head(x)
class CrossAttention_unet(nn.Module):
def __init__(self, ch_1, ch_2=64):
super().__init__()
self.ch_1 = ch_1
self.q = nn.Linear(ch_1, ch_1)
self.k = nn.Linear(ch_2, ch_1)
self.v = nn.Linear(ch_2, ch_1)
self.proj = nn.Linear(ch_1, ch_1)
def forward(self, x, cond):
is_4d = x.dim() == 4
if is_4d:
b, c, h, w = x.shape
x = x.view(b, c, h * w).transpose(1, 2)
if cond.dim() == 4:
cond = cond.flatten(2).transpose(1, 2)
q, k, v = self.q(x), self.k(cond), self.v(cond)
attn = torch.softmax(q @ k.transpose(1, 2) / math.sqrt(self.ch_1), dim=-1)
out = self.proj(attn @ v)
if is_4d:
out = out.transpose(1, 2).view(b, c, h, w)
return out
class Blender(nn.Module):
def __init__(self):
super().__init__()
self.cross_ver = CrossAttention_unet(64, 64)
self.self_ver_1 = SelfAttentionBlock(64)
self.self_ver_2 = SelfAttentionBlock(64)
self.cross_hor = CrossAttention_unet(64, 64)
self.self_hor_1 = SelfAttentionBlock(64)
self.self_hor_2 = SelfAttentionBlock(64)
self.res_out = Resblock(64, 64)
def forward(self, Q, S_ver, S_hor):
cond = self.cross_ver(Q, S_ver)
cond = self.self_ver_1(cond)
cond = self.self_ver_2(cond)
cond = self.cross_hor(cond, S_hor)
cond = self.self_hor_1(cond)
cond = self.self_hor_2(cond)
cond = cond + Q * 0.5
return self.res_out(cond)
class TimeEmbedding(nn.Module):
def __init__(self, dim):
super().__init__()
self.dim = dim
def forward(self, t):
half = self.dim // 2
freqs = torch.exp(-math.log(10000) * torch.arange(half, device=t.device) / (half - 1))
args = t[:, None] * freqs[None, :]
return torch.cat([torch.sin(args), torch.cos(args)], dim=1)
class UNET_ResidualBlock(nn.Module):
def __init__(self, in_channels, out_channels, time_dim=512):
super().__init__()
self.groupnorm_feature = nn.GroupNorm(32, in_channels)
self.conv_feature = nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1)
self.groupnorm_merged = nn.GroupNorm(32, out_channels)
self.conv_merged = nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1)
self.time_proj = nn.Linear(time_dim, out_channels)
self.residual_layer = (nn.Identity() if in_channels == out_channels
else nn.Conv2d(in_channels, out_channels, kernel_size=1, padding=0))
def forward(self, feature, time):
residue = feature
feature = self.groupnorm_feature(feature); feature = F.silu(feature); feature = self.conv_feature(feature)
time = F.silu(time); time = self.time_proj(time)
merged = feature + time.unsqueeze(-1).unsqueeze(-1)
merged = self.groupnorm_merged(merged); merged = F.silu(merged); merged = self.conv_merged(merged)
return merged + self.residual_layer(residue)
class UNET_AttentionBlock(nn.Module):
def __init__(self, channels, cond_dim=64, geglu_mult=2):
super().__init__()
self.conv_input = nn.Conv2d(channels, channels, kernel_size=1, padding=0)
self.layernorm_1 = nn.LayerNorm(channels)
self.attention_1 = SelfAttention(channels)
self.layernorm_2 = nn.LayerNorm(channels)
self.attention_2 = CrossAttention_unet(channels, cond_dim)
self.layernorm_3 = nn.LayerNorm(channels)
hidden = geglu_mult * channels
self.linear_geglu_1 = nn.Linear(channels, hidden * 2)
self.linear_geglu_2 = nn.Linear(hidden, channels)
self.conv_output = nn.Conv2d(channels, channels, kernel_size=1, padding=0)
def forward(self, x, cond):
residue_long = x
x = self.conv_input(x)
b, c, h, w = x.shape
x = x.view((b, c, h * w)).transpose(1, 2)
residue_short = x
x = self.layernorm_1(x)
x_spatial = x.transpose(1, 2).reshape(b, c, h, w)
x_spatial = self.attention_1(x_spatial)
x = x_spatial.reshape(b, c, h * w).transpose(1, 2)
x += residue_short
residue_short = x
x = self.layernorm_2(x)
x = self.attention_2(x, cond)
x += residue_short
residue_short = x
x = self.layernorm_3(x)
x, gate = self.linear_geglu_1(x).chunk(2, dim=-1)
x = x * F.gelu(gate)
x = self.linear_geglu_2(x)
x += residue_short
x = x.transpose(1, 2).view((b, c, h, w))
return self.conv_output(x) + residue_long
class SwitchSequential(nn.Sequential):
def forward(self, x, cond, time):
for layer in self:
if isinstance(layer, UNET_AttentionBlock):
x = layer(x, cond)
elif isinstance(layer, UNET_ResidualBlock):
x = layer(x, time)
else:
x = layer(x)
return x
class width_Upsample(nn.Module):
def __init__(self, channels):
super().__init__()
self.conv = nn.Conv2d(channels, channels, kernel_size=3, padding=1)
def forward(self, x):
return self.conv(F.interpolate(x, scale_factor=(1.0, 2.0), mode='nearest'))
class width_Downsample(nn.Module):
def __init__(self, channels):
super().__init__()
self.conv = nn.Conv2d(channels, channels, kernel_size=3, padding=1, stride=(1, 2))
def forward(self, x):
return self.conv(x)
class Height_Upsample(nn.Module):
def __init__(self, channels):
super().__init__()
self.conv = nn.Conv2d(channels, channels, kernel_size=3, padding=1)
def forward(self, x):
return self.conv(F.interpolate(x, scale_factor=(2.0, 1.0), mode='nearest'))
class Height_Downsample(nn.Module):
def __init__(self, channels):
super().__init__()
self.conv = nn.Conv2d(channels, channels, kernel_size=3, padding=1, stride=(2, 1))
def forward(self, x):
return self.conv(x)
class Unet(nn.Module):
def __init__(self, c1=224, c2=448, c3=896, time_dim=896, cond_dim=64, geglu_mult=2):
super().__init__()
def RB(i, o): return UNET_ResidualBlock(i, o, time_dim=time_dim)
def AB(c): return UNET_AttentionBlock(c, cond_dim, geglu_mult=geglu_mult)
self.encoders = nn.ModuleList([
SwitchSequential(nn.Conv2d(4, c1, kernel_size=3, padding=1)),
SwitchSequential(RB(c1, c1), AB(c1)),
SwitchSequential(width_Downsample(c1)),
SwitchSequential(RB(c1, c2), AB(c2)),
SwitchSequential(width_Downsample(c2)),
SwitchSequential(RB(c2, c3), AB(c3)),
SwitchSequential(width_Downsample(c3)),
SwitchSequential(Height_Downsample(c3)),
SwitchSequential(RB(c3, c3), AB(c3)),
])
self.bottleneck = SwitchSequential(RB(c3, c3), AB(c3), RB(c3, c3))
self.decoders = nn.ModuleList([
SwitchSequential(RB(c3 * 2, c3), AB(c3)),
SwitchSequential(RB(c3 * 2, c3), Height_Upsample(c3)),
SwitchSequential(RB(c3 * 2, c3), AB(c3), width_Upsample(c3)),
SwitchSequential(RB(c3 * 2, c2), AB(c2)),
SwitchSequential(RB(c2 * 2, c2), width_Upsample(c2)),
SwitchSequential(RB(c2 * 2, c1), AB(c1)),
SwitchSequential(RB(c1 * 2, c1), width_Upsample(c1)),
SwitchSequential(RB(c1 * 2, c1), AB(c1)),
SwitchSequential(RB(c1 * 2, c1)),
])
def forward(self, x, cond, time):
skip_connections = []
for layers in self.encoders:
x = layers(x, cond, time)
skip_connections.append(x)
x = self.bottleneck(x, cond, time)
for layers in self.decoders:
x = torch.cat((x, skip_connections.pop()), dim=1)
x = layers(x, cond, time)
return x
class GlobalEmbeddingHead(nn.Module):
def __init__(self, in_channels=4, embed_dim=512):
super().__init__()
self.pool = nn.AdaptiveAvgPool2d((1, 1))
self.fc = nn.Linear(in_channels, embed_dim)
def forward(self, x):
x = self.pool(x).flatten(1)
return F.normalize(self.fc(x), p=2, dim=1)
class UNET_OutputLayer(nn.Module):
def __init__(self, in_channels, out_channels):
super().__init__()
self.groupnorm = nn.GroupNorm(32, in_channels)
self.conv = nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1)
def forward(self, x):
x = self.groupnorm(x); x = F.silu(x)
return self.conv(x)
class Diffusion(nn.Module):
def __init__(self, font_path=FONT_PATH):
super().__init__()
self.style_encoder = StyleEncoder()
self.content_encoder = ContentEncoder(font_path)
self.blender = Blender()
self.time_embedding = TimeEmbedding(896)
self.unet = Unet()
self.final = UNET_OutputLayer(224, 4)
self.embedding_head = GlobalEmbeddingHead(in_channels=4, embed_dim=512)
# LOAD MODEL + VAE (runs once at Space startup)
def _load_partial(module, state_dict, name):
result = module.load_state_dict(state_dict, strict=False)
if result.missing_keys:
print(f"[{name}] missing keys: {result.missing_keys}")
if result.unexpected_keys:
print(f"[{name}] unexpected keys: {result.unexpected_keys}")
print("Downloading checkpoint from HF Hub...")
ckpt_path = hf_hub_download(repo_id=CKPT_REPO_ID, filename=CKPT_FILENAME)
print("Loading VAE...")
vae = AutoencoderKL.from_pretrained(VAE_PATH, subfolder="vae").to(device)
vae.eval()
for p in vae.parameters():
p.requires_grad = False
print("Loading Diffusion model...")
model = Diffusion(font_path=FONT_PATH).to(device)
ckpt = torch.load(ckpt_path, map_location=device, weights_only=False)
_load_partial(model.style_encoder, ckpt["style_encoder"], "style_encoder")
_load_partial(model.content_encoder, ckpt["content_encoder"], "content_encoder")
_load_partial(model.blender, ckpt["blender"], "blender")
_load_partial(model.unet, ckpt["unet"], "unet")
_load_partial(model.final, ckpt["final"], "final")
if "embedding_head" in ckpt:
_load_partial(model.embedding_head, ckpt["embedding_head"], "embedding_head")
model.eval()
print(f"[LOADED] checkpoint from epoch {ckpt.get('epoch', '?')}")
# INFERENCE
# @spaces.GPU(duration=120) # bump duration since up to 1000 steps may need more time
def generate(style_image, text, seed, num_steps, progress=gr.Progress()):
if style_image is None:
raise gr.Error("Please upload a style reference image.")
if not text or not text.strip():
raise gr.Error("Please enter some text to render.")
seed = int(seed) if seed not in (None, "") else random.randint(0, 999999)
torch.manual_seed(seed)
random.seed(seed)
display_text = text
if text.isascii():
display_text = transliterate(text.lower(), sanscript.ITRANS, sanscript.DEVANAGARI)
transform = transforms.Compose([
transforms.Resize((64, 1024)),
transforms.ToTensor(),
transforms.Normalize([0.5] * 3, [0.5] * 3),
])
style_img = transform(style_image.convert("RGB")).unsqueeze(0).to(device)
progress(0, desc="Encoding style and text...")
with torch.no_grad():
with autocast(device_type=device.type):
ver_map, hor_map, _, _, _ = model.style_encoder(style_img)
Q = model.content_encoder([display_text])
cond = model.blender(Q, ver_map, hor_map)
latent = torch.randn(1, 4, 8, 128, device=device)
NUM_DDIM_STEPS = int(num_steps)
ddim_timesteps = torch.linspace(TIMESTEPS - 1, 0, NUM_DDIM_STEPS, dtype=torch.long, device=device)
for i in range(len(ddim_timesteps)):
t = ddim_timesteps[i].item()
t_prev = ddim_timesteps[i + 1].item() if i + 1 < len(ddim_timesteps) else -1
t_tensor = torch.tensor([t], device=device)
with autocast(device_type=device.type):
time_emb = model.time_embedding(t_tensor)
pred_noise = model.unet(latent, cond, time_emb)
pred_noise = model.final(pred_noise)
alpha_bar_t = alpha_bars[t]
alpha_bar_t_prev = alpha_bars[t_prev] if t_prev >= 0 else torch.tensor(1.0, device=device)
x0_pred = (latent - torch.sqrt(1 - alpha_bar_t) * pred_noise) / torch.sqrt(alpha_bar_t)
x0_pred = torch.clamp(x0_pred, -3.0, 3.0)
latent = torch.sqrt(alpha_bar_t_prev) * x0_pred + torch.sqrt(1 - alpha_bar_t_prev) * pred_noise
progress((i + 1) / NUM_DDIM_STEPS, desc=f"Step {i + 1}/{NUM_DDIM_STEPS}")
latent = latent / 0.18215
with autocast(device_type=device.type):
img = vae.decode(latent).sample
progress(1.0, desc="Done")
img = (img.clamp(-1, 1) + 1) / 2
img = img.squeeze(0).permute(1, 2, 0).cpu().numpy()
img = (img * 255).astype("uint8")
return Image.fromarray(img), display_text, seed
# GRADIO UI
with gr.Blocks(title="DiffBrush — Hindi Handwriting Generation") as demo:
gr.Markdown(
"""
# DiffBrush — Devanagari Handwriting Generation
Upload a handwriting style reference image and enter text
(English/ITRANS transliteration or native Devanagari).
"""
)
with gr.Row():
with gr.Column():
style_input = gr.Image(label="Style reference image", type="pil")
text_input = gr.Textbox(
label="Text to render",
placeholder="Type in English (ITRANS) or Devanagari, e.g. 'namaste' or 'नमस्ते'",
)
seed_input = gr.Textbox(label="Seed (optional)", placeholder="Leave blank for random")
steps_input = gr.Slider(
minimum=100, maximum=1000, value=100, step=50,
label="DDIM sampling steps (higher = slower, potentially better quality)",
)
submit_btn = gr.Button("Generate", variant="primary")
with gr.Column():
output_image = gr.Image(label="Generated handwriting")
transliterated_text = gr.Textbox(label="Text used (after transliteration)", interactive=False)
used_seed = gr.Textbox(label="Seed used", interactive=False)
submit_btn.click(
fn=generate,
inputs=[style_input, text_input, seed_input, steps_input],
outputs=[output_image, transliterated_text, used_seed],
)
if __name__ == "__main__":
demo.launch()