One config drives everything

name: audio_v1
dataset: data/raw/audio
input:      { type: timeseries, axes: [audio], frequency_hz: 16000, window_ms: 1000, stride_ms: 500 }
processing: { type: mfcc, num_coefficients: 13, frame_length_ms: 20, frame_stride_ms: 20, num_filters: 32, fft_length: 512 }
learning:   { type: classifier, arch: conv1d, epochs: 100, lr: 0.005 }
export:     [onnx, tflite_int8, c_array]

name: sensor_v1
dataset: data/raw/sensor
input:      { type: timeseries, axes: [accX, accY, accZ], frequency_hz: 50, window_ms: 4000, stride_ms: 1000 }
processing: { type: raw, scale: 1.0 }
learning:   { type: classifier, arch: transformer_tiny, d_model: 32, heads: 2, layers: 2, epochs: 80, lr: 0.001, fallback: conv1d }
export:     [onnx, tflite_int8, c_array]

name: image_v1
dataset: data/processed/images
input:      { type: image, width: 96, height: 96, color: rgb }
processing: { type: image, normalize: true }
learning:   { type: transfer_learning_image, arch: mobilenetv2_050, epochs: 30, lr: 0.001 }
export:     [onnx, tflite_int8, c_array, tensorrt]

REGISTRY = {}

def register(name):
    def deco(fn):
        REGISTRY[name] = fn
        return fn
    return deco

@register("raw")
def raw_block(cfg, windows):                     # windows: [N, T, axes]
    return (windows * cfg.get("scale", 1.0)).astype(np.float32)

@register("mfcc")
def mfcc_block(cfg, windows):                    # windows: [N, samples]
    return np.stack([mfcc(w, n_coef=cfg["num_coefficients"], frame_ms=cfg["frame_length_ms"],
                          stride_ms=cfg["frame_stride_ms"], n_mels=cfg["num_filters"],
                          n_fft=cfg["fft_length"]) for w in windows])

def apply(cfg, windows):
    return REGISTRY[cfg["type"]](cfg, windows)

import timm

def build(learning, n_classes, sample_shape, mu=None, sd=None):
    """sample_shape: shape of ONE input, e.g. (50, 13) for audio, (200, 3) for sensor."""
    arch = learning["arch"]
    if arch == "conv1d":
        core = Conv1DNet(sample_shape[-1], n_classes)
    elif arch == "transformer_tiny":
        core = TinyTransformer(sample_shape[-1], n_classes, sample_shape[0],
                               learning.get("d_model", 32), learning.get("heads", 2), learning.get("layers", 2))
    elif arch.startswith("mobilenet"):
        return timm.create_model(arch, pretrained=True, num_classes=n_classes)
    else:
        raise ValueError(f"unknown arch {arch}")
    return nn.Sequential(Normalize(mu, sd), core) if mu is not None else core

import json, random, sys, numpy as np, torch, yaml, librosa
from pathlib import Path
from . import dsp, models
from .data import load_phyphox, resample, windows
from .train import fit_arrays, accuracy

ROOT = Path.home() / "sutraai"

def load_timeseries(cfg):
    """Returns {file: windows} and labels. Audio windows: [N, samples]. Sensor: [N, T, axes]."""
    inp, root = cfg["input"], ROOT / cfg["dataset"]
    files = sorted([*root.glob("*/*.wav"), *root.glob("*/*.csv")])
    labels = sorted({f.parent.name for f in files})
    hz, win, hop = inp["frequency_hz"], inp["window_ms"] / 1000, inp["stride_ms"] / 1000
    out = {}
    for f in files:
        if f.suffix == ".wav":
            sig, _ = librosa.load(f, sr=hz, mono=True)
            w, s = int(hz * win), int(hz * hop)
            out[f] = np.stack([sig[i:i + w] for i in range(0, len(sig) - w + 1, s)]) if len(sig) >= w else None
        else:
            out[f] = windows(resample(load_phyphox(f), hz), hz, win, hop)
    return {f: w for f, w in out.items() if w is not None and len(w)}, labels

def split_files(files, seed=42):
    random.seed(seed)
    return {f: random.choices(["train", "val", "test"], weights=[70, 15, 15])[0] for f in files}

def run(config_path, out_dir=None):
    cfg = yaml.safe_load(open(config_path))
    out = Path(out_dir) if out_dir else ROOT
    torch.manual_seed(42); np.random.seed(42)
    per_file, labels = load_timeseries(cfg)
    split = split_files(list(per_file))
    def arrays(name):
        X = [dsp.apply(cfg["processing"], per_file[f]) for f in per_file if split[f] == name]
        y = [[labels.index(f.parent.name)] * len(per_file[f]) for f in per_file if split[f] == name]
        return np.concatenate(X).astype(np.float32), np.concatenate(y)
    Xtr, ytr = arrays("train"); Xva, yva = arrays("val"); Xte, yte = arrays("test")
    mu, sd = Xtr.mean(axis=(0, 1)), Xtr.std(axis=(0, 1)) + 1e-6
    model = models.build(cfg["learning"], len(labels), Xtr.shape[1:], mu, sd)
    model, val_acc = fit_arrays(model, Xtr, ytr, Xva, yva, cfg["learning"]["epochs"], cfg["learning"]["lr"])
    metrics = {"run": cfg["name"], "classes": labels, "val_acc": val_acc, "test_acc": accuracy(model, Xte, yte)}
    (out / "reports" / cfg["name"]).mkdir(parents=True, exist_ok=True)
    (out / "models" / cfg["name"]).mkdir(parents=True, exist_ok=True)
    json.dump(metrics, open(out / "reports" / cfg["name"] / "metrics.json", "w"), indent=2)
    torch.save(model.state_dict(), out / "models" / cfg["name"] / "best.pt")
    print(json.dumps({"event": "done", **metrics}))
    return metrics

if __name__ == "__main__":
    run(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else None)

%load_ext autoreload
%autoreload 2
import sys
from pathlib import Path
ROOT = Path.home() / "sutraai"; sys.path.insert(0, str(ROOT))
from sutra_core.run import run

results = {}
for name in ["audio", "sensor"]:
    results[name] = run(ROOT / "configs" / f"{name}.yaml")
results