File size: 1,868 Bytes
acc1556 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 | #!/usr/bin/env python3
"""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 # noqa: E402
from src.scorer import PatchScorer # noqa: E402
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"))
|