| |
| """Agent Calibration Layer (ACL): per-camera score statistics from normal data. |
| |
| Scores a subset of TRAIN (all-normal) samples with the saved memory bank and |
| records each camera agent's median / IQR of its anomaly-score reports. Fusion |
| then operates on per-agent standardized scores, removing systematic camera |
| bias before pooling. Label-free (uses normal training data only). |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import os |
| import sys |
|
|
| import numpy as np |
|
|
| HERE = os.path.dirname(os.path.abspath(__file__)) |
| ROOT = os.path.dirname(HERE) |
| sys.path.insert(0, ROOT) |
|
|
| from src.data import load_samples, subsample_samples |
| from src.scorer import PatchScorer |
|
|
| DATA = "/home/z/my-project/data" |
| IMG_ROOT = os.path.join(DATA, "realiad_256") |
| JSON_ROOT = os.path.join(DATA, "mirror/realiad_jsons") |
| RES = os.path.join(ROOT, "results") |
|
|
| cats = ["usb", "pcb", "mint", "eraser", "bottle_cap", "zipper"] |
| out = {} |
| for cat in cats: |
| bank_path = os.path.join(RES, "scores", f"{cat}_bank.npy") |
| scorer = PatchScorer(seed=0) |
| scorer.load(bank_path) |
| train = load_samples(cat, IMG_ROOT, JSON_ROOT, "train") |
| sub = subsample_samples(train, n_ok=150, n_ng=0, seed=1)[:150] |
| flat = [p for s in sub for p in s.image_paths] |
| sc, _ = scorer.score_images(flat) |
| sc = sc.reshape(len(sub), 5) |
| per_cam = {} |
| for v in range(5): |
| col = sc[:, v] |
| per_cam[f"C{v+1}"] = { |
| "median": float(np.median(col)), |
| "iqr": float(np.subtract(*np.percentile(col, [75, 25]))) + 1e-9, |
| "mean": float(np.mean(col)), |
| } |
| out[cat] = per_cam |
| print(cat, {k: round(v["median"], 4) for k, v in per_cam.items()}) |
|
|
| with open(os.path.join(RES, "agent_calibration.json"), "w") as f: |
| json.dump(out, f, indent=1) |
| print("saved", os.path.join(RES, "agent_calibration.json")) |
|
|
