Reproducing Wearable Activity Classification

https://archive.ics.uci.edu/static/public/256/daily+and+sports+activities.zip

(UCI Machine Learning Repository, dataset 256 -- Barshan & Altun, 2010.) 162.9MB. Unzip it -- each of 19 activities has its own folder (a01 through a19), each containing one subfolder per subject (p1 through p8), each containing 60 five-second segment files (s01.txt through s60.txt). Every segment file is 45 columns (5 body units × 9 axes: x/y/z accelerometer, gyroscope, magnetometer) by 125 rows (5 seconds at 25Hz), comma-delimited, no header. This guide uses a01 (sitting), a17 (rowing), and a19 (basketball).

import pandas as pd
from pathlib import Path

columns = [f"{unit}_{axis}" for unit in ["T", "RA", "LA", "RL", "LL"]
           for axis in ["xacc","yacc","zacc","xgyro","ygyro","zgyro","xmag","ymag","zmag"]]

def load_subject_activity(root, activity_code, subject_id):
    """Concatenates all 60 five-second segments into one continuous
    per-subject-per-activity stream -- matching how a real wearable
    actually streams in production, rather than 480 disconnected windows."""
    folder = Path(root) / activity_code / subject_id
    segments = sorted(folder.glob("s*.txt"))
    frames = [pd.read_csv(f, header=None, names=columns) for f in segments]
    return pd.concat(frames, ignore_index=True)

This per-subject, per-activity loading is the important part -- see the methodology note below before you change it.

from vaasx import Bootstrap

def ingest_stream(subject_id, activity_code, activity_name, root):
    brain = Bootstrap(api_key="YOUR_API_KEY", device_id=f"{subject_id}_{activity_name}")
    df = load_subject_activity(root, activity_code, subject_id)
    records = df.to_dict(orient="records")
    for i in range(0, len(records), 500):
        batch = records[i:i+500]
        brain.ingest([{"payload": r} for r in batch])
    return df

sitting = ingest_stream("p1", "a01", "sitting", "daily_sports_activities")
basketball = ingest_stream("p1", "a19", "basketball", "daily_sports_activities")

from vaasx import Bootstrap

brain = Bootstrap(api_key="YOUR_API_KEY", device_id="p1_basketball")
hits = brain.query("elevated multi-axis motion, high dynamic load", k=5)
for h in hits:
    print(h["score"], h["timestamp"], h["text"])

from vaasx.bootstrap import StatisticalProfiler, SchemaClassifier

def classify_subject_activity(root, activity_code, subject_id):
    df = load_subject_activity(root, activity_code, subject_id)
    profiler = StatisticalProfiler(device_id=f"{subject_id}_{activity_code}")
    for _, row in df.iterrows():
        profiler.observe(row.to_dict())
    return SchemaClassifier().classify(profiler.snapshot())

subjects = [f"p{i}" for i in range(1, 9)]
for activity_code, name in [("a01", "sitting"), ("a19", "basketball"), ("a17", "rowing")]:
    stable_counts, significant_counts = [], []
    for subject_id in subjects:
        result = classify_subject_activity("daily_sports_activities", activity_code, subject_id)
        stable_counts.append(len(result["stable_channels"]))
        significant_counts.append(len(result["significant_channels"]))
    print(name, "mean stable:", sum(stable_counts)/8, "mean significant:", sum(significant_counts)/8)

This runs entirely locally -- no data leaves the machine for the classification step itself. Expect sitting to leave a real number of channels classified stable across subjects, and basketball/rowing to collapse toward zero stable channels in every subject -- consistently, not just on average.

import numpy as np

def mean_channel_variance(root, activity_code, subject_id):
    df = load_subject_activity(root, activity_code, subject_id)
    return df.var().mean()

for activity_code, name in [("a19", "basketball"), ("a17", "rowing")]:
    ratios = [
        mean_channel_variance("daily_sports_activities", activity_code, s)
        / mean_channel_variance("daily_sports_activities", "a01", s)
        for s in subjects
    ]
    print(name, "vs. sitting, mean variance ratio:", sum(ratios) / len(ratios))

Plain pandas/numpy, nothing from the SDK. Compare this ratio directly against whether the classifier's stable/significant split moved the way you'd expect -- don't take either number on faith.

The first version of this test pooled all 8 subjects together per activity before profiling. That gave a physically nonsensical result -- sitting showed up with more significant channels than basketball. The bug wasn't in the classifier: pooling subjects mixes each person's own sensor baseline and IMU-orientation differences into the between-subject variance for every channel, and that cross-subject noise swamps the real signal hardest for the low-motion activity, where there's no true movement to dominate it. Basketball's actual movement signal is large enough to win regardless of pooling; sitting's isn't. Profiling one continuous per-subject, per-activity stream at a time -- exactly as a single wearable on a single athlete actually streams in production -- removes that contamination and gives the correct, consistent answer. Worth remembering any time you're validating a per-entity classification claim: pool the data first and you can quietly validate the wrong thing.

What this reproduces: zero-config channel profiling correctly separating a sedentary activity from a dynamic one, on real human-movement data, in a domain never tuned for in advance. What it does not show: injury prediction, movement-quality scoring, or that a flagged pattern corresponds to injury risk -- that would need labelled injury-outcome data specific to a sport and population, which this dataset doesn't have. Treat the ingestion and classification layer as what's proven here, and any injury-risk model built on top of it as a separate claim needing its own validation.

More guides: Reproduction Guides. Full reference: Developer Guide. Technical writeup with the full 8-subject results: Sports & Human Performance whitepaper.