devin-lai's picture
Release v0.4.0: selective MLX model downloads
3da88e3 verified
Raw History Blame Contribute Delete
2.23 kB
"""Isolate VAE tile batching from diffusion drift using one saved final latent."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import mlx.core as mx
from h3_mlx.benchmark import measure
from h3_mlx.compare import image_metrics
from h3_mlx.runtime import configure_memory, environment
from h3_mlx.vae import VAE, to_image
def main():
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("--vae", required=True)
p.add_argument("--trace", required=True)
p.add_argument("--record", required=True)
p.add_argument("--batches", type=int, nargs="+", default=[1, 2, 4, 8, 16, 32])
p.add_argument("--repeats", type=int, default=3)
p.add_argument("--output", default="reports/vae")
a = p.parse_args()
configure_memory()
root = Path(a.output)
root.mkdir(parents=True, exist_ok=True)
data = mx.load(a.trace)
latent = data["latent"]
mx.eval(latent)
del data
record = json.loads(Path(a.record).read_text())
width, height = record["size"]
model = VAE(a.vae)
baseline = model.decode(latent)
mx.eval(baseline)
ref = to_image(baseline, width, height)
trials = []
for batch in a.batches:
if batch < 1:
p.error("tile batches must be positive")
model.tile_batch = batch
timer = measure(lambda: model.decode(latent), warmup=1, repeats=a.repeats)
pixels = model.decode(latent)
image = to_image(pixels, width, height)
image.save(root / f"batch-{batch}.png")
item = {
"batch": batch,
**timer,
**image_metrics(ref, image),
"max_float_error": mx.max(mx.abs(baseline - pixels)).item(),
}
trials.append(item)
print(json.dumps(item), flush=True)
model.clear()
(root / "trials.json").write_text(
json.dumps(
{
"environment": environment(),
"source_trace": a.trace,
"scope": "Same final latent; decoder only, loaded weights and one warmup excluded",
"trials": trials,
},
indent=2,
)
+ "\n"
)
if __name__ == "__main__":
main()