Spaces:

YashKotak
/

devatlas-ai-gpu

Running on Zero

devatlas-ai-gpu / scripts /verify_image_generation.py

Upload Python project

6ef0a95 23 days ago

15.4 kB

"""Prove the Companion image pipeline really is SDXL inpainting over a mask.
Runs the REAL `LocalDiffusionImageProvider` (no HTTP calls, no hosted API) on one
Companion avatar and one or more requests, then checks the contract objectively
instead of trusting the model's good intentions:
  1. the resident pipeline IS `StableDiffusionXLInpaintPipeline` and the model IS
     `diffusers/stable-diffusion-xl-1.0-inpainting-0.1`,
  2. every output is a NEW file on its own per-job path (never the avatar path),
  3. a real MASK was produced for every request - binary (WHITE editable, BLACK
     frozen), the size of the frame, with a non-empty edit region - and written
     beside the output as `<name>.mask.png`,
  4. every pixel the mask left BLACK is byte-for-byte the avatar's own pixel (so
     the face, hair, hands, pose and background are preserved) while the WHITE
     region really changed,
  5. the prompt that reached the model is short, image-specific and inside the
     77-token context of BOTH of SDXL's text encoders, and every required log
     field (MODEL, PIPELINE, DEVICE, MODEL_LOAD_TIME, GENERATION_TIME, IMAGE_SIZE,
     MASK_SIZE, STEPS, GUIDANCE, PROMPT) was emitted.
    python scripts/verify_image_generation.py
    python scripts/verify_image_generation.py --prompt "Change only the clothing"
    python scripts/verify_image_generation.py --load-only
    python scripts/verify_image_generation.py --prompts-file my_prompts.txt
"""

from __future__ import annotations

import argparse
import asyncio
import logging
import os
import sys
import tempfile
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from PIL import Image  # noqa: E402

from app.config.settings import get_settings  # noqa: E402
from app.services import companion_prompt  # noqa: E402
from app.services.providers import LocalDiffusionImageProvider  # noqa: E402
from app.services.providers import image_provider  # noqa: E402

DEFAULT_PROMPTS = [
    "Change only the clothing",
    "Change only the clothing to a blue evening dress",
    "change the background to a garden",
]

# The fields the operator asked to see for every render.
REQUIRED_LOG_FIELDS = (
    "MODEL",
    "PIPELINE",
    "DEVICE",
    "MODEL_LOAD_TIME",
    "GENERATION_TIME",
    "IMAGE_SIZE",
    "MASK_SIZE",
    "STEPS",
    "GUIDANCE",
    "PROMPT",
)

# The memory block the Z-Image load and render must always emit, so "which model,
# at which quantisation, on which device, against how much memory" is answerable
# from the log instead of being inferred from a crash.
REQUIRED_Z_IMAGE_FIELDS = (
    "model",
    "quantization",
    "device",
    "total_ram",
    "free_ram_before",
    "estimated_peak_ram",
    "free_ram_after_load",
    "generation_time",
)


class _Capture(logging.Handler):
    """Keeps every record the provider logs, so the checks can read them."""

    def __init__(self) -> None:
        super().__init__(level=logging.INFO)
        self.messages: list[str] = []

    def emit(self, record: logging.LogRecord) -> None:  # pragma: no cover - trivial
        self.messages.append(record.getMessage())


def _synthetic_portrait(size: int = 512) -> Image.Image:
    """A deterministic stand-in avatar when the caller has no real one.
    Shaped like a head-and-shoulders portrait on a flat backdrop, with a torso
    band in a different colour: a mask built from it can be checked pixel by
    pixel (the head must stay BLACK, the torso must be the editable region).
    """
    from PIL import ImageDraw

    img = Image.new("RGB", (size, size), (30, 40, 60))
    draw = ImageDraw.Draw(img)
    cx, cy, r = size // 2, int(size * 0.34), int(size * 0.14)
    draw.ellipse((cx - r, cy - r, cx + r, cy + r), fill=(214, 168, 140))
    draw.rectangle((cx - int(size * 0.20), cy + r, cx + int(size * 0.20), size - 20),
                   fill=(64, 92, 148))
    draw.rectangle((cx - int(size * 0.26), cy + r, cx - int(size * 0.20), size - 80),
                   fill=(214, 168, 140))
    draw.rectangle((cx + int(size * 0.20), cy + r, cx + int(size * 0.26), size - 80),
                   fill=(214, 168, 140))
    return img


