Step 06 of 16Days 8–10 · Mon 2 – Wed 4 NovWeek 2Phase A · notebooks

Sensor model: CNN vs Transformer

Turn phone accelerometer recordings into training windows, then compare a 1D-CNN with a tiny Transformer on the same data. This is template T4 (Fall Detection); T6 (Machine Vibration) uses the same code.

Runs on: Phone (phyphox app) → JupyterLab on the Jetson
Notebooks: 20_sensor_data.ipynb, 21_sensor_train.ipynb
Creates: data/raw/sensor/<label>/*.csv, exports/sensor_v1*, reports/sensor_v1/metrics.json
Time: Three days (8–10)

Where Where to implement

Data

One CSV per recording: data/raw/sensor/<label>/rec_01.csv
At least 6 recordings per class, so whole recordings can go to test

Code

sutra_core/data.py: load_phyphox, resample, windows
sutra_core/models.py: Conv1DNet, TinyTransformer
sutra_core/train.py: fit_arrays, accuracy

Do Do it

1Record the data (Day 8)

Install phyphox on your phone. Use the Acceleration (without g) or Accelerometer experiment and export as CSV.
Falls (T4): put the phone in a pouch on your waist and fall onto a mattress (safely), 10–20 s per recording. Daily activity: walking, sitting down, stairs, picking things up.
Vibration (T6): tape the phone to a table fan: off / low / high / a coin taped to one blade (imbalance).
Copy the files to data/raw/sensor/fall/, data/raw/sensor/adl/, and so on.

2NB20: load, resample to 50 Hz, window

Filesutra_core/data.py

import numpy as np, pandas as pd

def load_phyphox(path):
    """phyphox CSV: first column is time in seconds, then x, y, z."""
    df = pd.read_csv(path)
    df = df.iloc[:, :4]
    df.columns = ["t", "ax", "ay", "az"]
    return df

def resample(df, hz=50):
    t = np.arange(df.t.iloc[0], df.t.iloc[-1], 1 / hz)
    return np.stack([np.interp(t, df.t, df[c]) for c in ["ax", "ay", "az"]], axis=1)   # [T, 3]

def windows(sig, hz=50, win_s=4.0, stride_s=1.0):
    w, s = int(hz * win_s), int(hz * stride_s)
    if len(sig) < w:
        return np.empty((0, w, sig.shape[1]))
    return np.stack([sig[i:i + w] for i in range(0, len(sig) - w + 1, s)])               # [N, w, 3]

Notebook cellnotebooks/20_sensor_data.ipynb

%load_ext autoreload
%autoreload 2
import sys, random, numpy as np, matplotlib.pyplot as plt
from pathlib import Path
ROOT = Path.home() / "sutraai"; sys.path.insert(0, str(ROOT))
from sutra_core.data import load_phyphox, resample, windows

recs = {}
for f in sorted((ROOT / "data/raw/sensor").glob("*/*.csv")):
    recs.setdefault(f.parent.name, []).append(f)
labels = sorted(recs)
print({k: len(v) for k, v in recs.items()})

