Spaces:
Runtime error
Runtime error
| # 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() |