def _resolve_avatar(explicit: str | None) -> Path:
    """Use the given avatar, else a real stored one, else a synthetic portrait."""
    if explicit:
        path = Path(explicit)
        if not path.is_file():
            raise SystemExit(f"avatar not found: {path}")
        return path

    cache = Path(os.environ.get("VIDEO_STORAGE_DIR", "storage")) / "companion" / "avatars"
    if cache.is_dir():
        for candidate in sorted(cache.glob("*")):
            if candidate.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"}:
                return candidate

    fallback = Path(tempfile.gettempdir()) / "verify_image_avatar.png"
    _synthetic_portrait().save(fallback)
    print(f"! no stored avatar found; using synthetic portrait {fallback}")
    return fallback


def _mask_stats(mask: Image.Image) -> tuple[int, int, float]:
    """(editable pixels, total pixels, coverage) of a WHITE/BLACK mask."""
    values = list(mask.convert("L").getdata())
    editable = sum(1 for value in values if value == 255)
    other = sum(1 for value in values if value not in (0, 255))
    if other:
        raise AssertionError(f"the mask is not binary: {other} grey pixels")
    return editable, len(values), editable / max(1, len(values))


def _check_one(
    provider: LocalDiffusionImageProvider,
    index: int,
    prompt: str,
    avatar: Path,
    out_dir: Path,
    capture: _Capture,
) -> bool:
    """Render one request and check the whole contract. True when it passes."""
    print(f"--- [{index}] {prompt}")
    out_path = out_dir / f"verify_{index:02d}.png"
    started = time.perf_counter()
    written = asyncio.run(
        provider.generate_image(
            prompt=prompt,
            character_name="Aria",
            character_details="young woman, shoulder-length dark hair, warm smile",
            aspect_ratio="1:1",
            output_path=str(out_path),
            reference_image_path=str(avatar),
        )
    )
    print(f"    wrote {written} in {time.perf_counter() - started:.1f}s")
    ok = True

    # 1. A new file, never the avatar's own path.
    if os.path.abspath(written) == os.path.abspath(str(avatar)):
        print("FAIL  output path IS the avatar path")
        return False

    # 2. The mask: real, binary, frame-sized, with a region that can change.
    mask_path = Path(provider._mask_path_for(written))
    if not mask_path.is_file():
        print(f"FAIL  no mask was written ({mask_path})")
        return False
    with Image.open(mask_path) as handle:
        mask = handle.convert("L")
    try:
        editable, total, coverage = _mask_stats(mask)
    except AssertionError as exc:
        print(f"FAIL  {exc}")
        return False
    with Image.open(written) as handle:
        result = handle.convert("RGB")
    if mask.size != result.size:
        print(f"FAIL  mask {mask.size} does not match the image {result.size}")
        return False
    if editable == 0:
        print("FAIL  the mask has no editable region - nothing could change")
        return False
    print(
        f"PASS  mask {mask.size[0]}x{mask.size[1]}, {editable}/{total} pixels editable "
        f"({coverage * 100:.1f}%), written to {mask_path}"
    )

    # 3. Preservation: every frozen pixel is the avatar's own pixel.
    base = provider._prepare_reference_image(str(avatar), "1:1")
    if base.size != result.size:
        base = base.resize(result.size, Image.LANCZOS)
    base_pixels = list(base.getdata())
    result_pixels = list(result.getdata())
    frozen = [i for i, value in enumerate(mask.getdata()) if value == 0]
    white = [i for i, value in enumerate(mask.getdata()) if value == 255]
    drifted = [i for i in frozen if result_pixels[i] != base_pixels[i]]
    if drifted:
        print(f"FAIL  {len(drifted)} pixels OUTSIDE the mask changed (of {len(frozen)})")
        ok = False
    else:
        print(f"PASS  all {len(frozen)} frozen pixels are byte-identical to the avatar")
    if all(result_pixels[i] == base_pixels[i] for i in white):
        print("FAIL  the editable region did not change at all")
        ok = False
    else:
        changed = sum(1 for i in white if result_pixels[i] != base_pixels[i])
        print(f"PASS  editable region changed: {changed}/{len(white)} pixels")

    # 4. The prompt: short, image-specific, inside the encoders' context.
    sent = provider._last_prompt
    tokens = companion_prompt.clip_token_count(sent)
    print(f"      prompt ({tokens}/{companion_prompt.CLIP_MAX_TOKENS} CLIP tokens): {sent}")
    if tokens > companion_prompt.CLIP_MAX_TOKENS:
        print("FAIL  the prompt is over the text encoders' context")
        ok = False

    # 5. Every required log field was emitted for this render.
    missing = [
        field for field in REQUIRED_LOG_FIELDS
        if not any(message.startswith(f"{field}:") for message in capture.messages)
    ]
    if missing:
        print(f"FAIL  missing log fields: {', '.join(missing)}")
        ok = False
    else:
        print(f"PASS  log fields present: {', '.join(REQUIRED_LOG_FIELDS)}")
    # 6. The [Z_IMAGE] memory block: which model, at which quantisation, on which
    # device, against how much RAM - and that the quantisation really is the one
    # that was configured, not a heavier or lighter one nobody asked for.
    z_lines = [m for m in capture.messages if m.startswith("[Z_IMAGE]")]
    z_fields = {
        m.split("=", 1)[0][len("[Z_IMAGE] "):]
        for m in z_lines if "=" in m
    }
    missing_z = [f for f in REQUIRED_Z_IMAGE_FIELDS if f not in z_fields]
    if missing_z:
        print(f"FAIL  missing [Z_IMAGE] fields: {', '.join(missing_z)}")
        ok = False
    else:
        print("PASS  [Z_IMAGE] block present: " + ", ".join(REQUIRED_Z_IMAGE_FIELDS))
        for line in z_lines:
            print(f"      {line}")
    capture.messages.clear()
    return ok


