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