Profiler and device simulator

import time, numpy as np, torch, onnxruntime as ort
from thop import profile as thop_profile

TARGETS = {                     # rough MCU settings: clock in MHz, multiply-accumulates per cycle (estimate)
    "ESP32-S3 (240 MHz)": (240, 1.0),
    "Silicon Labs xG24 (78 MHz)": (78, 1.0),
    "Cortex-M4F (80 MHz)": (80, 0.5),
    "Cortex-M7 (480 MHz)": (480, 1.0),
}

def peak_activation_bytes(model, sample, bytes_per_value=1):
    """int8 tensor-arena estimate: the largest (inputs + output) of any single layer. Add ~40% margin."""
    peak = 0
    def hook(m, inp, out):
        nonlocal peak
        n = sum(t.numel() for t in inp if torch.is_tensor(t)) + (out.numel() if torch.is_tensor(out) else 0)
        peak = max(peak, n * bytes_per_value)
    hooks = [m.register_forward_hook(hook) for m in model.modules() if not list(m.children())]
    model.eval()
    with torch.no_grad(): model(sample)
    for h in hooks: h.remove()
    return peak

def macs(model, sample):
    m, _ = thop_profile(model, inputs=(sample,), verbose=False)
    return int(m)

def onnx_latency_ms(onnx_path, sample, provider="CPUExecutionProvider", runs=50):
    sess = ort.InferenceSession(str(onnx_path), providers=[provider])
    feed = {sess.get_inputs()[0].name: sample.numpy()}
    for _ in range(5): sess.run(None, feed)
    t = time.perf_counter()
    for _ in range(runs): sess.run(None, feed)
    return (time.perf_counter() - t) / runs * 1000

import sys, torch
from pathlib import Path
ROOT = Path.home() / "sutraai"; sys.path.insert(0, str(ROOT)); EXP = ROOT / "exports"
from sutra_core.profile import TARGETS, peak_activation_bytes, macs, onnx_latency_ms

RUNTIME_ALLOWANCE_KB = 100     # placeholder for TFLite Micro + your ops; replace with a real build size in Month 3

def row(name, model, sample, onnx_file, tflite_file):
    flash_kb = Path(tflite_file).stat().st_size / 1024 + RUNTIME_ALLOWANCE_KB
    ram_kb = peak_activation_bytes(model, sample) * 1.4 / 1024
    m = macs(model, sample)
    cpu_ms = onnx_latency_ms(onnx_file, sample)
    est = {t: m / (mhz * 1e6 * per_cycle) * 1000 for t, (mhz, per_cycle) in TARGETS.items()}
    return name, flash_kb, ram_kb, m, cpu_ms, est

# fill with your models from Steps 05–09, for example:
# rows = [row("audio_v1", audio_model, torch.randn(1, 50, 13), EXP/"audio_v1.onnx", EXP/"audio_v1_tf/audio_v1_full_integer_quant.tflite"), ...]

lines = ["| model | flash KB (est.) | RAM KB (est.) | MACs | Jetson CPU ms (measured) | " + " | ".join(f"{t} ms (est.)" for t in TARGETS) + " |",
         "|---|---|---|---|---|" + "---|" * len(TARGETS)]
for name, flash, ram, m, cpu, est in rows:
    lines.append(f"| {name} | {flash:.0f} | {ram:.0f} | {m:,} | {cpu:.2f} | " + " | ".join(f"{v:.0f}" for v in est.values()) + " |")
(ROOT / "reports/profile.md").write_text("\n".join(lines))
print("\n".join(lines))

import json, hashlib, numpy as np
from ai_edge_litert.interpreter import Interpreter

def make_manifest(version, artifact_path, labels, pipeline, out_path):
    blob = open(artifact_path, "rb").read()
    m = {"version": version, "artifact_path": str(artifact_path), "sha256": hashlib.sha256(blob).hexdigest(),
         "size": len(blob), "labels": list(labels), "pipeline": pipeline}
    json.dump(m, open(out_path, "w"), indent=2)
    return m

class FakeDevice:
    def __init__(self, manifest_path):
        self.telemetry = []
        self.install(manifest_path)

    def install(self, manifest_path):                          # "OTA update"
        m = json.load(open(manifest_path))
        blob = open(m["artifact_path"], "rb").read()
        if hashlib.sha256(blob).hexdigest() != m["sha256"]:
            raise ValueError("corrupt download: sha256 mismatch, keeping the old model")
        self.m = m
        self.itp = Interpreter(model_content=blob); self.itp.allocate_tensors()
        self.inp, self.out = self.itp.get_input_details()[0], self.itp.get_output_details()[0]

    def infer(self, features):                                # features: output of the SAME DSP as training
        s, z = self.inp["quantization"]
        q = np.clip(np.round(features / s + z), -128, 127).astype(np.int8)
        self.itp.set_tensor(self.inp["index"], q.reshape(self.inp["shape"])); self.itp.invoke()
        so, zo = self.out["quantization"]
        scores = (self.itp.get_tensor(self.out["index"])[0].astype(np.float32) - zo) * so
        e = np.exp(scores - scores.max()); probs = e / e.sum()     # softmax (outputs are logits)
        label, conf = self.m["labels"][int(probs.argmax())], float(probs.max())
        self.telemetry.append({"model_version": self.m["version"], "label": label, "confidence": round(conf, 3)})
        return label, conf

import sys, json, yaml, numpy as np
from pathlib import Path
ROOT = Path.home() / "sutraai"; sys.path.insert(0, str(ROOT)); EXP = ROOT / "exports"
from sutra_core.device_sim import make_manifest, FakeDevice

d = np.load(ROOT / "data/processed/audio_v1.npz")          # saved from NB32: Xte, yte, labels
labels, Xte, yte = list(d["labels"]), d["Xte"], d["yte"]
pipeline = yaml.safe_load(open(ROOT / "configs/audio.yaml"))
make_manifest(1, EXP / "audio_v1_tf/audio_v1_full_integer_quant.tflite", labels, pipeline, EXP / "manifest_v1.json")

dev = FakeDevice(EXP / "manifest_v1.json")
results = [dev.infer(x) for x in Xte]
acc_v1 = np.mean([labels.index(lab) == t for (lab, _), t in zip(results, yte)])
review = [i for i, (_, c) in enumerate(results) if c < 0.6]
print(f"v1 on-device accuracy {acc_v1:.3f} | {len(review)} windows sent for review")
json.dump(dev.telemetry, open(ROOT / "reports/telemetry.json", "w"))

# after retraining and exporting audio_v2 ...
make_manifest(2, EXP / "audio_v2_tf/audio_v2_full_integer_quant.tflite", labels, pipeline, EXP / "manifest_v2.json")
dev.install(EXP / "manifest_v2.json")
acc_v2 = np.mean([labels.index(dev.infer(x)[0]) == t for x, t in zip(Xte, yte)])
print(f"v1 {acc_v1:.3f} → v2 {acc_v2:.3f}")

bad = json.load(open(EXP / "manifest_v2.json")); bad["sha256"] = "0" * 64
json.dump(bad, open(EXP / "manifest_bad.json", "w"))
try:
    dev.install(EXP / "manifest_bad.json"); rejected = False
except ValueError as e:
    rejected = True; print("✔", e)