async def main_async() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--avatar", help="path to a companion avatar image")
    parser.add_argument(
        "--prompt", action="append", default=None,
        help="one request to run (repeatable); defaults to the clothing test set",
    )
    parser.add_argument("--prompts-file", help="one prompt per line")
    parser.add_argument("--strength", type=float, default=None)
    parser.add_argument("--out", default="storage/verification", help="where to write outputs")
    parser.add_argument(
        "--load-only",
        action="store_true",
        help="resolve the device and build the real pipeline, without rendering",
    )
    args = parser.parse_args()

    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")

    if args.prompts_file:
        prompts = [
            line.strip()
            for line in Path(args.prompts_file).read_text(encoding="utf-8").splitlines()
            if line.strip()
        ]
    elif args.prompt:
        prompts = list(args.prompt)
    else:
        prompts = list(DEFAULT_PROMPTS)

    settings = get_settings()
    avatar = _resolve_avatar(args.avatar)
    out_dir = Path(args.out)
    out_dir.mkdir(parents=True, exist_ok=True)

    provider = LocalDiffusionImageProvider(
        settings.local_image_model_dir,
        # The REAL configured setup, so this harness verifies what the service
        # actually runs - quantisation, device, canvas and text-encoder strategy
        # included - rather than the class defaults.
        model_id=settings.local_image_model_id,
        quantization=settings.local_image_quantization,
        gguf_repo_id=settings.local_image_gguf_repo,
        device=settings.local_image_device,
        text_encoder_offload=settings.local_image_text_encoder_offload,
        steps=settings.local_image_steps,
        strength=args.strength if args.strength is not None else settings.local_image_strength,
        guidance_scale=settings.local_image_guidance,
        width=settings.local_image_width,
        height=settings.local_image_height,
        size=settings.local_image_size,
        max_sequence_length=settings.local_image_max_sequence_length,
        negative_prompt=settings.local_image_negative_prompt,
        threads=settings.local_image_threads,
        use_fp16=settings.local_image_fp16,
    )

    print(f"provider      : {provider.name}")
    print(f"model         : {provider.model_id}")
    print(f"quantization  : {provider._quantization}")
    print(f"device        : {settings.local_image_device}")
    print(f"text encoder  : {settings.local_image_text_encoder_offload}")
    print(f"pipeline      : {image_provider._PIPELINE_CLASS_NAME}")
    print(f"model dir     : {settings.local_image_model_dir} (legacy, unused for loading)")
    print(f"avatar        : {avatar}")
    print(f"prompts       : {len(prompts)}")
    print(f"output dir    : {out_dir.resolve()}\n")

    # The provider must genuinely condition on the avatar, not copy it.
    if not provider.supports_reference_image:
        print("FAIL  provider does not report supports_reference_image=True")
        return 1
    print("PASS  provider reports supports_reference_image=True (avatar is a real input)")

    # Captured BEFORE the gate and the load, so the whole [Z_IMAGE] memory block
    # (including free_ram_after_load) is on record rather than only the render.
    capture = _Capture()
    logging.getLogger("app.services.providers.image_provider").addHandler(capture)

    # The gate must be honest about what this machine can hold, before any work.
    try:
        provider.ensure_ready()
    except RuntimeError as exc:
        print(f"FAIL  this machine cannot run the model:\n      {exc}")
        print(
            "\nThe reported requirement above is exact. There is NO fallback to "
            "Stable Diffusion 1.5 or FLUX: free memory, or run on a host with "
            "more RAM (or a CUDA GPU with ~8 GB free), then re-run."
        )
        return 1

    pipeline = provider._load()
    pipeline_name = type(pipeline).__name__
    if pipeline_name != image_provider._PIPELINE_CLASS_NAME:
        print(f"FAIL  pipeline is {pipeline_name}, not {image_provider._PIPELINE_CLASS_NAME}")
        return 1
    print(f"PASS  pipeline = {pipeline_name}")
    print(f"PASS  model    = {provider.model_id}")
    print(f"PASS  quant    = {provider._quantization}")
    print("PASS  weights  = loaded with from_pretrained() from the Hugging Face cache\n")
    if args.load_only:
        return 0

    failed = 0
    for index, prompt in enumerate(prompts, start=1):
        if not _check_one(provider, index, prompt, avatar, out_dir, capture):
            failed += 1
        print()

    print(f"\nOutputs are in: {out_dir.resolve()}")
    if failed:
        print(f"FAIL  {failed} of {len(prompts)} requests failed their checks")
        return 1
    print(f"PASS  all {len(prompts)} requests preserved the unmasked pixels and "
          "changed only the mask")
    return 0


