Wearable Aggregators & Cloud APIs

If your architecture doesn't rely on a native mobile app, or if you wish to support multiple hardware brands (e.g. Garmin, Whoop, Oura, Fitbit, Suunto, Coros), you will typically collect biometrics using:

1. Wearable Aggregators (such as OpenWearables, Terra, or Rook), which provide a single unified API across dozens of device manufacturers.
2. Direct Vendor Cloud APIs (e.g., Garmin Connect Developer API, Whoop Developer Platform, Oura Cloud API).

This guide walks through the event-driven webhook architecture and data normalization pipeline required to bridge these cloud APIs into the Sensia API.

---

1. Ingestion Stages & Event Timing

A prediction requires two temporal components:

1. Sleep Stage Summary & Night Biometrics: In cloud-on-cloud architecture this data arrives once a day (usually early morning) when the user wakes up and their device completes a sleep sync.
2. Pre-Prediction Biometrics (1h Window): Queried when the prediction is evaluated (e.g. before a cognitive task, during a midday wellness check, or on a recurring schedule).

Recommended Implementation Strategy

1. Step 1: On sleep.completed webhook:

   • Receive the sleep payload from the aggregator / vendor.
   • Extract and store: sleepTimeSeconds, deepSleepSeconds, remSleepSeconds, lightSleepSeconds, awakeSleepSeconds, and the time-series array of HR_night and RR_night.
   • Store these in your database keyed by user_id and date.

2. Step 2: On Prediction Request (Evaluation Window):

   • Query the aggregator's intraday biometric endpoint for the user's heart rate (HR_day_1h) and respiratory rate (RR_day_1h) over the interval [now - 1h, now].
   • Retrieve the stored night sleep record for the user.
   • Assemble the current_session payload (and optional past_sessions + past_targets).
   • Dispatch POST https://api.sensia.ai/predict.
   • Persist the resulting scores to serve as the user's historical baseline for subsequent predictions.

---

3. Data Transformation Reference (OpenWearables / Generic Aggregator)

Most aggregators output sleep breakdowns in minutes or seconds and intraday biometrics as time-series arrays.

Sensia Field	OpenWearables / Aggregator Field	Conversion Rule
sleepTimeSeconds	sleep.total_sleep_duration (seconds) or duration_in_bed - awake_duration	Ensure awake time is excluded.
deepSleepSeconds	sleep.deep_duration	Convert to integer seconds if in minutes (min * 60).
remSleepSeconds	sleep.rem_duration	Convert to integer seconds if in minutes (min * 60).
lightSleepSeconds	sleep.light_duration	Convert to integer seconds if in minutes (min * 60).
awakeSleepSeconds	sleep.awake_duration	Seconds spent awake during the sleep interval.
HR_night	sleep.heart_rate_samples	Array of float BPM readings taken during sleep.
RR_night	sleep.respiration_samples	Array of float breaths/min readings taken during sleep.
HR_day_1h	intraday.heart_rate where timestamp >= now - 1h	Array of float BPM readings.
RR_day_1h	intraday.respiration where timestamp >= now - 1h	Array of float breaths/min readings.

---

4. Production Transformation Pipeline (TypeScript / Node.js)

Below is an end-to-end service demonstrating webhook handling, transformation, and historical rolling context injection:

sensia-aggregator-service.ts

import axios from "axios";

export interface StoredUserHistory {
  session: Record<string, any>;
  scores: {
    C_SCORE: number;
    attention: number;
    memory: number;
    reasoning: number;
    stress: number;
    valence: "Low" | "High";
    arousal: "Low" | "High";
  };
}

export class SensiaIntegrationService {
  private sensiaApiKey: string;
  private sensiaEndpoint: string = "https://api.sensia.ai/predict";

  constructor(apiKey: string) {
    this.sensiaApiKey = apiKey;
  }

  /**
   * Transforms aggregator sleep & intraday data into a Sensia prediction payload.
   */
  public buildPayload(
    sleepData: {
      deepSec: number;
      remSec: number;
      lightSec: number;
      awakeSec: number;
      hrNight: number[];
      rrNight?: number[];
    },
    intraday1h: {
      hrSamples: number[];
      rrSamples?: number[];
    },
    userProfile: {
      age?: number;
      height_cm?: number;
      weight_kg?: number;
      sex?: "M" | "F";
    },
    history: StoredUserHistory[] = []
  ) {
    const totalSleepSec = sleepData.deepSec + sleepData.remSec + sleepData.lightSec;

    const currentSession = {
      HR_day_1h: intraday1h.hrSamples,
      RR_day_1h: intraday1h.rrSamples ?? [],
      HR_night: sleepData.hrNight,
      RR_night: sleepData.rrNight ?? [],
      sleepTimeSeconds: Math.round(totalSleepSec),
      deepSleepSeconds: Math.round(sleepData.deepSec),
      remSleepSeconds: Math.round(sleepData.remSec),
      lightSleepSeconds: Math.round(sleepData.lightSec),
      awakeSleepSeconds: Math.round(sleepData.awakeSec),
      deepPercentage: totalSleepSec > 0 ? Number((sleepData.deepSec / totalSleepSec).toFixed(4)) : 0,
      remPercentage: totalSleepSec > 0 ? Number((sleepData.remSec / totalSleepSec).toFixed(4)) : 0,
      lightPercentage: totalSleepSec > 0 ? Number((sleepData.lightSec / totalSleepSec).toFixed(4)) : 0,
      age: userProfile.age,
      height_cm: userProfile.height_cm,
      weight_kg: userProfile.weight_kg,
      sex: userProfile.sex ?? null,
    };

    // Sensia accepts up to 5 historical sessions for temporal calibration
    const pastSessions = history.slice(-5).map((h) => h.session);
    const pastTargets = history.slice(-5).map((h) => h.scores);

    return {
      current_session: currentSession,
      ...(pastSessions.length > 0
        ? { past_sessions: pastSessions, past_targets: pastTargets }
        : {}),
    };
  }

