File size: 3,098 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 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | 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],
}
|