| from __future__ import annotations |
|
|
| import hashlib |
| from pathlib import Path |
| from typing import Any |
|
|
| from PIL import Image, ImageFilter, ImageStat |
|
|
| from adam.models import Job |
|
|
|
|
| IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".bmp"} |
|
|
|
|
| def _candidate_images(job: Job, limit: int = 64) -> list[Path]: |
| found: list[Path] = [] |
| if job.preview_path: |
| preview = Path(job.preview_path) |
| if preview.is_file(): |
| found.append(preview) |
| output = Path(job.output_folder).expanduser() if job.output_folder else None |
| if output is not None and output.is_dir(): |
| try: |
| for path in output.rglob("*"): |
| if len(found) >= limit: |
| break |
| if not path.is_file() or path.suffix.casefold() not in IMAGE_EXTENSIONS: |
| continue |
| lowered = str(path.relative_to(output)).casefold() |
| if any(token in lowered for token in ("preview", "sample", "generation", "epoch")) and path not in found: |
| found.append(path) |
| except OSError: |
| pass |
| return found |
|
|
|
|
| def evaluate_job_output(job: Job) -> dict[str, Any]: |
| """Evaluate technical sample health without claiming to judge artistic quality.""" |
| if not any(step.tool_id.endswith("_trainer") for step in job.plan.steps): |
| return {} |
| paths = _candidate_images(job) |
| hashes: list[str] = [] |
| sharpness: list[float] = [] |
| corrupted: list[str] = [] |
| for path in paths: |
| try: |
| with Image.open(path) as source: |
| image = source.convert("RGB") |
| thumb = image.resize((32, 32)).convert("L") |
| hashes.append(hashlib.sha1(thumb.tobytes()).hexdigest()) |
| edges = image.resize((256, 256)).convert("L").filter(ImageFilter.FIND_EDGES) |
| sharpness.append(float(ImageStat.Stat(edges).var[0])) |
| except (OSError, ValueError): |
| corrupted.append(str(path)) |
| valid = len(hashes) |
| unique = len(set(hashes)) |
| duplicate_count = valid - unique |
| if valid < 4: |
| status = "NEEDS SAMPLES" |
| summary = f"Only {valid} readable training preview(s) were available. Generate a fixed sample set for a meaningful comparison." |
| elif corrupted or duplicate_count / max(valid, 1) >= 0.25: |
| status = "NEEDS REVIEW" |
| summary = f"Reviewed {valid} samples; found {duplicate_count} exact-looking duplicate(s) and {len(corrupted)} unreadable file(s)." |
| else: |
| status = "TECHNICALLY HEALTHY" |
| summary = f"Reviewed {valid} samples with no obvious corruption or exact duplicate collapse. Subject and artistic quality still need human review." |
| return { |
| "agent": "NOVA", |
| "status": status, |
| "summary": summary, |
| "sample_count": valid, |
| "unique_count": unique, |
| "duplicate_count": duplicate_count, |
| "corrupted_count": len(corrupted), |
| "average_edge_variance": round(sum(sharpness) / len(sharpness), 2) if sharpness else None, |
| "sample_paths": [str(path) for path in paths], |
| } |
|
|
