HARSHIT-hash-07
feat: integrated cloud-based diffusion inference module
17f1f54
Raw
History Blame Contribute Delete
12 kB
# coding: utf-8
import torch
import torch.nn as nn
from torch import Tensor
from typing import Any, Dict, Optional
from encoder import Encoder
from ACD import ACD
from batch import Batch
from embeddings import Embeddings
from vocabulary import Vocabulary
from initialization import initialize_model
from constants import PAD_TOKEN, EOS_TOKEN, BOS_TOKEN, TARGET_PAD, UNK_TOKEN
def _get_stoi_mapping(vocab: Any) -> Dict[str, int]:
"""
Return a token->index mapping for several vocab implementations:
- objects with get_stoi()
- objects with .stoi (dict-like or mapping)
- plain dicts
- objects supporting __getitem__ (try to extract PAD/UNK)
Returns an empty dict on failure.
"""
if vocab is None:
return {}
# 1) custom Vocabulary with get_stoi()
try:
if hasattr(vocab, "get_stoi") and callable(getattr(vocab, "get_stoi")):
m = vocab.get_stoi()
if isinstance(m, dict):
return m
except Exception:
pass
# 2) TorchText-like object with .stoi attribute (dict or mapping)
try:
if hasattr(vocab, "stoi"):
stoi = getattr(vocab, "stoi")
if callable(stoi):
stoi = stoi()
if isinstance(stoi, dict):
return stoi
try:
return dict(stoi)
except Exception:
pass
except Exception:
pass
# 3) If it's a plain dict
try:
if isinstance(vocab, dict):
return vocab
except Exception:
pass
# 4) Try basic __getitem__ access to extract PAD/UNK indices
try:
pad_idx = vocab[PAD_TOKEN]
mapping = {PAD_TOKEN: pad_idx}
try:
mapping[UNK_TOKEN] = vocab[UNK_TOKEN]
except Exception:
pass
return mapping
except Exception:
pass
# fallback empty mapping
return {}
def _get_vocab_size(vocab: Any) -> int:
"""
Determine vocabulary size in a robust way.
Tries len(vocab), len(vocab.stoi), or highest index+1 from stoi dict.
Falls back to 0.
"""
try:
# If vocab supports len()
sz = len(vocab)
return sz
except Exception:
pass
try:
if hasattr(vocab, "stoi"):
stoi = getattr(vocab, "stoi")
if callable(stoi):
stoi = stoi()
if isinstance(stoi, dict):
# assume indices are 0..N-1 or similar
max_idx = max(stoi.values()) if stoi else -1
return max_idx + 1
try:
return len(dict(stoi))
except Exception:
pass
except Exception:
pass
try:
# if it's a dict
if isinstance(vocab, dict):
max_idx = max(vocab.values()) if vocab else -1
return max_idx + 1
except Exception:
pass
return 0
class Model(nn.Module):
def __init__(self, cfg: dict,
encoder: Encoder,
ACD: ACD,
src_embed: Embeddings,
src_vocab: Vocabulary,
trg_vocab: Vocabulary,
in_trg_size: int,
out_trg_size: int):
"""
Create Sign-IDD
"""
super(Model, self).__init__()
self.src_embed = src_embed
self.encoder = encoder
self.ACD = ACD
self.src_vocab = src_vocab
self.trg_vocab = trg_vocab
# robustly obtain stoi mapping
stoi = _get_stoi_mapping(self.src_vocab)
# ensure special tokens exist (give informative error)
for tok in [BOS_TOKEN, PAD_TOKEN, EOS_TOKEN]:
if tok not in stoi:
raise ValueError(
f"Special token '{tok}' missing in vocab. "
f"Available tokens (sample): {list(stoi.keys())[:20]} ..."
)
self.bos_index = stoi[BOS_TOKEN]
self.pad_index = stoi[PAD_TOKEN]
self.eos_index = stoi[EOS_TOKEN]
self.target_pad = TARGET_PAD
self.use_cuda = cfg["training"].get("use_cuda", False) if "training" in cfg else False
self.in_trg_size = in_trg_size
self.out_trg_size = out_trg_size
def forward(self, is_train: bool, src: Tensor, trg_input: Tensor,
src_mask: Tensor, src_lengths: Tensor, trg_mask: Tensor):
""" Encode source, then diffusion decode """
encoder_output = self.encode(src=src,
src_length=src_lengths,
src_mask=src_mask)
diffusion_output = self.diffusion(is_train=is_train,
encoder_output=encoder_output,
trg_input=trg_input,
src_mask=src_mask,
trg_mask=trg_mask)
return diffusion_output
def encode(self, src: Tensor, src_length: Tensor, src_mask: Tensor):
""" Encodes the source sequence """
return self.encoder(embed_src=self.src_embed(src),
src_length=src_length,
mask=src_mask)
def diffusion(self, is_train: bool, encoder_output: Tensor,
src_mask: Tensor, trg_input: Tensor, trg_mask: Tensor):
""" Diffusion decoding """
return self.ACD(is_train=is_train,
encoder_output=encoder_output,
input_3d=trg_input,
src_mask=src_mask,
trg_mask=trg_mask)
def get_loss_for_batch(self, is_train, batch: Batch, loss_function: nn.Module) -> Tensor:
""" Compute batch loss """
skel_out = self.forward(src=batch.src,
trg_input=batch.trg_input[:, :, :150],
src_mask=batch.src_mask,
src_lengths=batch.src_lengths,
trg_mask=batch.trg_mask,
is_train=is_train)
batch_loss = loss_function(skel_out, batch.trg_input[:, :, :150])
return batch_loss
def build_model(cfg: dict, src_vocab: 'Vocabulary', trg_vocab: 'Vocabulary', checkpoint: Optional[dict] = None):
"""
Build and initialize the Sign-IDD model.
Optionally load a checkpoint and resize embeddings if necessary.
"""
# Full configuration
full_cfg = cfg
cfg_model = cfg["model"]
# Padding indices (robust)
src_stoi = _get_stoi_mapping(src_vocab)
src_padding_idx = src_stoi.get(PAD_TOKEN, 0)
if not isinstance(trg_vocab, (list, tuple)):
trg_stoi = _get_stoi_mapping(trg_vocab)
trg_padding_idx = trg_stoi.get(PAD_TOKEN, 0)
else:
trg_padding_idx = 0
in_trg_size = cfg_model["trg_size"]
out_trg_size = cfg_model["trg_size"]
# Determine vocab_size robustly
vocab_size = _get_vocab_size(src_vocab)
if vocab_size == 0:
# best-effort fallback: try to infer from stoi mapping
vocab_size = max(src_stoi.values()) + 1 if src_stoi else 0
# --- Source embedding ---
src_embed = Embeddings(
**cfg_model["encoder"]["embeddings"],
vocab_size=vocab_size,
padding_idx=src_padding_idx
)
# --- Encoder ---
enc_dropout = cfg_model["encoder"].get("dropout", 0.0)
enc_emb_dropout = cfg_model["encoder"]["embeddings"].get("dropout", enc_dropout)
# Transformer-specific check (if your encoder assumes this)
if "embedding_dim" in cfg_model["encoder"]["embeddings"] and "hidden_size" in cfg_model["encoder"]:
assert cfg_model["encoder"]["embeddings"]["embedding_dim"] == cfg_model["encoder"]["hidden_size"], \
"For transformer, embedding_dim must equal hidden_size"
encoder = Encoder(
**cfg_model["encoder"],
emb_size=src_embed.embedding_dim,
emb_dropout=enc_emb_dropout
)
# --- ACD module ---
diffusion = ACD(args=cfg_model, trg_vocab=trg_vocab)
# --- Build the model ---
model = Model(
cfg=full_cfg,
encoder=encoder,
ACD=diffusion,
src_embed=src_embed,
src_vocab=src_vocab,
trg_vocab=trg_vocab,
in_trg_size=in_trg_size,
out_trg_size=out_trg_size
)
# --- Initialize model parameters ---
initialize_model(model, cfg_model, src_padding_idx, trg_padding_idx)
# --- Load checkpoint if provided ---
if checkpoint is not None:
# support both {"model_state": {...}} and raw state dicts
state_dict = checkpoint.get("model_state", None) if isinstance(checkpoint, dict) else None
state_dict = state_dict if state_dict is not None else (checkpoint if isinstance(checkpoint, dict) else None)
if isinstance(state_dict, dict):
# Handle source embedding mismatch safely if attributes exist
# Support both "src_embed.lut.weight" and "src_embed.weight" checkpoint keys
for emb_key in ("src_embed.lut.weight", "src_embed.weight"):
if emb_key in state_dict:
# Check current model embedding weight shape (if available)
curr_emb = None
# try to reach the attribute in a safe manner
lut = getattr(model.src_embed, "lut", None)
if lut is not None and hasattr(lut, "weight"):
curr_emb = lut.weight
elif hasattr(model.src_embed, "weight"):
curr_emb = getattr(model.src_embed, "weight")
if curr_emb is not None:
if state_dict[emb_key].shape != curr_emb.shape:
print(f"Skipping {emb_key} due to shape mismatch")
state_dict.pop(emb_key, None)
# if we couldn't inspect current embedding, leave as-is (load may still work)
try:
model.load_state_dict(state_dict, strict=False)
except RuntimeError as e:
# more informative failure message
print(f"Warning: problem loading state_dict: {e}")
else:
print("Warning: checkpoint provided but no valid state_dict found in provided object")
# Move to GPU if available & desired
if torch.cuda.is_available() and full_cfg.get("training", {}).get("use_cuda", True):
model.to(torch.device("cuda"))
return model
def load_model(checkpoint: Optional[dict], model: nn.Module, src_padding_idx: int):
"""
Load checkpoint into model safely (used if you want a separate utility).
"""
if checkpoint is None:
return model
model_state = checkpoint.get("model_state", checkpoint) if isinstance(checkpoint, dict) else checkpoint
if isinstance(model_state, dict):
# try to handle src_embed mismatch similarly to build_model
for emb_key in ("src_embed.lut.weight", "src_embed.weight"):
if emb_key in model_state:
curr_emb = None
lut = getattr(model.src_embed, "lut", None)
if lut is not None and hasattr(lut, "weight"):
curr_emb = lut.weight
elif hasattr(model.src_embed, "weight"):
curr_emb = getattr(model.src_embed, "weight")
if curr_emb is not None:
if model_state[emb_key].shape != curr_emb.shape:
print(f"[INFO] Checkpoint {emb_key} shape {model_state[emb_key].shape} "
f"does not match current model {curr_emb.shape}. Skipping.")
model_state.pop(emb_key, None)
try:
model.load_state_dict(model_state, strict=False)
except RuntimeError as e:
print(f"[INFO] Failed to load full checkpoint: {e}")
return model