Stroke-detector + pooling features prototype, robustness check in feature mode

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

augment/tta_sim.py

@@ -3,7 +3,9 @@
 Each image is fed several times with small shifts; votes of all copies are summed.
 Also measures robustness: the test images themselves are shifted / thickened first.
 
-usage: python tta_sim.py <dump_dir> <images.bin> <labels.target> --first 30000 --test 2000
+usage: python tta_sim.py <dump_dir> <images.bin> <labels.target> --first 30000 --test 2000 [--features]
+With --features the network was trained on stroke features: <images.bin> must hold raw pixels, distortions
+are applied to the pixels, then features are computed (no shift voting in this mode).
 """
 import argparse
 import sys
@@ -13,6 +15,9 @@ import numpy as np
 sys.path.insert(0, "/mnt/c/knp/viz")
 import altai_sim as s  # noqa: E402
 
+sys.path.insert(0, "/mnt/c/knp/features")
+import stroke_features  # noqa: E402
+
 CROSS = [(0, 0), (0, 1), (0, -1), (1, 0), (-1, 0)]
 SQUARE = [(dy, dx) for dy in (-1, 0, 1) for dx in (-1, 0, 1)]
 
@@ -48,15 +53,20 @@ def main():
     ap.add_argument("--first", type=int, default=30000)
     ap.add_argument("--test", type=int, default=2000)
     ap.add_argument("--inhib", type=float, default=255)
+    ap.add_argument("--features", action="store_true", help="network input is stroke features of the pixels")
     a = ap.parse_args()
     w, thr = s.load_dump(a.dump)
     imgs, labels = s.load_images(a.images, a.labels, a.first, a.test)
     imgs = imgs.reshape(-1, 28, 28).astype(np.int32)
     variants = {"as is": imgs, "shift 2px": shift(imgs, 0, 2), "thicker 1px": grow(imgs), "thinner 1px": thin(imgs)}
-    for vote_name, offsets in [("1 copy", [(0, 0)]), ("5 copies (cross)", CROSS), ("9 copies (3x3)", SQUARE)]:
+    votings = [("1 copy", [(0, 0)])] + ([] if a.features else [("5 copies (cross)", CROSS), ("9 copies (3x3)", SQUARE)])
+    for vote_name, offsets in votings:
         row = []
         for name, x in variants.items():
-            acc, sil = evaluate(w, thr, x.clip(0, 255).astype(np.uint8), labels, offsets, a.inhib)
+            x = x.clip(0, 255).astype(np.uint8)
+            if a.features:
+                x = stroke_features.features(x).reshape(-1, 28, 28)
+            acc, sil = evaluate(w, thr, x, labels, offsets, a.inhib)
             row.append(f"{name} {acc:5.1f}%")
         print(f"{vote_name:17s} | " + " | ".join(row), flush=True)
 

features/stroke_features.py

@@ -0,0 +1,82 @@
+"""Stroke-detector + pooling features for the AltAI column network (prototype of a spiking feature layer).
+
+Layer 1: 4 oriented line detectors (0°, 45°, 90°, 135°), 5×5 Gabor-like kernels, rectified.
+Layer 2: max pooling 3×3 with stride 2 → 4 maps × 14×14 = 784 values, so the network input size is unchanged.
+Values are scaled to 0–255 with a fixed global scale (no per-image normalisation, labels are not used).
+
+usage: python stroke_features.py <src.bin> <src.target> <out_prefix>
+The output keeps the source layout (training images, then evaluation images); augment first if needed.
+"""
+import argparse
+import shutil
+import sys
+
+import numpy as np
+
+SIDE, ORIENT, POOLED = 28, 4, 14
+SCALE = 3.5  # fixed scale: 99th percentile of pooled responses on MNIST maps to 255
+
+
+def kernels(size=5, sigma=1.4, wavelength=4.0):
+    half = size // 2
+    y, x = np.mgrid[-half:half + 1, -half:half + 1].astype(np.float32)
+    out = []
+    for k in range(ORIENT):
+        t = np.pi * k / ORIENT
+        xr = x * np.cos(t) + y * np.sin(t)
+        yr = -x * np.sin(t) + y * np.cos(t)
+        g = np.exp(-(xr ** 2 + 0.25 * yr ** 2) / (2 * sigma ** 2)) * np.cos(2 * np.pi * xr / wavelength)
+        out.append(g - g.mean())
+    return np.stack(out)
+
+
+KERNELS = kernels()
+
+
+def detect(imgs):
+    """imgs: [n, 28, 28] uint8 → rectified responses [n, 4, 28, 28] float32."""
+    x = imgs.astype(np.float32) / 255
+    pad = KERNELS.shape[-1] // 2
+    xp = np.pad(x, ((0, 0), (pad, pad), (pad, pad)))
+    out = np.zeros((len(x), ORIENT, SIDE, SIDE), dtype=np.float32)
+    for dy in range(KERNELS.shape[1]):
+        for dx in range(KERNELS.shape[2]):
+            patch = xp[:, dy:dy + SIDE, dx:dx + SIDE]
+            out += KERNELS[None, :, dy, dx, None, None] * patch[:, None]
+    return np.maximum(out, 0)
+
+
+def pool(resp):
+    """Max pooling 3×3, stride 2, padding 1: [n, 4, 28, 28] → [n, 4, 14, 14]."""
+    p = np.pad(resp, ((0, 0), (0, 0), (1, 1), (1, 1)))
+    out = np.zeros(resp.shape[:2] + (POOLED, POOLED), dtype=np.float32)
+    for dy in range(3):
+        for dx in range(3):
+            out = np.maximum(out, p[:, :, dy:dy + 2 * POOLED:2, dx:dx + 2 * POOLED:2])
+    return out
+
+
+def features(imgs, batch=5000):
+    res = []
+    for i in range(0, len(imgs), batch):
+        f = pool(detect(imgs[i:i + batch]))
+        res.append(np.clip(f * (255 / SCALE), 0, 255).astype(np.uint8).reshape(len(f), -1))
+    return np.concatenate(res)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("images")
+    ap.add_argument("labels")
+    ap.add_argument("out")
+    a = ap.parse_args()
+    imgs = np.fromfile(a.images, dtype=np.uint8).reshape(-1, SIDE, SIDE)
+    f = features(imgs)
+    f.tofile(a.out + ".bin")
+    shutil.copyfile(a.labels, a.out + ".target")
+    print(f"{a.out}: {len(f)} images, mean value {f.mean():.1f}, ~{f.mean() / 255 * 10 * 784:.0f} spikes per image,"
+          f" saturated {100 * (f == 255).mean():.2f}%", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()