diff --git a/ProjectStructure.md b/ProjectStructure.md index 40f498a..e7330ce 100644 --- a/ProjectStructure.md +++ b/ProjectStructure.md @@ -39,6 +39,7 @@ ComfyUI-Ultimate-Sampler-Grid/ ├── generation_orchestrator.py # The "Conductor" (Loop Logic, 2281 lines) ├── image_generation.py # KSampler & VAE Wrappers (748 lines) ├── image_metrics.py # Optional per-image metrics (sharpness/entropy/...), run context, scorer-pack adapters +├── metrics_runner.py # Background job: compute metrics for an existing session (CPU, manifest lock) ├── model_loader.py # Checkpoint/LoRA/VAE/GGUF Loading Logic (805 lines) ├── model_cache.py # 3-Tier Model Caching System (900 lines) ├── remote_vae.py # Remote VAE Offloading (340 lines) @@ -191,6 +192,9 @@ All routes registered via `PromptServer.instance.routes`. | POST | `/configbuilder/lookup_lora_metadata` | `config_builder_node.py` | 768 | Full CivitAI metadata for LoRA `{lora_name}` | | POST | `/configbuilder/lookup_model_metadata` | `config_builder_node.py` | 916 | Full CivitAI metadata for model `{model_name, model_type}` | | GET | `/configbuilder/model_lists` | `config_builder_node.py` | 1029 | All model/sampler/scheduler/VAE/upscale lists | +| POST | `/config_tester/compute_metrics` | `__init__.py` | - | Start the background metrics job for a session (body: `{session_name, overwrite?}`) | +| GET | `/config_tester/metrics_status` | `__init__.py` | - | Job progress (`?job_id=`) | +| POST | `/config_tester/cancel_metrics` | `__init__.py` | - | Cancel a metrics job (body: `{job_id}`) | | GET | `/configbuilder/metrics_info` | `config_builder_node.py` | - | Built-in metric names + optional scorer-pack install status (Run Settings hint) | | POST | `/configbuilder/refresh_models` | `config_builder_node.py` | 1093 | Signal frontend cache clear | | GET | `/configbuilder/get_lora_triggers` | `config_builder_node.py` | 1115 | Get triggers for single LoRA | diff --git a/README.md b/README.md index 57a2263..06d9c35 100644 --- a/README.md +++ b/README.md @@ -798,6 +798,7 @@ Notes: * Items produced by remote Distributed workers inside the companion plugin get no metrics yet; master-side saves and upscales do. * Optional scorer packs (aesthetic / reward models) plug in through `nodes.NODE_CLASS_MAPPINGS` without any import, and their scores are stored next to the built-ins. The shipped entries stay disabled until their node ids are verified against the real packs, so only the built-in metrics are available today. The Builder lists each pack with an installed / not installed hint (text only, nothing is downloaded). * **In the Dashboard:** once a session has metrics, the Sort popup gains one option per metric (descending, items without the metric sort last, in both the primary and secondary lists), the Logic Filter field list gains a **Metrics** group, and a **Metric range** row adds `>= min` / `<= max` filters in one click (placeholders show the observed range). The label panel has a **Metrics** checkbox that prints `Shp:812 Ent:7.41 Col:41.2 ...` on each card. Logic filters skip items that lack the field, so images without metrics (older items, videos) stay visible; Quick Filter is categorical and does not list metrics, use the Logic Filter for ranges. Sessions without metrics show none of this. +* **Older sessions:** the cog panel's **IMAGE METRICS** section has **Compute metrics for this session**, a background job (CPU only, no VRAM, about 50 ms per image) that measures every image without metrics, skips videos, shows progress and can be cancelled by clicking the button again. It rewrites only the `metrics` field under the manifest lock, so favorites and notes you change meanwhile survive. Do not run it on a session that is still generating. * The setting is session-level (like GPU Cooldown), so it is NOT a per-config field: it lives in `_session_settings.metrics`, not in `_EXPECTED_CONFIG_KEYS` or `expand_configs`. ## 🖥️ Dashboard Interface diff --git a/__init__.py b/__init__.py index 2a6478c..b3f6e81 100644 --- a/__init__.py +++ b/__init__.py @@ -1119,6 +1119,44 @@ async def cancel_upscale(request): except Exception as e: return web.json_response({"error": str(e)}, status=500) +# --- Image metrics for an EXISTING session (background job, CPU only) --- +@server.PromptServer.instance.routes.post("/config_tester/compute_metrics") +async def compute_metrics(request): + """Start a job that computes image metrics for the session's items that have none.""" + try: + data = await request.json() + session_name = re.sub(r'[^\w\-]', '', str(data.get("session_name") or "")) + if not session_name: + return web.json_response({"error": "Missing session_name"}, status=400) + base_dir = os.path.join(folder_paths.get_output_directory(), "benchmarks", session_name) + if not _is_path_within(base_dir, _get_benchmarks_base()): + return web.json_response({"error": "Forbidden: path outside benchmarks directory"}, status=403) + from .metrics_runner import start_metrics_job, get_metrics_status + job_id, error = start_metrics_job(session_name, overwrite=bool(data.get("overwrite", False))) + if error: + return web.json_response({"error": error}, status=400) + return web.json_response({"job_id": job_id, "total_images": get_metrics_status(job_id)["total"]}) + except Exception as e: + print(f"[ConfigTester] Error starting metrics job: {e}") + return web.json_response({"error": str(e)}, status=500) + +@server.PromptServer.instance.routes.get("/config_tester/metrics_status") +async def metrics_status(request): + job_id = request.query.get("job_id", "") + if not job_id: + return web.json_response({"error": "Missing job_id"}, status=400) + from .metrics_runner import get_metrics_status + return web.json_response(get_metrics_status(job_id)) + +@server.PromptServer.instance.routes.post("/config_tester/cancel_metrics") +async def cancel_metrics(request): + try: + data = await request.json() + from .metrics_runner import cancel_metrics_job + return web.json_response({"cancelled": cancel_metrics_job(data.get("job_id", ""))}) + except Exception as e: + return web.json_response({"error": str(e)}, status=500) + def _create_image_links(source_dir, link_dir, items): """ diff --git a/metrics_runner.py b/metrics_runner.py new file mode 100644 index 0000000..159965f --- /dev/null +++ b/metrics_runner.py @@ -0,0 +1,166 @@ +""" +Phase 2 of the image-metrics feature: "Compute metrics for this session". + +A background job (same shape as upscale_runner: job dict + daemon thread + +status/cancel helpers) that measures the items of an EXISTING session that have +no `metrics` yet. Pure CPU (PIL -> numpy), no VRAM, never touches the GPU. + +Manifest writes happen under manifest_utils._manifest_lock: the manifest is +RELOADED from disk, the new `metrics` are set by item id, and the file is +replaced atomically, so a stale full copy is never written back and a user's +concurrent favorites/notes survive. Do not run it on a session that a +generation run is currently writing: that run saves its own in-memory copy of +the manifest and would drop metrics written meanwhile. +""" +import json +import os +import re +import threading +import uuid + +import numpy as np +from PIL import Image + +from . import image_metrics +from . import manifest_utils + +_jobs = {} +FLUSH_EVERY = 25 + + +def _image_path_for(item, images_dir): + """Resolve the item's image file INSIDE images_dir (None for videos/sentinels/anything else).""" + if item.get("media_type") == "video" or item.get("florence2_no_detection"): + return None + name = item.get("filename") + if not name: + m = re.search(r'[?&]filename=([^&]+)', str(item.get("file") or "")) + name = m.group(1) if m else None + if not name or name != os.path.basename(name) or "%" in name: + return None + if not name.lower().endswith((".webp", ".png", ".jpg", ".jpeg")): + return None + root = os.path.realpath(images_dir) + path = os.path.realpath(os.path.join(root, name)) + if os.path.dirname(path) != root or not os.path.isfile(path): + return None + return path + + +def start_metrics_job(session_name, overwrite=False, keys=None): + """Start the job. Returns (job_id, error). session_name must already be sanitized by the caller.""" + import folder_paths + benchmarks_dir = os.path.realpath(os.path.join(folder_paths.get_output_directory(), "benchmarks")) + session_dir = os.path.realpath(os.path.join(benchmarks_dir, session_name)) + if os.path.dirname(session_dir) != benchmarks_dir: + return None, "Invalid session" + manifest_path = os.path.join(session_dir, "manifest.json") + images_dir = os.path.join(session_dir, "images") + if not os.path.exists(manifest_path): + return None, "Manifest not found" + + spec = image_metrics.normalize_spec({"enabled": True, "builtins": list(keys or image_metrics.BUILTIN_METRICS)}) + if not spec: + return None, "No valid metric names" + + with manifest_utils._manifest_lock: + with open(manifest_path, "r", encoding="utf-8") as f: + items = json.load(f).get("items", []) + targets = [it["id"] for it in items if "id" in it and (overwrite or not it.get("metrics")) + and _image_path_for(it, images_dir)] + if not targets: + return None, "Nothing to compute (every image already has metrics, or there are no images)" + for jid, j in list(_jobs.items()): + if j["status"] == "running" and j["session"] == session_name: + return None, "A metrics job is already running for this session" + + job_id = f"metrics_{uuid.uuid4().hex[:12]}" + job = {"id": job_id, "session": session_name, "status": "running", "completed": 0, "total": len(targets), + "failed": 0, "error": None, "cancel": threading.Event()} + _jobs[job_id] = job + threading.Thread(target=_run_metrics_thread, + args=(job, targets, spec["builtins"], overwrite, manifest_path, images_dir, session_name), + daemon=True).start() + return job_id, None + + +def get_metrics_status(job_id): + job = _jobs.get(job_id) + if not job: + return {"status": "not_found"} + return {k: job[k] for k in ("status", "completed", "total", "failed", "error")} + + +def cancel_metrics_job(job_id): + job = _jobs.get(job_id) + if job and job["status"] == "running": + job["cancel"].set() + return True + return False + + +def _flush(manifest_path, results, overwrite): + """Merge `results` ({id: metrics}) into the manifest on disk under the lock; atomic replace.""" + if not results: + return + with manifest_utils._manifest_lock: + with open(manifest_path, "r", encoding="utf-8") as f: + data = json.load(f) + for it in data.get("items", []): + m = results.get(it.get("id")) + if m and (overwrite or not it.get("metrics")): + it["metrics"] = m + tmp = manifest_path + ".metrics.tmp" + with open(tmp, "w", encoding="utf-8") as f: + json.dump(data, f, indent=4) + os.replace(tmp, manifest_path) + + +def _send_progress(job, session_name): + try: + from server import PromptServer + PromptServer.instance.send_sync("ultimate_grid.metrics_progress", { + "session_name": session_name, "job_id": job["id"], "status": job["status"], + "completed": job["completed"], "total": job["total"], + }) + except Exception: + pass + + +def _run_metrics_thread(job, target_ids, keys, overwrite, manifest_path, images_dir, session_name): + pending = {} + try: + with manifest_utils._manifest_lock: + with open(manifest_path, "r", encoding="utf-8") as f: + by_id = {it.get("id"): it for it in json.load(f).get("items", [])} + for tid in target_ids: + if job["cancel"].is_set(): + break + it = by_id.get(tid) + path = _image_path_for(it, images_dir) if it else None + try: + if path is None: + raise ValueError("no image file") + with Image.open(path) as im: + arr = np.asarray(im.convert("RGB")) + m = image_metrics.compute_metrics_numpy(arr, keys) + if m: + pending[tid] = m + else: + job["failed"] += 1 + except Exception as e: + job["failed"] += 1 + print(f"[GridTester] ⚠️ metrics skipped for item {tid}: {e}") + job["completed"] += 1 + if len(pending) >= FLUSH_EVERY: + _flush(manifest_path, pending, overwrite) + pending = {} + _send_progress(job, session_name) + _flush(manifest_path, pending, overwrite) + job["status"] = "cancelled" if job["cancel"].is_set() else "complete" + print(f"[GridTester] 📏 metrics job {job['status']}: {job['completed']}/{job['total']} images, {job['failed']} skipped") + except Exception as e: + job["status"] = "error" + job["error"] = str(e) + print(f"[GridTester] ❌ metrics job failed: {e}") + _send_progress(job, session_name) diff --git a/resources/logic_ui.js b/resources/logic_ui.js index b1b2313..f2e8709 100644 --- a/resources/logic_ui.js +++ b/resources/logic_ui.js @@ -2777,4 +2777,56 @@ function showAnalyticsModal(title, results, type) { overlay.addEventListener('click', (e) => { if (e.target === overlay) overlay.remove(); }); -} \ No newline at end of file +} + + +// ============================================================ +// IMAGE METRICS FOR AN EXISTING SESSION (background job, CPU only) +// ============================================================ +var _sessionMetricsJob = { id: null, timer: null }; + +async function startSessionMetrics() { + const btn = document.getElementById('compute-metrics-btn'); + const out = document.getElementById('compute-metrics-progress'); + if (_sessionMetricsJob.id) { + // Second click while running = cancel + try { + await fetch('/config_tester/cancel_metrics', { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ job_id: _sessionMetricsJob.id }) + }); + } catch (e) { /* ignore */ } + if (out) out.textContent = 'Cancelling...'; + return; + } + const sessionName = document.getElementById('session-input')?.value || ''; + if (!sessionName) { alert('No session selected'); return; } + if (out) { out.style.display = 'block'; out.textContent = 'Starting...'; } + try { + const resp = await fetch('/config_tester/compute_metrics', { + method: 'POST', headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ session_name: sessionName }) + }); + const result = await resp.json(); + if (result.error) { if (out) out.textContent = result.error; return; } + _sessionMetricsJob.id = result.job_id; + if (btn) btn.textContent = 'CANCEL (click again)'; + _sessionMetricsJob.timer = setInterval(async () => { + try { + const st = await (await fetch('/config_tester/metrics_status?job_id=' + _sessionMetricsJob.id)).json(); + if (out) out.textContent = 'Measuring ' + st.completed + '/' + st.total + '...'; + if (st.status !== 'running') { + clearInterval(_sessionMetricsJob.timer); + _sessionMetricsJob = { id: null, timer: null }; + if (btn) btn.textContent = 'COMPUTE METRICS FOR THIS SESSION'; + if (out) out.textContent = st.status === 'error' ? ('Error: ' + (st.error || 'unknown')) + : (st.status === 'cancelled' ? 'Cancelled at ' : 'Done: ') + st.completed + '/' + st.total + + (st.failed ? ' (' + st.failed + ' skipped)' : '') + (st.status === 'complete' ? '. Reloading...' : ''); + if (st.status === 'complete' && typeof loadSession === 'function') loadSession(); + } + } catch (e) { /* ignore poll errors */ } + }, 1500); + } catch (e) { + if (out) out.textContent = 'Failed to start: ' + e; + } +} diff --git a/resources/template.html b/resources/template.html index abe0921..fdf824c 100644 --- a/resources/template.html +++ b/resources/template.html @@ -232,6 +232,20 @@ + +
+
IMAGE METRICS
+ + +
+ Adds sortable/filterable metrics to older sessions. Skips videos and images that already have metrics. Do not run it while this session is still generating. +
+
+
EXPORT FAVORITES
diff --git a/tests/test_metrics_runner.py b/tests/test_metrics_runner.py new file mode 100644 index 0000000..428b638 --- /dev/null +++ b/tests/test_metrics_runner.py @@ -0,0 +1,132 @@ +"""Phase 2: background 'compute metrics for this session' job.""" +import importlib +import importlib.util +import json +import os +import sys +import time + +import numpy as np +import pytest +from PIL import Image + +_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +_PKG = os.path.basename(_ROOT) +mr = importlib.import_module(f"{_PKG}.metrics_runner") +# conftest registers a lock-less STUB manifest_utils; load the real one for the runner's lock +_mu_spec = importlib.util.spec_from_file_location(f"{_PKG}.manifest_utils_real", os.path.join(_ROOT, "manifest_utils.py")) +_mu = importlib.util.module_from_spec(_mu_spec) +_mu.__package__ = _PKG +_mu_spec.loader.exec_module(_mu) +im = importlib.import_module(f"{_PKG}.image_metrics") + + +def _save(path, value, size=(40, 30)): + Image.fromarray(np.full((size[1], size[0], 3), value, np.uint8)).save(path) + + +@pytest.fixture +def session(tmp_path, monkeypatch): + out = tmp_path / "out" + sdir = out / "benchmarks" / "s1" + (sdir / "images").mkdir(parents=True) + monkeypatch.setattr(sys.modules["folder_paths"], "get_output_directory", lambda: str(out), raising=False) + for n, v in (("a.webp", 10), ("b.webp", 200), ("c.png", 90)): + _save(sdir / "images" / n, v) + outside = tmp_path / "secret.png" + _save(outside, 50) + items = [ + {"id": 1, "file": "/view?filename=a.webp&type=output&subfolder=benchmarks/s1/images", "favorited": True, "note": "keep"}, + {"id": 2, "file": "/view?filename=b.webp&type=output&subfolder=benchmarks/s1/images", "metrics": {"luma": 9.0}}, + {"id": 3, "filename": "c.png", "file": "/view?filename=c.png&type=output"}, + {"id": 4, "file": "/view?filename=a.webp", "media_type": "video"}, + {"id": 5, "file": "/view?filename=..%2F..%2F..%2Fsecret.png"}, + {"id": 6, "filename": "../../../secret.png"}, + {"id": 7, "filename": "missing.webp"}, + {"id": 8, "florence2_no_detection": True, "filename": "a.webp"}, + ] + mp = sdir / "manifest.json" + mp.write_text(json.dumps({"session_name": "s1", "meta": {}, "items": items}), encoding="utf-8") + monkeypatch.setattr(mr, "manifest_utils", _mu) + mr._jobs.clear() + return mp + + +def _wait(job_id): + for _ in range(200): + st = mr.get_metrics_status(job_id) + if st["status"] != "running": + return st + time.sleep(0.02) + raise AssertionError("job did not finish") + + +def _items(mp): + return {i["id"]: i for i in json.loads(mp.read_text(encoding="utf-8"))["items"]} + + +def test_computes_only_missing_and_safe_items(session): + job_id, err = mr.start_metrics_job("s1") + assert err is None + st = _wait(job_id) + assert st["status"] == "complete" + it = _items(session) + assert set(it[1]["metrics"]) == set(im.BUILTIN_METRICS) + assert it[1]["metrics"]["luma"] == pytest.approx(10 / 255, abs=1e-3) + assert it[3]["metrics"]["luma"] == pytest.approx(90 / 255, abs=1e-3) + assert it[2]["metrics"] == {"luma": 9.0} # existing metrics untouched + for skipped in (4, 5, 6, 7, 8): + assert "metrics" not in it[skipped] + assert it[1]["favorited"] is True and it[1]["note"] == "keep" # other fields intact + assert st["total"] == 2 + + +def test_overwrite_recomputes_existing(session): + job_id, err = mr.start_metrics_job("s1", overwrite=True) + assert err is None and _wait(job_id)["status"] == "complete" + assert _items(session)[2]["metrics"]["luma"] == pytest.approx(200 / 255, abs=1e-3) + + +def test_nothing_to_do_and_bad_session(session): + job_id, _ = mr.start_metrics_job("s1") + _wait(job_id) + assert mr.start_metrics_job("s1")[0] is None # everything measured now + assert mr.start_metrics_job("nope")[1] == "Manifest not found" + assert mr.start_metrics_job("..")[0] is None + + +def test_flush_reloads_disk_so_concurrent_edits_survive(session): + # a user edit lands on disk between the job's initial read and its flush + mr._flush(str(session), {1: {"luma": 0.5}}, False) + data = json.loads(session.read_text(encoding="utf-8")) + for i in data["items"]: + if i["id"] == 3: + i["favorited"] = True + session.write_text(json.dumps(data), encoding="utf-8") + mr._flush(str(session), {3: {"luma": 0.7}}, False) + it = _items(session) + assert it[3]["favorited"] is True and it[3]["metrics"] == {"luma": 0.7} and it[1]["metrics"] == {"luma": 0.5} + assert not os.path.exists(str(session) + ".metrics.tmp") + + +def test_cancel_stops_job(session, monkeypatch): + orig = im.compute_metrics_numpy + + def slow(*a, **k): + time.sleep(0.15) + return orig(*a, **k) + monkeypatch.setattr(im, "compute_metrics_numpy", slow) + job_id, _ = mr.start_metrics_job("s1") + assert mr.cancel_metrics_job(job_id) is True + st = _wait(job_id) + assert st["status"] == "cancelled" and st["completed"] <= st["total"] + assert mr.cancel_metrics_job(job_id) is False + + +def test_second_job_for_same_session_rejected(session, monkeypatch): + orig = im.compute_metrics_numpy + monkeypatch.setattr(im, "compute_metrics_numpy", lambda *a, **k: (time.sleep(0.2), orig(*a, **k))[1]) + job_id, _ = mr.start_metrics_job("s1") + assert mr.start_metrics_job("s1")[1].startswith("A metrics job is already running") + mr.cancel_metrics_job(job_id) + _wait(job_id)