File size: 1,839 Bytes
e0265b9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | """Small compatibility layer for live denoising previews from local backends."""
from __future__ import annotations
import inspect
from pathlib import Path
from typing import Any, Callable
from adam.executor import ToolContext
def accepts_preview_callback(function: Callable[..., Any]) -> bool:
"""Only opt in to the explicit protocol; ``**kwargs`` is not enough."""
try:
return "preview_callback" in inspect.signature(function).parameters
except (TypeError, ValueError):
return False
def publish_generation_preview(
context: ToolContext, output: Path, payload: Any, *, image_index: int,
image_count: int, step: int = 0, total_steps: int = 0,
) -> None:
"""Persist only the newest preview and publish it to ADAM's live viewer."""
if isinstance(payload, dict):
step = int(payload.get("step", payload.get("current_step", step)) or step)
total_steps = int(payload.get("total_steps", payload.get("steps", total_steps)) or total_steps)
payload = payload.get("image", payload.get("path"))
if not payload:
return
preview_dir = output / ".live_previews"
preview_dir.mkdir(parents=True, exist_ok=True)
destination = preview_dir / f"{context.job_id}_latest.png"
try:
if isinstance(payload, (str, Path)):
source = Path(payload).expanduser().resolve()
if not source.is_file():
return
destination.write_bytes(source.read_bytes())
elif callable(getattr(payload, "save", None)):
payload.save(destination, format="PNG")
else:
return
except (OSError, ValueError, TypeError):
return
context.preview(destination, kind="generation", current=step, total=total_steps,
image_index=image_index + 1, image_count=image_count)
|