"""第 42 章：真实 SD-Turbo 条件生成。默认核对已保存记录；--run 才加载模型。""" from argparse import ArgumentParser from hashlib import sha256 import json from pathlib import Path import platform import struct from time import perf_counter MODEL = "stabilityai/sd-turbo" REVISION = "b261bac6fd2cf515557d5d0707481eafa0485ec2" DEFAULT = Path(__file__).parent / "data" / "sd-turbo" / "report.json" PROMPT = "A studio photograph of one {} ceramic mug on a plain white table, soft natural light." CASES = [ {"id": f"{color}-{seed}-1", "prompt": PROMPT.format(color), "seed": seed, "steps": 1} for seed in [2026, 2027] for color in ["red", "blue"] ] + [ {"id": "red-2026-4", "prompt": PROMPT.format("red"), "seed": 2026, "steps": 4}, {"id": "three-tea-2026-1", "prompt": ( "A studio photograph of exactly three red ceramic mugs on a white table, " "each labeled TEA in clear black letters."), "seed": 2026, "steps": 1}, ] def file_hash(path): digest = sha256() with Path(path).open("rb") as source: for chunk in iter(lambda: source.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def validate(path=DEFAULT): report = json.loads(path.read_text()) assert report["model"] == MODEL and report["revision"] == REVISION assert report["source_sha256"] == file_hash(__file__), "记录对应的脚本版本已改变" assert report["cases"] == CASES and len(report["outputs"]) == len(CASES) for case, row in zip(CASES, report["outputs"]): assert all(row[k] == v for k, v in case.items()) filename = path.parent / row["image"] assert filename.name == case["id"] + ".png" assert file_hash(filename) == row["image_sha256"] header = filename.read_bytes()[:24] assert header[:8] == b"\x89PNG\r\n\x1a\n" assert struct.unpack(">II", header[16:24]) == (512, 512) assert len(row["timesteps"]) == case["steps"] and row["seconds"] > 0 for seed in [2026, 2027]: group = [r for r in report["outputs"] if r["seed"] == seed] assert len({r["noise_sha256"] for r in group}) == 1 assert len({r["initial_latent_sha256"] for r in group}) == 1 return report def run(folder): # 所有模型推理依赖放在显式运行分支，默认检查只需标准库。 import accelerate import diffusers import numpy as np import torch import transformers from diffusers import StableDiffusionPipeline from huggingface_hub import snapshot_download report_path = folder / "report.json" if report_path.exists() or any(folder.glob("*.png")): raise ValueError("输出目录已有实验结果，请换一个空目录以保留原始记录") folder.mkdir(parents=True, exist_ok=True) torch.set_num_threads(4) torch.use_deterministic_algorithms(True) snapshot = Path(snapshot_download( MODEL, revision=REVISION, allow_patterns=[ "model_index.json", "scheduler/*.json", "text_encoder/config.json", "text_encoder/model.fp16.safetensors", "tokenizer/*", "unet/config.json", "unet/diffusion_pytorch_model.fp16.safetensors", "vae/config.json", "vae/diffusion_pytorch_model.fp16.safetensors", "README.md", "LICENSE.md"])) pipeline = StableDiffusionPipeline.from_pretrained( snapshot, variant="fp16", torch_dtype=torch.float32, local_files_only=True, use_safetensors=True).to("cpu") pipeline.set_progress_bar_config(disable=True) report = { "model": MODEL, "revision": REVISION, "cases": CASES, "outputs": [], "source_sha256": file_hash(__file__), "environment": {"python": platform.python_version(), "platform": platform.platform(), "torch": torch.__version__, "numpy": np.__version__, "diffusers": diffusers.__version__, "transformers": transformers.__version__, "accelerate": accelerate.__version__, "device": "cpu", "dtype": "float32", "threads": 4}, "settings": {"height": 512, "width": 512, "guidance_scale": 0.0, "variant": "fp16", "negative_prompt": None}, "scheduler": type(pipeline.scheduler).__name__, "scheduler_config": dict(pipeline.scheduler.config), "weights": [{"file": str(p.relative_to(snapshot)), "bytes": p.stat().st_size, "sha256": file_hash(p)} for p in sorted(snapshot.rglob("*.safetensors"))], "license_sha256": file_hash(snapshot / "LICENSE.md"), "parameters": {name: sum(p.numel() for p in getattr(pipeline, name).parameters()) for name in ["text_encoder", "unet", "vae"]}, } def tensor_hash(tensor): return sha256(tensor.contiguous().numpy().tobytes()).hexdigest() for case in CASES: generator = torch.Generator(device="cpu").manual_seed(case["seed"]) noise = torch.randn((1, 4, 64, 64), generator=generator, dtype=torch.float32) pipeline.scheduler.set_timesteps(case["steps"]) initial = noise * pipeline.scheduler.init_noise_sigma started = perf_counter() with torch.inference_mode(): result = pipeline(prompt=case["prompt"], height=512, width=512, num_inference_steps=case["steps"], guidance_scale=0.0, latents=noise.clone(), generator=generator) elapsed = perf_counter() - started filename = folder / (case["id"] + ".png") result.images[0].save(filename) row = {**case, "image": filename.name, "image_sha256": file_hash(filename), "noise_sha256": tensor_hash(noise), "initial_latent_sha256": tensor_hash(initial), "init_noise_sigma": float(pipeline.scheduler.init_noise_sigma), "timesteps": pipeline.scheduler.timesteps.tolist(), "seconds": elapsed, "prompt_tokens": pipeline.tokenizer(case["prompt"]).input_ids, "nsfw_content_detected": result.nsfw_content_detected} report["outputs"].append(row) report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n") print(f"{case['id']}：完成，{elapsed:.2f} 秒", flush=True) validate(report_path) return report def show(report): print(f"模型：{report['model']}；revision={report['revision']}") print("已保存 6 次真实生成：512×512，CPU float32，guidance_scale=0") for row in report["outputs"]: print(f"{row['id']}：{row['steps']} 步，seed={row['seed']}，图像指纹通过") print("同一种子各组的起始噪声一致；默认模式只核对记录，不重新生成。") if __name__ == "__main__": parser = ArgumentParser(description=__doc__) parser.add_argument("--run", action="store_true", help="下载/加载固定模型并实际生成") parser.add_argument("--out", type=Path, help="真实生成使用的空输出目录") parser.add_argument("--record", type=Path, default=DEFAULT, help="默认核对的 JSON 路径") args = parser.parse_args() if args.run and args.out is None: parser.error("--run 需要 --out 指定一个空目录") show(run(args.out) if args.run else validate(args.record))