encoding_metrics.py
Cross-dataset encoding metrics - 464 lines.
View on GitHub (encoding_metrics.py).
Source
Section titled “Source”"""Cross-dataset DyNED fidelity + DyNEDc compression metrics.
Computes encoder reconstruction fidelity (RMSE / SNR / R^2) at multiple
quantisation levels and standalone DyNEDc compression ratio for binary
encoder output, across CIFAR-10, Speech Commands, CIFAR10-DVS, and SSC.
No model training required - DyNED quantisation is deterministic and
DyNEDc is lossless arithmetic, so this runs in ~1 minute on CPU.
Outputs (all under assets/encoding_metrics/ by default):
- fidelity.csv
- compression.csv
- results.json
- encoding-fidelity-cross-dataset.png
- dynedc-cr-cross-dataset.png
- Typst-formatted tables printed to stdout.
Pass --img-dir to redirect the PNGs (e.g. directly into the thesis images folder).
"""
import argparse
import csv
import json
import os
import random
import sys
from pathlib import Path
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
import torch
from dyned import (
DyNEDcCompressorV4,
dyned_encode_features,
dyned_quantise_2d,
)
SEED = 42
ASSETS = Path("assets")
OUT_DIR = ASSETS / "encoding_metrics"
# ---------------------------------------------------------------------------
# Per-dataset preprocessing (mirrors the trained scripts so fidelity numbers
# are comparable to what the SNNs actually consume)
# ---------------------------------------------------------------------------
def preprocess_cifar10(n_samples):
"""Grayscale image -> log-magnitude 2D FFT, returned as [H, W] = [32, 32].
Sigma-delta runs across rows at each column (consistent with the 2D
sigma-delta pipeline used elsewhere).
"""
import torchvision
ds = torchvision.datasets.CIFAR10(root=str(ASSETS), train=False, download=False)
rng = random.Random(SEED)
indices = rng.sample(range(len(ds)), n_samples)
out = []
for idx in indices:
img, _ = ds[idx]
gray = np.array(img.convert("L"), dtype=np.float32) / 255.0 # [32, 32]
fft = np.fft.fftshift(np.fft.fft2(gray))
log_mag = np.log1p(np.abs(fft)).astype(np.float32)
out.append(log_mag)
return out
def preprocess_speech(n_samples, target_sr=8000, n_mels=80, n_fft=1024, hop_length=80, win_length=200):
"""Waveform -> log-mel spectrogram [n_mels, T_frames]."""
import torchaudio
import torchaudio.transforms as T
ds = torchaudio.datasets.SPEECHCOMMANDS(root=str(ASSETS), subset="testing", download=False)
rng = random.Random(SEED)
indices = rng.sample(range(len(ds)), n_samples)
mel = T.MelSpectrogram(
sample_rate=target_sr,
n_mels=n_mels,
n_fft=n_fft,
win_length=win_length,
hop_length=hop_length,
f_min=20,
f_max=target_sr // 2,
)
resamplers = {}
out = []
target_len = target_sr # 1 s
for idx in indices:
wav, sr, *_ = ds[idx]
if sr != target_sr:
if sr not in resamplers:
resamplers[sr] = T.Resample(sr, target_sr)
wav = resamplers[sr](wav)
if wav.shape[1] < target_len:
wav = torch.nn.functional.pad(wav, (0, target_len - wav.shape[1]))
else:
wav = wav[:, :target_len]
log_mel = torch.log1p(mel(wav).squeeze(0)).numpy().astype(np.float32)
out.append(log_mel)
return out
def _ssc_to_dense(times, units, n_channels=700, n_bins=50, t_max=1.0):
dense = np.zeros((n_channels, n_bins), dtype=np.float32)
times = np.asarray(times)
units = np.asarray(units, dtype=np.int64)
bin_edges = np.linspace(0.0, t_max, n_bins + 1)
bin_idx = np.clip(np.searchsorted(bin_edges, times, side="right") - 1, 0, n_bins - 1)
np.add.at(dense, (units, bin_idx), 1.0)
return dense
def preprocess_ssc(n_samples, n_bins=50):
"""Cochlea spike events -> log dense histogram [700, n_bins]."""
import h5py
h5_path = ASSETS / "SSC" / "ssc_test.h5"
out = []
with h5py.File(str(h5_path), "r") as f:
n = len(f["labels"])
rng = random.Random(SEED)
indices = sorted(rng.sample(range(n), n_samples))
for i in indices:
times = f["spikes/times"][i]
units = f["spikes/units"][i]
dense = _ssc_to_dense(times, units, n_channels=700, n_bins=n_bins)
out.append(np.log1p(dense).astype(np.float32))
return out
def preprocess_cifar10dvs(n_samples, n_time_bins=16, target_hw=32):
"""Event stream -> dense voxel grid -> reshape to [P*H*W, T] = [2048, 16].
Sigma-delta runs across the spatial+polarity axis at each time step (same
convention as SSC: features x time bins).
"""
import tonic
import tonic.transforms as tt
spatial_factor = target_hw / 128.0
transform = tt.Compose(
[
tt.Downsample(spatial_factor=spatial_factor),
tt.ToFrame(sensor_size=(target_hw, target_hw, 2), n_time_bins=n_time_bins),
]
)
ds = tonic.datasets.CIFAR10DVS(save_to=str(ASSETS), transform=transform)
rng = random.Random(SEED)
indices = rng.sample(range(len(ds)), n_samples)
out = []
for idx in indices:
frame, _ = ds[idx] # [T, 2, H, W]
frame = np.asarray(frame, dtype=np.float32)
T_, P, H, W = frame.shape
# Per-frame [-1, 1] normalisation matches the trained pipeline so the
# fidelity numbers are comparable to what the SNN actually sees.
flat = frame.reshape(T_, -1)
f_min = flat.min(axis=1, keepdims=True)
f_max = flat.max(axis=1, keepdims=True)
rng_ = np.maximum(f_max - f_min, 1e-8)
frame = (flat - f_min) / rng_ * 2.0 - 1.0
frame = frame.reshape(T_, P, H, W)
# [P, H, W, T] -> [P*H*W, T]
feat_time = np.transpose(frame, (1, 2, 3, 0)).reshape(P * H * W, T_)
out.append(feat_time.astype(np.float32))
return out
# ---------------------------------------------------------------------------
# Metrics
# ---------------------------------------------------------------------------
def fidelity(x, x_q):
x = x.flatten().astype(np.float64)
xq = x_q.flatten().astype(np.float64)
err = x - xq
rmse = float(np.sqrt(np.mean(err**2)))
sig_p = float(np.mean(x**2))
noise_p = float(np.mean(err**2))
snr_db = 10.0 * np.log10(sig_p / noise_p) if noise_p > 1e-30 else float("inf")
var_x = float(np.var(x))
if var_x < 1e-30:
r2 = 1.0 if rmse < 1e-12 else 0.0
else:
r2 = 1.0 - noise_p / var_x
return rmse, float(snr_db), float(r2)
def fidelity_per_dataset(name, samples, levels_list):
rows = []
for levels in levels_list:
rmses, snrs, r2s = [], [], []
for x in samples:
x_q = dyned_quantise_2d(x, levels=levels).numpy()
r, s, r2 = fidelity(x, x_q)
rmses.append(r)
snrs.append(s)
r2s.append(r2)
rows.append(
{
"dataset": name,
"levels": levels,
"rmse_mean": float(np.mean(rmses)),
"rmse_std": float(np.std(rmses)),
"snr_db_mean": float(np.mean(snrs)),
"snr_db_std": float(np.std(snrs)),
"r2_mean": float(np.mean(r2s)),
"r2_std": float(np.std(r2s)),
"n_samples": len(samples),
}
)
return rows
def compression_per_dataset(name, samples):
"""DyNEDc CR on binary encoder output (DyNED step signal at 256 levels)."""
crs, savings = [], []
comp = DyNEDcCompressorV4()
for x in samples:
spike_train = dyned_encode_features(x, levels=256).numpy()
flat = spike_train.flatten().astype(np.uint8)
if len(flat) < 2:
continue
_, info = comp.compress(flat)
if info["original_length"] == 0 or info["compressed_length"] == 0:
continue
ratio_in_out = info["original_length"] / info["compressed_length"]
crs.append(ratio_in_out)
savings.append(100.0 * (1.0 - info["compressed_length"] / info["original_length"]))
return {
"dataset": name,
"cr_mean": float(np.mean(crs)),
"cr_std": float(np.std(crs)),
"saving_pct_mean": float(np.mean(savings)),
"saving_pct_std": float(np.std(savings)),
"n_samples": len(crs),
}
# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------
def write_csv(path, rows, fieldnames):
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for row in rows:
writer.writerow(row)
def plot_fidelity(fid_rows, out_path):
datasets = sorted(
{r["dataset"] for r in fid_rows}, key=lambda d: ["CIFAR-10", "Speech Commands", "CIFAR10-DVS", "SSC"].index(d)
)
levels_list = sorted({r["levels"] for r in fid_rows})
fig, axes = plt.subplots(3, 1, figsize=(8, 11), dpi=140)
metrics = [
("rmse_mean", "rmse_std", "RMSE (lower is better)"),
("snr_db_mean", "snr_db_std", "SNR (dB, higher is better)"),
("r2_mean", "r2_std", "R^2 (higher is better)"),
]
x = np.arange(len(datasets))
width = 0.8 / len(levels_list)
for ax, (mkey, skey, title) in zip(axes, metrics):
for i, lv in enumerate(levels_list):
heights = [next(r[mkey] for r in fid_rows if r["dataset"] == d and r["levels"] == lv) for d in datasets]
stds = [next(r[skey] for r in fid_rows if r["dataset"] == d and r["levels"] == lv) for d in datasets]
ax.bar(
x + (i - (len(levels_list) - 1) / 2) * width,
heights,
width=width,
yerr=stds,
capsize=2,
label=f"{lv} levels",
)
ax.set_title(title)
ax.set_xticks(x)
ax.set_xticklabels(datasets, rotation=15, ha="right")
ax.legend(fontsize=8)
ax.grid(True, axis="y", alpha=0.3)
fig.suptitle("DyNED reconstruction fidelity across datasets")
fig.tight_layout()
out_path.parent.mkdir(parents=True, exist_ok=True)
fig.savefig(out_path)
plt.close(fig)
def plot_compression(cr_rows, out_path):
datasets = [r["dataset"] for r in cr_rows]
crs = [r["cr_mean"] for r in cr_rows]
stds = [r["cr_std"] for r in cr_rows]
saves = [r["saving_pct_mean"] for r in cr_rows]
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(9, 3.5), dpi=140)
ax1.bar(datasets, crs, yerr=stds, capsize=3, color="steelblue")
ax1.set_ylabel("CR (input / output)")
ax1.set_title("DyNEDc compression ratio (higher is better)")
ax1.set_xticks(range(len(datasets)))
ax1.set_xticklabels(datasets, rotation=15, ha="right")
ax1.grid(True, axis="y", alpha=0.3)
ax2.bar(datasets, saves, color="seagreen")
ax2.set_ylabel("Bit savings (%)")
ax2.set_title("DyNEDc bit savings vs raw bitstream")
ax2.set_xticks(range(len(datasets)))
ax2.set_xticklabels(datasets, rotation=15, ha="right")
ax2.grid(True, axis="y", alpha=0.3)
fig.tight_layout()
out_path.parent.mkdir(parents=True, exist_ok=True)
fig.savefig(out_path)
plt.close(fig)
def print_typst_fidelity(fid_rows):
datasets = sorted(
{r["dataset"] for r in fid_rows}, key=lambda d: ["CIFAR-10", "Speech Commands", "CIFAR10-DVS", "SSC"].index(d)
)
levels_list = sorted({r["levels"] for r in fid_rows})
n_cols = 1 + 3 * len(levels_list)
print()
print("// Encoding fidelity table - paste into Ch7 Section Cross-Dataset Summary")
print("#figure(")
header_levels = "".join(f" [*L={lv}*], [*L={lv}*], [*L={lv}*],\n" for lv in levels_list)
print(" table(")
print(f" columns: {n_cols},")
print(" align: (left,) + (center,) * " + str(n_cols - 1) + ",")
print(f" table.header(")
print(f" [*Dataset*]," + "".join(f" table.cell(colspan: 3)[*{lv} levels*]," for lv in levels_list))
print(" []," + (" [RMSE], [SNR (dB)], [R^2]," * len(levels_list)))
print(" ),")
for d in datasets:
cells = [f"[{d}]"]
for lv in levels_list:
row = next(r for r in fid_rows if r["dataset"] == d and r["levels"] == lv)
cells.append(f"[{row['rmse_mean']:.3f}]")
cells.append(f"[{row['snr_db_mean']:.1f}]")
cells.append(f"[{row['r2_mean']:.3f}]")
print(" " + ", ".join(cells) + ",")
print(" ),")
print(
" caption: [DyNED reconstruction fidelity across datasets at three quantisation levels (mean over 100 random samples).],"
)
print(") <tab:cross-dataset-fidelity>")
def print_typst_compression(cr_rows):
print()
print("// DyNEDc compression table - paste into Ch7 Section Cross-Dataset Summary")
print("#figure(")
print(" table(")
print(" columns: 4,")
print(" align: (left, center, center, center),")
print(" table.header([*Dataset*], [*CR (in/out)*], [*Savings (%)*], [*Samples*]),")
for r in cr_rows:
print(f" [{r['dataset']}], [{r['cr_mean']:.2f}], [{r['saving_pct_mean']:.1f}], [{r['n_samples']}],")
print(" ),")
print(" caption: [Standalone DyNEDc compression on DyNED binary encoder output (256 levels).],")
print(") <tab:cross-dataset-dynedc-cr>")
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
DATASET_FNS = {
"cifar10": ("CIFAR-10", preprocess_cifar10),
"speech": ("Speech Commands", preprocess_speech),
"cifar10dvs": ("CIFAR10-DVS", preprocess_cifar10dvs),
"ssc": ("SSC", preprocess_ssc),
}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--n-samples", type=int, default=100)
parser.add_argument("--levels", type=str, default="16,64,256")
parser.add_argument("--datasets", type=str, default="cifar10,speech,cifar10dvs,ssc")
parser.add_argument(
"--out-dir",
type=str,
default=str(OUT_DIR),
help="Where CSV/JSON results are written (default: assets/encoding_metrics)",
)
parser.add_argument(
"--img-dir", type=str, default=None, help="Where PNG charts are written (default: same as --out-dir)"
)
args = parser.parse_args()
levels_list = [int(x) for x in args.levels.split(",")]
dataset_keys = [k.strip() for k in args.datasets.split(",")]
out_dir = Path(args.out_dir)
img_dir = Path(args.img_dir) if args.img_dir else out_dir
np.random.seed(SEED)
torch.manual_seed(SEED)
random.seed(SEED)
fid_rows, cr_rows = [], []
for key in dataset_keys:
if key not in DATASET_FNS:
print(f"[skip] unknown dataset: {key}", file=sys.stderr)
continue
name, fn = DATASET_FNS[key]
print(f"[{name}] preprocessing {args.n_samples} samples...", flush=True)
try:
samples = fn(args.n_samples)
except (FileNotFoundError, RuntimeError, OSError) as e:
print(f"[{name}] skipped - {type(e).__name__}: {e}", file=sys.stderr)
continue
print(f"[{name}] computing fidelity at levels={levels_list}...", flush=True)
fid_rows.extend(fidelity_per_dataset(name, samples, levels_list))
print(f"[{name}] computing DyNEDc CR on binary encoder output...", flush=True)
cr_rows.append(compression_per_dataset(name, samples))
if not fid_rows:
print("No datasets succeeded.", file=sys.stderr)
return 1
write_csv(
out_dir / "fidelity.csv",
fid_rows,
["dataset", "levels", "rmse_mean", "rmse_std", "snr_db_mean", "snr_db_std", "r2_mean", "r2_std", "n_samples"],
)
write_csv(
out_dir / "compression.csv",
cr_rows,
["dataset", "cr_mean", "cr_std", "saving_pct_mean", "saving_pct_std", "n_samples"],
)
with open(out_dir / "results.json", "w") as f:
json.dump(
{
"fidelity": fid_rows,
"compression": cr_rows,
"config": {"n_samples": args.n_samples, "levels": levels_list, "datasets": dataset_keys, "seed": SEED},
},
f,
indent=2,
)
print(f"\nWrote {out_dir}/fidelity.csv, compression.csv, results.json")
plot_fidelity(fid_rows, img_dir / "encoding-fidelity-cross-dataset.png")
plot_compression(cr_rows, img_dir / "dynedc-cr-cross-dataset.png")
print(f"Wrote charts to {img_dir}/ (copy into ../phd-thesis/images/ when ready to update the thesis)")
print_typst_fidelity(fid_rows)
print_typst_compression(cr_rows)
return 0
if __name__ == "__main__":
sys.exit(main())