UCI's Occupancy Detection dataset: five environmental channels sampled about once a minute in a single office, with a real occupancy label on every row. Two things below are checkable end to end: whether the local channel classifier picks out the channels that actually move with occupancy, and whether outcome-tagged retrieval returns results consistent with that same label.
Requirements: a VAAS-X API key, pip install vaas-x pandas, and the dataset.
https://archive.ics.uci.edu/static/public/357/occupancy+detection.zip(UCI Machine Learning Repository, dataset 357.) Unzip it -- datatraining.txt is used here: 8143 rows, columns date, Temperature, Humidity, Light, CO2, HumidityRatio, Occupancy (0/1), comma-delimited, with header.
import pandas as pd
df = pd.read_csv("datatraining.txt", index_col=0, parse_dates=["date"])
channels = ["Temperature", "Humidity", "Light", "CO2", "HumidityRatio"]Occupancy is the ground truth. It doesn't go in the ingest payload -- it's used afterward, to tag outcomes and to check the classifier's output.
from vaasx import Bootstrap
brain = Bootstrap(api_key="YOUR_API_KEY", device_id="office_occupancy")
records = df[channels].to_dict(orient="records")
timestamps = df.index.astype(str).tolist()
ids = []
for i in range(0, len(records), 500):
batch = [{"payload": r, "timestamp": t} for r, t in zip(records[i:i+500], timestamps[i:i+500])]
resp = brain.ingest(batch)
ids.extend(resp["episode_ids"])occupied = df["Occupancy"].tolist()
for episode_id, is_occupied in zip(ids, occupied):
if episode_id is None:
continue
brain.outcome(episode_id, success=bool(is_occupied))hits = brain.query("occupied room, motion detected, elevated CO2 and light", k=5, prefer_success=True)
for h in hits:
print(h["timestamp"], h["outcome"]["success"], h["score"])Outcomes were tagged straight from Occupancy, so with prefer_success=True every hit above should read success: True. Anything that doesn't is worth checking against df.loc[timestamp] directly.
from vaasx.bootstrap import StatisticalProfiler, SchemaClassifier
profiler = StatisticalProfiler(device_id="office_occupancy")
for r in records:
profiler.observe(r)
result = SchemaClassifier().classify(profiler.snapshot())
print("significant channels:", sorted(result["significant_channels"]))correlations = df[channels + ["Occupancy"]].corr()["Occupancy"].drop("Occupancy").abs().sort_values(ascending=False)
print(correlations)Light and CO2 correlate with Occupancy far more strongly than Temperature, Humidity, or HumidityRatio -- check that ranking against significant_channels above rather than taking either list on faith.
datatest.txt and datatest2.txt cover different date ranges under the same schema and are also labeled -- ingest either as a second device_id and repeat the correlation check against its own labels.
More guides: Reproduction Guides. Full reference: Developer Guide.