feat: shortlist builder — read JSON, set thresholds, copy passing photos
This commit is contained in:
Kareem Horstink
@@ -0,0 +1,16 @@
|
||||
@echo off
|
||||
REM Photo Judgers — Shortlist Builder (Windows)
|
||||
REM Reads output.json, asks for thresholds, copies photos to shortlist folder
|
||||
|
||||
echo ========================================
|
||||
echo Photo Judgers — Shortlist Builder
|
||||
echo ========================================
|
||||
echo.
|
||||
|
||||
REM Activate conda env
|
||||
call conda activate photo_judgers
|
||||
|
||||
echo Running shortlist builder...
|
||||
echo.
|
||||
|
||||
python -m src.shortlist %*
|
||||
@@ -0,0 +1,19 @@
|
||||
#!/bin/bash
|
||||
# Photo Judgers — Shortlist Builder (Linux/Mac)
|
||||
# Reads output.json, asks for thresholds, copies photos to shortlist folder
|
||||
|
||||
set -e
|
||||
|
||||
echo "========================================"
|
||||
echo " Photo Judgers — Shortlist Builder"
|
||||
echo "========================================"
|
||||
echo ""
|
||||
|
||||
# Activate conda env
|
||||
source $(conda info --base)/etc/profile.d/conda.sh
|
||||
conda activate photo_judgers
|
||||
|
||||
echo "Running shortlist builder..."
|
||||
echo ""
|
||||
|
||||
python -m src.shortlist "$@"
|
||||
@@ -0,0 +1,265 @@
|
||||
"""
|
||||
Photo Judgers — Shortlist Builder
|
||||
|
||||
Reads output.json, asks for thresholds per scorer,
|
||||
copies photos that pass to a shortlist folder.
|
||||
|
||||
Usage:
|
||||
python -m src.shortlist
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
# Add project root to path
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# IO Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def load_results(output_path: str) -> list[dict]:
|
||||
"""Load results from JSON file."""
|
||||
if not os.path.exists(output_path):
|
||||
print(f" ERROR: File not found: {output_path}")
|
||||
sys.exit(1)
|
||||
try:
|
||||
with open(output_path, encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
if not isinstance(data, list):
|
||||
print(" ERROR: Expected a JSON array.")
|
||||
sys.exit(1)
|
||||
return data
|
||||
except json.JSONDecodeError as e:
|
||||
print(f" ERROR: Invalid JSON: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def ask_path(prompt: str, default: str = "") -> str:
|
||||
"""Prompt user for a path."""
|
||||
while True:
|
||||
answer = input(f"\n {prompt}").strip().strip('"\'')
|
||||
if answer:
|
||||
return answer
|
||||
if default:
|
||||
return default
|
||||
print(" Path cannot be empty.")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Display
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def show_summary(results: list[dict]) -> None:
|
||||
"""Show a summary of scored photos grouped by scorer."""
|
||||
print()
|
||||
print("=" * 50)
|
||||
print(" Scored Photos Summary")
|
||||
print("=" * 50)
|
||||
print(f"\n Total scored: {len(results)} images\n")
|
||||
|
||||
# Group by scorer (subtype)
|
||||
by_scoring: dict[str, list[dict]] = defaultdict(list)
|
||||
for entry in results:
|
||||
subtype = entry.get("subtype", "unknown")
|
||||
by_scoring[subtype].append(entry)
|
||||
|
||||
for subtype, entries in sorted(by_scoring.items()):
|
||||
scores = [e.get("laion_score") for e in entries if "laion_score" in e]
|
||||
if scores:
|
||||
avg = sum(scores) / len(scores)
|
||||
mn = min(scores)
|
||||
mx = max(scores)
|
||||
print(f" {subtype}:")
|
||||
print(f" Count: {len(entries)}")
|
||||
print(f" Score range: {mn:.2f} — {mx:.2f}")
|
||||
print(f" Average: {avg:.2f}")
|
||||
else:
|
||||
print(f" {subtype}: {len(entries)} images (no scores)")
|
||||
print()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Thresholds
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def ask_thresholds(results: list[dict]) -> dict[str, float]:
|
||||
"""Ask user for thresholds per scorer type."""
|
||||
thresholds: dict[str, float] = {}
|
||||
|
||||
# Group by subtype
|
||||
by_scoring: dict[str, list[dict]] = defaultdict(list)
|
||||
for entry in results:
|
||||
subtype = entry.get("subtype", "unknown")
|
||||
by_scoring[subtype].append(entry)
|
||||
|
||||
print()
|
||||
print("=" * 50)
|
||||
print(" Thresholds")
|
||||
print("=" * 50)
|
||||
print("\n Set thresholds for each scorer type.")
|
||||
print(" Photos with scores ABOVE the threshold will be shortlisted.")
|
||||
print(" Press Enter to skip a scorer (won't filter by it).\n")
|
||||
|
||||
for subtype in sorted(by_scoring.keys()):
|
||||
scores = [e.get("laion_score") for e in by_scoring[subtype] if "laion_score" in e]
|
||||
if scores:
|
||||
avg = sum(scores) / len(scores)
|
||||
default_val = f"{avg:.1f}"
|
||||
else:
|
||||
default_val = "5.0"
|
||||
|
||||
answer = input(f" {subtype} threshold [default: {default_val}]: ").strip()
|
||||
if answer:
|
||||
try:
|
||||
thresholds[subtype] = float(answer)
|
||||
except ValueError:
|
||||
print(f" Invalid value, skipping {subtype}.")
|
||||
else:
|
||||
if default_val:
|
||||
try:
|
||||
thresholds[subtype] = float(default_val)
|
||||
print(f" Using default: {thresholds[subtype]}")
|
||||
except ValueError:
|
||||
print(f" No threshold set for {subtype}.")
|
||||
|
||||
return thresholds
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Filtering & Copying
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def filter_photos(
|
||||
results: list[dict], thresholds: dict[str, float]
|
||||
) -> list[dict]:
|
||||
"""Filter photos that pass all thresholds."""
|
||||
kept: list[dict] = []
|
||||
|
||||
for entry in results:
|
||||
subtype = entry.get("subtype", "unknown")
|
||||
score = entry.get("laion_score")
|
||||
|
||||
# If no threshold for this subtype, keep it
|
||||
if subtype not in thresholds:
|
||||
kept.append(entry)
|
||||
continue
|
||||
|
||||
# If no score, skip
|
||||
if score is None:
|
||||
continue
|
||||
|
||||
# Check if score meets threshold
|
||||
if score >= thresholds[subtype]:
|
||||
kept.append(entry)
|
||||
|
||||
return kept
|
||||
|
||||
|
||||
def copy_to_shortlist(
|
||||
photos: list[dict], shortlist_dir: str, dry_run: bool = False
|
||||
) -> tuple[int, int]:
|
||||
"""Copy photos to shortlist directory. Returns (copied, skipped)."""
|
||||
os.makedirs(shortlist_dir, exist_ok=True)
|
||||
|
||||
copied = 0
|
||||
skipped = 0
|
||||
|
||||
for entry in photos:
|
||||
filepath = entry.get("filepath")
|
||||
filename = entry.get("filename", os.path.basename(filepath))
|
||||
|
||||
if not filepath or not os.path.exists(filepath):
|
||||
print(f" Warning: File not found: {filepath}")
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
dest = os.path.join(shortlist_dir, filename)
|
||||
|
||||
# Handle duplicates
|
||||
if os.path.exists(dest):
|
||||
base, ext = os.path.splitext(filename)
|
||||
counter = 1
|
||||
while os.path.exists(dest):
|
||||
dest = os.path.join(shortlist_dir, f"{base}_{counter}{ext}")
|
||||
counter += 1
|
||||
|
||||
if not dry_run:
|
||||
shutil.copy2(filepath, dest)
|
||||
copied += 1
|
||||
|
||||
return copied, skipped
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def main() -> None:
|
||||
print()
|
||||
print("=" * 50)
|
||||
print(" Photo Judgers — Shortlist Builder")
|
||||
print("=" * 50)
|
||||
|
||||
# Load results
|
||||
output_path = ask_path("Output JSON file [output.json]: ", "output.json")
|
||||
results = load_results(output_path)
|
||||
|
||||
# Show summary
|
||||
show_summary(results)
|
||||
|
||||
# Ask for thresholds
|
||||
thresholds = ask_thresholds(results)
|
||||
|
||||
# Ask for shortlist destination
|
||||
shortlist_dir = ask_path("Shortlist folder: ", "shortlist")
|
||||
|
||||
# Confirm
|
||||
print()
|
||||
print("=" * 50)
|
||||
print(" Summary")
|
||||
print("=" * 50)
|
||||
print(f" Results file: {output_path}")
|
||||
print(f" Total scored: {len(results)}")
|
||||
print(f" Shortlist folder: {shortlist_dir}")
|
||||
print(f" Thresholds: {thresholds}")
|
||||
print()
|
||||
|
||||
confirm = input(" Continue? [y/N]: ").strip().lower()
|
||||
if confirm not in ("y", "yes"):
|
||||
print(" Aborted.")
|
||||
return
|
||||
|
||||
# Filter
|
||||
kept = filter_photos(results, thresholds)
|
||||
print(f"\n Photos passing thresholds: {len(kept)}")
|
||||
|
||||
if not kept:
|
||||
print(" No photos pass the thresholds. Nothing to copy.")
|
||||
return
|
||||
|
||||
# Copy
|
||||
print(f"\n Copying to {shortlist_dir}...")
|
||||
copied, skipped = copy_to_shortlist(kept, shortlist_dir)
|
||||
|
||||
print()
|
||||
print(f" Copied: {copied} images")
|
||||
if skipped:
|
||||
print(f" Skipped: {skipped} images")
|
||||
print(f" Shortlist saved to: {shortlist_dir}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user