""" Persist test uploads and flattened analysis metrics to CSV + JSON. """ from __future__ import annotations import csv import json from datetime import datetime, timezone from pathlib import Path from PIL import Image from .config import LOG_DIR try: import fcntl _HAS_FCNTL = True except ImportError: _HAS_FCNTL = False CSV_NAME = "analysis_log.csv" CSV_COLUMNS = [ "image_name", "saved_filename", "analyzed_at", "lang", "device", "rationale_source", "laion_aesthetic_score", "color_harmony_value", "color_harmony_level", "hue_diversity", "luminance_contrast", "composition_balance_value", "composition_balance_level", "centeredness", "center_offset", "saturation_intensity_value", "saturation_intensity_level", "mean_saturation", "edge_complexity_value", "edge_complexity_level", "edge_density", "warm_cool_value", "warm_cool_level", "warmth_index", "hue_centroid", "depth_foreground_pct", "depth_mid_pct", "depth_background_pct", "depth_spread", "depth_error", ] _IMAGES_DIR = LOG_DIR / "test_images" _JSON_DIR = LOG_DIR / "analysis_json" _CSV_PATH = LOG_DIR / CSV_NAME def ensure_log_dirs() -> None: LOG_DIR.mkdir(parents=True, exist_ok=True) _IMAGES_DIR.mkdir(parents=True, exist_ok=True) _JSON_DIR.mkdir(parents=True, exist_ok=True) def extract_image_name(image) -> str: """Best-effort original filename from Gradio upload payload.""" if isinstance(image, dict): orig = image.get("orig_name") if orig: return Path(str(orig)).name path = image.get("path") if path: return Path(str(path)).name return f"upload_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jpg" def versioned_filename(original_name: str, ts: datetime | None = None) -> str: ts = ts or datetime.now(timezone.utc) path = Path(original_name) stem = path.stem or "upload" stamp = ts.strftime("%Y%m%d_%H%M%S_%f") return f"{stem}_{stamp}.jpg" def _dim_map(detail: dict) -> dict[str, dict]: return {d["id"]: d for d in detail.get("dimensions", []) if "id" in d} def _sub_value(dim: dict, label: str) -> str: for sub in dim.get("sub_metrics", []): if sub.get("label") == label: return str(sub.get("value", "")) return "" def flatten_analysis( detail: dict, lang: str, image_name: str, saved_filename: str, analyzed_at: datetime | None = None, ) -> dict[str, str]: analyzed_at = analyzed_at or datetime.now(timezone.utc) dims = _dim_map(detail) color = dims.get("color_harmony", {}) comp = dims.get("composition_balance", {}) sat = dims.get("saturation_intensity", {}) edge = dims.get("edge_complexity", {}) warm = dims.get("warm_cool", {}) depth = detail.get("depth_layers") or {} centeredness = _sub_value(comp, "Centeredness") or comp.get("value", "") center_offset = _sub_value(comp, "Offset from center") or comp.get("raw", {}).get("offset", "") row = { "image_name": image_name, "saved_filename": saved_filename, "analyzed_at": analyzed_at.isoformat(timespec="seconds"), "lang": lang, "device": str(detail.get("device", "")), "rationale_source": str(detail.get("rationale_source", "")), "laion_aesthetic_score": str(detail.get("laion_aesthetic_score", "")), "color_harmony_value": str(color.get("value", "")), "color_harmony_level": str(color.get("level", "")), "hue_diversity": str(color.get("raw", {}).get("hue_entropy_norm", "")), "luminance_contrast": str(color.get("raw", {}).get("luminance_contrast", "")), "composition_balance_value": str(comp.get("value", "")), "composition_balance_level": str(comp.get("level", "")), "centeredness": str(centeredness), "center_offset": str(center_offset), "saturation_intensity_value": str(sat.get("value", "")), "saturation_intensity_level": str(sat.get("level", "")), "mean_saturation": str(sat.get("raw", {}).get("mean_saturation", "")), "edge_complexity_value": str(edge.get("value", "")), "edge_complexity_level": str(edge.get("level", "")), "edge_density": str(edge.get("raw", {}).get("edge_density", "")), "warm_cool_value": str(warm.get("value", "")), "warm_cool_level": str(warm.get("level", "")), "warmth_index": _sub_value(warm, "Warmth index") or str(warm.get("value", "")), "hue_centroid": str(warm.get("raw", {}).get("hue_centroid", "")), "depth_foreground_pct": str(depth.get("foreground_coverage", "")), "depth_mid_pct": str(depth.get("midground_coverage", "")), "depth_background_pct": str(depth.get("background_coverage", "")), "depth_spread": str(depth.get("depth_spread", "")), "depth_error": str(detail.get("depth_error") or ""), } return {col: row.get(col, "") for col in CSV_COLUMNS} def save_test_image(pil: Image.Image, original_name: str) -> tuple[str, Path]: ensure_log_dirs() saved_filename = versioned_filename(original_name) dest = _IMAGES_DIR / saved_filename pil.convert("RGB").save(dest, format="JPEG", quality=92) return saved_filename, dest def save_analysis_json(saved_filename: str, detail: dict) -> Path: ensure_log_dirs() stem = Path(saved_filename).stem dest = _JSON_DIR / f"{stem}.json" dest.write_text(json.dumps(detail, indent=2, ensure_ascii=False), encoding="utf-8") return dest def append_analysis_row(row: dict[str, str]) -> Path: ensure_log_dirs() write_header = not _CSV_PATH.exists() or _CSV_PATH.stat().st_size == 0 with _CSV_PATH.open("a", encoding="utf-8", newline="") as fh: if _HAS_FCNTL: fcntl.flock(fh.fileno(), fcntl.LOCK_EX) try: writer = csv.DictWriter(fh, fieldnames=CSV_COLUMNS, extrasaction="ignore") if write_header: writer.writeheader() writer.writerow(row) finally: if _HAS_FCNTL: fcntl.flock(fh.fileno(), fcntl.LOCK_UN) return _CSV_PATH def csv_log_path() -> Path | None: return _CSV_PATH if _CSV_PATH.exists() else None def persist_analysis( pil: Image.Image, detail: dict, lang: str, image, ) -> tuple[str, Path, Path]: """Save image, JSON detail, and append CSV row. Returns (saved_filename, image_path, csv_path).""" image_name = extract_image_name(image) saved_filename, image_path = save_test_image(pil, image_name) row = flatten_analysis(detail, lang, image_name, saved_filename) save_analysis_json(saved_filename, detail) csv_path = append_analysis_row(row) return saved_filename, image_path, csv_path