| """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) |
|
|