# split by RECORDING, never by window: windows from one recording are nearly identical
random.seed(42); split_of = {}
for label, files in recs.items():
    files = files[:]; random.shuffle(files); n = len(files)
    for i, f in enumerate(files):
        split_of[f] = "test" if i < max(1, n // 5) else ("val" if i < max(2, 2 * n // 5) else "train")

def build(split):
    X, y = [], []
    for f, s in split_of.items():
        if s != split: continue
        w = windows(resample(load_phyphox(f)))
        if len(w): X.append(w); y += [labels.index(f.parent.name)] * len(w)
    return np.concatenate(X).astype(np.float32), np.array(y)

Xtr, ytr = build("train"); Xva, yva = build("val"); Xte, yte = build("test")
mu, sd = Xtr.mean(axis=(0, 1)), Xtr.std(axis=(0, 1)) + 1e-6       # normalization comes from TRAIN only
np.savez(ROOT / "data/processed/sensor_v1.npz", Xtr=Xtr, ytr=ytr, Xva=Xva, yva=yva, Xte=Xte, yte=yte,
         mu=mu, sd=sd, labels=np.array(labels))
print(Xtr.shape, Xva.shape, Xte.shape)

fig, axes = plt.subplots(1, len(labels), figsize=(5 * len(labels), 3), squeeze=False)
for ax, lab in zip(axes[0], labels):
    ax.plot(Xtr[ytr == labels.index(lab)][0]); ax.set_title(lab)
plt.show()

3NB21: 1D-CNN on raw windows (Day 9)

Filesutra_core/models.py

import torch, torch.nn as nn

class Normalize(nn.Module):
    """Bakes the train mean/std into the model, so a device can feed raw sensor values."""
    def __init__(self, mu, sd):
        super().__init__()
        self.register_buffer("mu", torch.tensor(mu, dtype=torch.float32))
        self.register_buffer("sd", torch.tensor(sd, dtype=torch.float32))
    def forward(self, x):
        return (x - self.mu) / self.sd

class Conv1DNet(nn.Module):
    def __init__(self, n_axes, n_classes):
        super().__init__()
        self.net = nn.Sequential(
            nn.Conv1d(n_axes, 16, 5, padding=2), nn.BatchNorm1d(16), nn.ReLU(), nn.MaxPool1d(2),
            nn.Conv1d(16, 32, 5, padding=2), nn.BatchNorm1d(32), nn.ReLU(), nn.MaxPool1d(2),
            nn.Conv1d(32, 64, 3, padding=1), nn.BatchNorm1d(64), nn.ReLU(),
            nn.AdaptiveAvgPool1d(1), nn.Flatten(), nn.Linear(64, n_classes))
    def forward(self, x):                    # x: [batch, time, axes]
        return self.net(x.transpose(1, 2))

class TinyTransformer(nn.Module):
    def __init__(self, n_axes, n_classes, seq_len, d=32, heads=2, layers=2):
        super().__init__()
        self.embed = nn.Linear(n_axes, d)
        self.pos = nn.Parameter(torch.zeros(1, seq_len, d))
        layer = nn.TransformerEncoderLayer(d, heads, dim_feedforward=64, dropout=0.1, batch_first=True)
        self.enc = nn.TransformerEncoder(layer, layers, enable_nested_tensor=False)
        self.head = nn.Linear(d, n_classes)
    def forward(self, x):                    # x: [batch, time, axes]
        return self.head(self.enc(self.embed(x) + self.pos).mean(dim=1))

Filesutra_core/train.py (add)

import json, torch, torch.nn.functional as F
from torch.utils.data import DataLoader, TensorDataset

def accuracy(model, X, y, batch=256):
    model.eval(); correct = 0
    with torch.no_grad():
        for i in range(0, len(X), batch):
            out = model(torch.tensor(X[i:i + batch]).cuda())
            correct += int((out.argmax(1).cpu().numpy() == y[i:i + batch]).sum())
    return correct / len(y)

def fit_arrays(model, Xtr, ytr, Xva, yva, epochs=60, lr=1e-3, batch=64):
    model = model.cuda()
    dl = DataLoader(TensorDataset(torch.tensor(Xtr), torch.tensor(ytr)), batch_size=batch, shuffle=True)
    opt = torch.optim.AdamW(model.parameters(), lr=lr)
    best, best_state = -1.0, None
    for ep in range(epochs):
        model.train()
        for x, y in dl:
            loss = F.cross_entropy(model(x.cuda()), y.cuda())
            opt.zero_grad(); loss.backward(); opt.step()
        acc = accuracy(model, Xva, yva)
        print(json.dumps({"epoch": ep + 1, "loss": round(loss.item(), 4), "val_acc": round(acc, 4)}))
        if acc > best:
            best, best_state = acc, {k: v.detach().clone() for k, v in model.state_dict().items()}
    model.load_state_dict(best_state)
    return model, best