def main() -> int:
    try:
        return asyncio.run(main_async())
    except Exception as exc:
        print(f"\nERROR: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

	"""Prove the Companion image pipeline really is SDXL inpainting over a mask.

	Runs the REAL `LocalDiffusionImageProvider` (no HTTP calls, no hosted API) on one
	Companion avatar and one or more requests, then checks the contract objectively
	instead of trusting the model's good intentions:

	1. the resident pipeline IS `StableDiffusionXLInpaintPipeline` and the model IS
	`diffusers/stable-diffusion-xl-1.0-inpainting-0.1`,
	2. every output is a NEW file on its own per-job path (never the avatar path),
	3. a real MASK was produced for every request - binary (WHITE editable, BLACK
	frozen), the size of the frame, with a non-empty edit region - and written
	beside the output as `<name>.mask.png`,
	4. every pixel the mask left BLACK is byte-for-byte the avatar's own pixel (so
	the face, hair, hands, pose and background are preserved) while the WHITE
	region really changed,
	5. the prompt that reached the model is short, image-specific and inside the
	77-token context of BOTH of SDXL's text encoders, and every required log
	field (MODEL, PIPELINE, DEVICE, MODEL_LOAD_TIME, GENERATION_TIME, IMAGE_SIZE,
	MASK_SIZE, STEPS, GUIDANCE, PROMPT) was emitted.

	python scripts/verify_image_generation.py
	python scripts/verify_image_generation.py --prompt "Change only the clothing"
	python scripts/verify_image_generation.py --load-only
	python scripts/verify_image_generation.py --prompts-file my_prompts.txt
	"""

	from __future__ import annotations

	import argparse
	import asyncio
	import logging
	import os
	import sys
	import tempfile
	import time
	from pathlib import Path

	sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

	from PIL import Image # noqa: E402

	from app.config.settings import get_settings # noqa: E402
	from app.services import companion_prompt # noqa: E402
	from app.services.providers import LocalDiffusionImageProvider # noqa: E402
	from app.services.providers import image_provider # noqa: E402

	DEFAULT_PROMPTS = [
	"Change only the clothing",
	"Change only the clothing to a blue evening dress",
	"change the background to a garden",
	]

	# The fields the operator asked to see for every render.
	REQUIRED_LOG_FIELDS = (
	"MODEL",
	"PIPELINE",
	"DEVICE",
	"MODEL_LOAD_TIME",
	"GENERATION_TIME",
	"IMAGE_SIZE",
	"MASK_SIZE",
	"STEPS",
	"GUIDANCE",
	"PROMPT",
	)

	# The memory block the Z-Image load and render must always emit, so "which model,
	# at which quantisation, on which device, against how much memory" is answerable
	# from the log instead of being inferred from a crash.
	REQUIRED_Z_IMAGE_FIELDS = (
	"model",
	"quantization",
	"device",
	"total_ram",
	"free_ram_before",
	"estimated_peak_ram",
	"free_ram_after_load",
	"generation_time",
	)


	class _Capture(logging.Handler):
	"""Keeps every record the provider logs, so the checks can read them."""

	def __init__(self) -> None:
	super().__init__(level=logging.INFO)
	self.messages: list[str] = []

	def emit(self, record: logging.LogRecord) -> None: # pragma: no cover - trivial
	self.messages.append(record.getMessage())


	def _synthetic_portrait(size: int = 512) -> Image.Image:
	"""A deterministic stand-in avatar when the caller has no real one.

	Shaped like a head-and-shoulders portrait on a flat backdrop, with a torso
	band in a different colour: a mask built from it can be checked pixel by
	pixel (the head must stay BLACK, the torso must be the editable region).
	"""
	from PIL import ImageDraw

	img = Image.new("RGB", (size, size), (30, 40, 60))
	draw = ImageDraw.Draw(img)
	cx, cy, r = size // 2, int(size * 0.34), int(size * 0.14)
	draw.ellipse((cx - r, cy - r, cx + r, cy + r), fill=(214, 168, 140))
	draw.rectangle((cx - int(size * 0.20), cy + r, cx + int(size * 0.20), size - 20),
	fill=(64, 92, 148))
	draw.rectangle((cx - int(size * 0.26), cy + r, cx - int(size * 0.20), size - 80),
	fill=(214, 168, 140))
	draw.rectangle((cx + int(size * 0.20), cy + r, cx + int(size * 0.26), size - 80),
	fill=(214, 168, 140))
	return img


	def _resolve_avatar(explicit: str \| None) -> Path:
	"""Use the given avatar, else a real stored one, else a synthetic portrait."""
	if explicit:
	path = Path(explicit)
	if not path.is_file():
	raise SystemExit(f"avatar not found: {path}")
	return path

	cache = Path(os.environ.get("VIDEO_STORAGE_DIR", "storage")) / "companion" / "avatars"
	if cache.is_dir():
	for candidate in sorted(cache.glob("*")):
	if candidate.suffix.lower() in {".png", ".jpg", ".jpeg", ".webp"}:
	return candidate

	fallback = Path(tempfile.gettempdir()) / "verify_image_avatar.png"
	_synthetic_portrait().save(fallback)
	print(f"! no stored avatar found; using synthetic portrait {fallback}")
	return fallback


	def _mask_stats(mask: Image.Image) -> tuple[int, int, float]:
	"""(editable pixels, total pixels, coverage) of a WHITE/BLACK mask."""
	values = list(mask.convert("L").getdata())
	editable = sum(1 for value in values if value == 255)
	other = sum(1 for value in values if value not in (0, 255))
	if other:
	raise AssertionError(f"the mask is not binary: {other} grey pixels")
	return editable, len(values), editable / max(1, len(values))


	def _check_one(
	provider: LocalDiffusionImageProvider,
	index: int,
	prompt: str,
	avatar: Path,
	out_dir: Path,
	capture: _Capture,
	) -> bool:
	"""Render one request and check the whole contract. True when it passes."""
	print(f"--- [{index}] {prompt}")
	out_path = out_dir / f"verify_{index:02d}.png"
	started = time.perf_counter()
	written = asyncio.run(
	provider.generate_image(
	prompt=prompt,
	character_name="Aria",
	character_details="young woman, shoulder-length dark hair, warm smile",
	aspect_ratio="1:1",
	output_path=str(out_path),
	reference_image_path=str(avatar),
	)
	)
	print(f" wrote {written} in {time.perf_counter() - started:.1f}s")
	ok = True

	# 1. A new file, never the avatar's own path.
	if os.path.abspath(written) == os.path.abspath(str(avatar)):
	print("FAIL output path IS the avatar path")
	return False

	# 2. The mask: real, binary, frame-sized, with a region that can change.
	mask_path = Path(provider._mask_path_for(written))
	if not mask_path.is_file():
	print(f"FAIL no mask was written ({mask_path})")
	return False
	with Image.open(mask_path) as handle:
	mask = handle.convert("L")
	try:
	editable, total, coverage = _mask_stats(mask)
	except AssertionError as exc:
	print(f"FAIL {exc}")
	return False
	with Image.open(written) as handle:
	result = handle.convert("RGB")
	if mask.size != result.size:
	print(f"FAIL mask {mask.size} does not match the image {result.size}")
	return False
	if editable == 0:
	print("FAIL the mask has no editable region - nothing could change")
	return False
	print(
	f"PASS mask {mask.size[0]}x{mask.size[1]}, {editable}/{total} pixels editable "
	f"({coverage * 100:.1f}%), written to {mask_path}"
	)

	# 3. Preservation: every frozen pixel is the avatar's own pixel.
	base = provider._prepare_reference_image(str(avatar), "1:1")
	if base.size != result.size:
	base = base.resize(result.size, Image.LANCZOS)
	base_pixels = list(base.getdata())
	result_pixels = list(result.getdata())
	frozen = [i for i, value in enumerate(mask.getdata()) if value == 0]
	white = [i for i, value in enumerate(mask.getdata()) if value == 255]
	drifted = [i for i in frozen if result_pixels[i] != base_pixels[i]]
	if drifted:
	print(f"FAIL {len(drifted)} pixels OUTSIDE the mask changed (of {len(frozen)})")
	ok = False
	else:
	print(f"PASS all {len(frozen)} frozen pixels are byte-identical to the avatar")
	if all(result_pixels[i] == base_pixels[i] for i in white):
	print("FAIL the editable region did not change at all")
	ok = False
	else:
	changed = sum(1 for i in white if result_pixels[i] != base_pixels[i])
	print(f"PASS editable region changed: {changed}/{len(white)} pixels")

	# 4. The prompt: short, image-specific, inside the encoders' context.
	sent = provider._last_prompt
	tokens = companion_prompt.clip_token_count(sent)
	print(f" prompt ({tokens}/{companion_prompt.CLIP_MAX_TOKENS} CLIP tokens): {sent}")
	if tokens > companion_prompt.CLIP_MAX_TOKENS:
	print("FAIL the prompt is over the text encoders' context")
	ok = False

	# 5. Every required log field was emitted for this render.
	missing = [
	field for field in REQUIRED_LOG_FIELDS
	if not any(message.startswith(f"{field}:") for message in capture.messages)
	]
	if missing:
	print(f"FAIL missing log fields: {', '.join(missing)}")
	ok = False
	else:
	print(f"PASS log fields present: {', '.join(REQUIRED_LOG_FIELDS)}")
	# 6. The [Z_IMAGE] memory block: which model, at which quantisation, on which
	# device, against how much RAM - and that the quantisation really is the one
	# that was configured, not a heavier or lighter one nobody asked for.
	z_lines = [m for m in capture.messages if m.startswith("[Z_IMAGE]")]
	z_fields = {
	m.split("=", 1)[0][len("[Z_IMAGE] "):]
	for m in z_lines if "=" in m
	}
	missing_z = [f for f in REQUIRED_Z_IMAGE_FIELDS if f not in z_fields]
	if missing_z:
	print(f"FAIL missing [Z_IMAGE] fields: {', '.join(missing_z)}")
	ok = False
	else:
	print("PASS [Z_IMAGE] block present: " + ", ".join(REQUIRED_Z_IMAGE_FIELDS))
	for line in z_lines:
	print(f" {line}")
	capture.messages.clear()
	return ok


	async def main_async() -> int:
	parser = argparse.ArgumentParser(description=__doc__)
	parser.add_argument("--avatar", help="path to a companion avatar image")
	parser.add_argument(
	"--prompt", action="append", default=None,
	help="one request to run (repeatable); defaults to the clothing test set",
	)
	parser.add_argument("--prompts-file", help="one prompt per line")
	parser.add_argument("--strength", type=float, default=None)
	parser.add_argument("--out", default="storage/verification", help="where to write outputs")
	parser.add_argument(
	"--load-only",
	action="store_true",
	help="resolve the device and build the real pipeline, without rendering",
	)
	args = parser.parse_args()

	logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")

	if args.prompts_file:
	prompts = [
	line.strip()
	for line in Path(args.prompts_file).read_text(encoding="utf-8").splitlines()
	if line.strip()
	]
	elif args.prompt:
	prompts = list(args.prompt)
	else:
	prompts = list(DEFAULT_PROMPTS)

	settings = get_settings()
	avatar = _resolve_avatar(args.avatar)
	out_dir = Path(args.out)
	out_dir.mkdir(parents=True, exist_ok=True)

	provider = LocalDiffusionImageProvider(
	settings.local_image_model_dir,
	# The REAL configured setup, so this harness verifies what the service
	# actually runs - quantisation, device, canvas and text-encoder strategy
	# included - rather than the class defaults.
	model_id=settings.local_image_model_id,
	quantization=settings.local_image_quantization,
	gguf_repo_id=settings.local_image_gguf_repo,
	device=settings.local_image_device,
	text_encoder_offload=settings.local_image_text_encoder_offload,
	steps=settings.local_image_steps,
	strength=args.strength if args.strength is not None else settings.local_image_strength,
	guidance_scale=settings.local_image_guidance,
	width=settings.local_image_width,
	height=settings.local_image_height,
	size=settings.local_image_size,
	max_sequence_length=settings.local_image_max_sequence_length,
	negative_prompt=settings.local_image_negative_prompt,
	threads=settings.local_image_threads,
	use_fp16=settings.local_image_fp16,
	)

	print(f"provider : {provider.name}")
	print(f"model : {provider.model_id}")
	print(f"quantization : {provider._quantization}")
	print(f"device : {settings.local_image_device}")
	print(f"text encoder : {settings.local_image_text_encoder_offload}")
	print(f"pipeline : {image_provider._PIPELINE_CLASS_NAME}")
	print(f"model dir : {settings.local_image_model_dir} (legacy, unused for loading)")
	print(f"avatar : {avatar}")
	print(f"prompts : {len(prompts)}")
	print(f"output dir : {out_dir.resolve()}\n")

	# The provider must genuinely condition on the avatar, not copy it.
	if not provider.supports_reference_image:
	print("FAIL provider does not report supports_reference_image=True")
	return 1
	print("PASS provider reports supports_reference_image=True (avatar is a real input)")

	# Captured BEFORE the gate and the load, so the whole [Z_IMAGE] memory block
	# (including free_ram_after_load) is on record rather than only the render.
	capture = _Capture()
	logging.getLogger("app.services.providers.image_provider").addHandler(capture)

	# The gate must be honest about what this machine can hold, before any work.
	try:
	provider.ensure_ready()
	except RuntimeError as exc:
	print(f"FAIL this machine cannot run the model:\n {exc}")
	print(
	"\nThe reported requirement above is exact. There is NO fallback to "
	"Stable Diffusion 1.5 or FLUX: free memory, or run on a host with "
	"more RAM (or a CUDA GPU with ~8 GB free), then re-run."
	)
	return 1

	pipeline = provider._load()
	pipeline_name = type(pipeline).__name__
	if pipeline_name != image_provider._PIPELINE_CLASS_NAME:
	print(f"FAIL pipeline is {pipeline_name}, not {image_provider._PIPELINE_CLASS_NAME}")
	return 1
	print(f"PASS pipeline = {pipeline_name}")
	print(f"PASS model = {provider.model_id}")
	print(f"PASS quant = {provider._quantization}")
	print("PASS weights = loaded with from_pretrained() from the Hugging Face cache\n")
	if args.load_only:
	return 0

	failed = 0
	for index, prompt in enumerate(prompts, start=1):
	if not _check_one(provider, index, prompt, avatar, out_dir, capture):
	failed += 1
	print()

	print(f"\nOutputs are in: {out_dir.resolve()}")
	if failed:
	print(f"FAIL {failed} of {len(prompts)} requests failed their checks")
	return 1
	print(f"PASS all {len(prompts)} requests preserved the unmasked pixels and "
	"changed only the mask")
	return 0


	def main() -> int:
	try:
	return asyncio.run(main_async())
	except Exception as exc:
	print(f"\nERROR: {exc}", file=sys.stderr)
	return 1


	if __name__ == "__main__":
	raise SystemExit(main())