flow_copd/_static_check.py

import py_compile, sys, os
for f in ["flow_copd/copd_loss.py","flow_copd/config_copd.py","flow_copd/train_sd3_copd.py"]:
    py_compile.compile(f, doraise=True); print("OK syntax:", f, flush=True)

import torch
sys.path.insert(0, "flow_copd")
from copd_loss import opd_positive_loss, copd_loss, regen_velocity_per_step, sde_time_weight
B,C,H,W = 4,16,64,64
vs = torch.randn(B,C,H,W, requires_grad=True); vt = torch.randn(B,C,H,W)
x_t = torch.randn(B,C,H,W); x0 = torch.randn(B,C,H,W)
sigma_j = torch.tensor(0.7)
vneg = regen_velocity_per_step(x_t, x0, sigma_j)
wt = sde_time_weight(sigma_j, "snr")
adv = torch.tensor([1.5,-1.2,0.3,-2.0])
lp = opd_positive_loss(vs, vt, 1.0); lp.backward()
print("positive loss:", float(lp), "grad_ok:", vs.grad is not None, flush=True)
vs.grad=None
lc, info = copd_loss(vs, vt, vneg, adv, weight_t=1.0, lambda_neg=1.0, neg_clamp=1.0, adv_thresh=0.0)
lc.backward()
print("contrastive loss:", float(lc), {k: round(float(v),4) for k,v in info.items()}, flush=True)
assert torch.isfinite(lc)
print("neg_frac (expect 0.5):", float(info['neg_frac']), flush=True)

import config_copd as cc
for name in ["text_a0_positive","text_a2_copd","text_a1_symmetric","pick_a2_copd"]:
    cfg = getattr(cc, name)()
    tp = cfg.copd.teacher_lora_path
    ok = os.path.exists(os.path.join(tp,"adapter_model.safetensors"))
    ds = os.path.exists(cfg.dataset)
    print(f"{name:18s} mode={cfg.copd.mode:11s} teacher_ok={ok} dataset_ok={ds} steps={cfg.sample.num_steps} res={cfg.resolution}", flush=True)
print("ALL STATIC CHECKS PASSED", flush=True)