| |
| """Verify the local FLUX.1-dev ConvRot Hugging Face release package.""" |
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| import math |
| import os |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any |
|
|
| CHUNK_BYTES = 16 * 1024 * 1024 |
| ROOT = Path(__file__).resolve().parents[1] |
| MANIFEST_PATH = ROOT / "release_manifest.json" |
| OUTPUT_PATH = ROOT / "release_verification.json" |
|
|
|
|
| class VerificationError(RuntimeError): |
| pass |
|
|
|
|
| def parse_args() -> argparse.Namespace: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument( |
| "--model-root", |
| type=Path, |
| default=Path(os.environ.get("COMFYUI_MODELS_ROOT", "/mnt/d/comfyui/models")), |
| ) |
| parser.add_argument( |
| "--models", |
| action="store_true", |
| help="Hash all four model artifacts; required before upload.", |
| ) |
| return parser.parse_args() |
|
|
|
|
| def sha256_file(path: Path) -> str: |
| digest = hashlib.sha256() |
| with path.open("rb") as handle: |
| for chunk in iter(lambda: handle.read(CHUNK_BYTES), b""): |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def load_json(path: Path) -> dict[str, Any]: |
| value = json.loads(path.read_text(encoding="utf-8")) |
| if not isinstance(value, dict): |
| raise VerificationError(f"expected JSON object: {path}") |
| return value |
|
|
|
|
| def require(condition: bool, message: str) -> None: |
| if not condition: |
| raise VerificationError(message) |
|
|
|
|
| def verify_regular_file(path: Path) -> None: |
| require(path.exists(), f"missing file: {path}") |
| require(path.is_file(), f"not a regular file: {path}") |
| require(not path.is_symlink(), f"symlink is not allowed: {path}") |
|
|
|
|
| def main() -> int: |
| args = parse_args() |
| manifest = load_json(MANIFEST_PATH) |
| require( |
| manifest.get("status") in {"ready_for_local_verification", "verified_private_uploaded"}, |
| "release status drift", |
| ) |
| require(manifest.get("remote_status") in {"not_created", "private"}, "remote-status drift") |
| required = manifest.get("required_repository_files") |
| artifacts = manifest.get("artifacts") |
| if not isinstance(required, list): |
| raise VerificationError("missing required_repository_files") |
| if not isinstance(artifacts, list) or len(artifacts) != 4: |
| raise VerificationError("expected four artifacts") |
|
|
| support_records: list[dict[str, Any]] = [] |
| for relative in required: |
| path = ROOT / str(relative) |
| verify_regular_file(path) |
| support_records.append( |
| { |
| "path": str(relative), |
| "bytes": path.stat().st_size, |
| "sha256": sha256_file(path), |
| } |
| ) |
|
|
| unexpected_models = sorted(str(path.relative_to(ROOT)) for path in ROOT.rglob("*.safetensors")) |
| require(not unexpected_models, f"model files must not be duplicated into staging: {unexpected_models}") |
|
|
| readme = (ROOT / "README.md").read_text(encoding="utf-8") |
| for marker in ( |
| "license: other", |
| "license_name: flux-1-dev-non-commercial-license", |
| "base_model: black-forest-labs/FLUX.1-dev", |
| "base_model_relation: quantized", |
| "pipeline_tag: text-to-image", |
| "FLUX.1-dev ConvRot for ComfyUI", |
| ): |
| require(marker in readme, f"model-card marker missing: {marker}") |
|
|
| workflow = load_json(ROOT / "workflows/FLUX1_Dev_ConvRot_API.json") |
| require(workflow["1"]["class_type"] == "UNETLoader", "workflow loader drift") |
| require( |
| workflow["1"]["inputs"]["unet_name"] == "FLUX.1-dev-w8a8-convrot.safetensors", |
| "workflow DiT default drift", |
| ) |
| require( |
| workflow["2"]["inputs"]["clip_name2"] == "t5xxl_flux1_int8_convrot.safetensors", |
| "workflow T5 default drift", |
| ) |
|
|
| evidence = manifest["evidence"] |
| overview = ROOT / evidence["overview_contact_sheet"]["repo_path"] |
| require(overview.stat().st_size == evidence["overview_contact_sheet"]["bytes"], "overview byte drift") |
| require(sha256_file(overview) == evidence["overview_contact_sheet"]["sha256"], "overview hash drift") |
|
|
| quality = load_json(ROOT / "assets/quality_metrics.json") |
| require(quality.get("status") == "verified_complete", "quality evidence status drift") |
| require( |
| quality.get("source_grid_manifest_sha256") == evidence["source_grid_manifest_sha256"], |
| "quality source-manifest drift", |
| ) |
| results = quality.get("results") |
| if not isinstance(results, list) or len(results) != 4: |
| raise VerificationError("quality result inventory drift") |
| measured = {str(item["model_id"]): float(item["global_psnr_db"]) for item in results} |
| for model_id, expected in evidence["quality_global_psnr_db"].items(): |
| require(model_id in measured, f"missing quality result: {model_id}") |
| require(math.isclose(measured[model_id], float(expected), rel_tol=0.0, abs_tol=1e-12), f"quality drift: {model_id}") |
|
|
| model_root = args.model_root.expanduser().resolve(strict=True) |
| artifact_records: list[dict[str, Any]] = [] |
| total_bytes = 0 |
| repo_paths: set[str] = set() |
| for artifact in artifacts: |
| repo_path = str(artifact["repo_path"]) |
| require(repo_path not in repo_paths, f"duplicate repo path: {repo_path}") |
| repo_paths.add(repo_path) |
| source = (model_root / str(artifact["source_relative_path"])).resolve(strict=True) |
| require(source.is_file(), f"artifact is not a file: {source}") |
| actual_bytes = source.stat().st_size |
| require(actual_bytes == int(artifact["bytes"]), f"artifact byte drift: {source}") |
| total_bytes += actual_bytes |
| actual_hash = None |
| if args.models: |
| actual_hash = sha256_file(source) |
| require(actual_hash == str(artifact["sha256"]), f"artifact hash drift: {source}") |
| source_manifest_relative = artifact.get("source_manifest_relative_path") |
| if source_manifest_relative is not None: |
| source_manifest = (model_root / str(source_manifest_relative)).resolve(strict=True) |
| verify_regular_file(source_manifest) |
| require( |
| sha256_file(source_manifest) == str(artifact["source_manifest_sha256"]), |
| f"source conversion-manifest drift: {source_manifest}", |
| ) |
| artifact_records.append( |
| { |
| "repo_path": repo_path, |
| "source_relative_path": str(artifact["source_relative_path"]), |
| "bytes": actual_bytes, |
| "sha256": actual_hash if actual_hash is not None else str(artifact["sha256"]), |
| "hash_verified": bool(args.models), |
| } |
| ) |
|
|
| require(total_bytes == int(manifest["artifact_total_bytes"]), "artifact total-byte drift") |
| result = { |
| "schema_version": 1, |
| "status": "PASS", |
| "verified_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), |
| "suggested_repo_id": manifest["suggested_repo_id"], |
| "remote_status": manifest["remote_status"], |
| "model_hashes_verified": bool(args.models), |
| "artifact_count": len(artifact_records), |
| "artifact_total_bytes": total_bytes, |
| "artifacts": artifact_records, |
| "support_file_count": len(support_records), |
| "support_files": support_records, |
| "overview_sha256": evidence["overview_contact_sheet"]["sha256"], |
| "quality_source_grid_manifest_sha256": evidence["source_grid_manifest_sha256"], |
| } |
| temporary = OUTPUT_PATH.with_name(f".{OUTPUT_PATH.name}.{os.getpid()}.tmp") |
| temporary.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8") |
| os.replace(temporary, OUTPUT_PATH) |
| print(json.dumps(result, indent=2, sort_keys=True)) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