  /**
   * Executes prediction against Sensia API.
   */
  public async getPrediction(payload: ReturnType<typeof this.buildPayload>) {
    const response = await axios.post(this.sensiaEndpoint, payload, {
      headers: {
        "X-API-Key": this.sensiaApiKey,
        "Content-Type": "application/json",
      },
    });

    return response.data;
  }
}

---

5. Production Transformation Pipeline (Python)

If your backend is built on FastAPI, Flask, or Django:

sensia_aggregator_service.py

import os
import requests
from typing import Dict, Any, List, Optional

SENSIA_API_URL = "https://api.sensia.ai/predict"
SENSIA_API_KEY = os.getenv("SENSIA_API_KEY")

def create_sensia_payload(
    sleep_record: Dict[str, Any],
    intraday_1h: Dict[str, Any],
    demographics: Optional[Dict[str, Any]] = None,
    history: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    """
    Transforms aggregator sleep and 1h intraday biometrics into the Sensia format.
    """
    deep_sec = float(sleep_record.get("deepSleepSeconds", 0))
    rem_sec = float(sleep_record.get("remSleepSeconds", 0))
    light_sec = float(sleep_record.get("lightSleepSeconds", 0))
    awake_sec = float(sleep_record.get("awakeSleepSeconds", 0))
    total_sleep_sec = deep_sec + rem_sec + light_sec

    demo = demographics or {}

    current_session = {
        "HR_day_1h": [float(x) for x in intraday_1h.get("HR_day_1h", [])],
        "RR_day_1h": [float(x) for x in intraday_1h.get("RR_day_1h", [])],
        "HR_night": [float(x) for x in sleep_record.get("HR_night", [])],
        "RR_night": [float(x) for x in sleep_record.get("RR_night", [])],
        "sleepTimeSeconds": int(round(total_sleep_sec)),
        "deepSleepSeconds": int(round(deep_sec)),
        "remSleepSeconds": int(round(rem_sec)),
        "lightSleepSeconds": int(round(light_sec)),
        "awakeSleepSeconds": int(round(awake_sec)),
        "deepPercentage": round(deep_sec / total_sleep_sec, 4) if total_sleep_sec > 0 else 0.0,
        "remPercentage": round(rem_sec / total_sleep_sec, 4) if total_sleep_sec > 0 else 0.0,
        "lightPercentage": round(light_sec / total_sleep_sec, 4) if total_sleep_sec > 0 else 0.0,
        "age": demo.get("age"),
        "height_cm": demo.get("height_cm"),
        "weight_kg": demo.get("weight_kg"),
        "sex": demo.get("sex"),
    }

    payload = {"current_session": current_session}

    # If historical sessions are present, append them (up to 5)
    if history:
        recent_history = history[-5:]
        payload["past_sessions"] = [item["session"] for item in recent_history]
        payload["past_targets"] = [item["scores"] for item in recent_history]

    return payload


def request_sensia_prediction(payload: Dict[str, Any]) -> Dict[str, Any]:
    """
    Calls the Sensia inference endpoint.
    """
    headers = {
        "X-API-Key": SENSIA_API_KEY,
        "Content-Type": "application/json",
    }
    response = requests.post(SENSIA_API_URL, json=payload, headers=headers, timeout=10)
    response.raise_for_status()
    return response.json()

---

6. Historical Temporal Calibration

Providing past session history significantly refines prediction precision because the model adapts to individual baselines:

• Each entry in past_sessions must have its corresponding output score object in past_targets.
• We recommend maintaining a table in your database such as sensia_evaluations:

  CREATE TABLE sensia_evaluations (
      id UUID PRIMARY KEY,
      user_id VARCHAR(255) NOT NULL,
      session_payload JSONB NOT NULL,
      result_scores JSONB NOT NULL,
      created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
  );

• When predicting for a user, query their last 5 records ordered by created_at ASC and inject them as past_sessions and past_targets.

---

Next Steps

• Understand Session Data fields & constraints →
• Check Output Scores & classifications →
• Review the API Reference →