Sensor Data API

# Python 3.10+ / Google Colab. Install first: pip install requests
import csv
import getpass
import os
import time
from datetime import datetime, timedelta, timezone
import requests

BASE_URL = os.environ.get("JK_API_BASE_URL", "https://jk.cyberlative.com/api/v1/ml").rstrip("/")
TOKEN = os.environ.get("JK_ML_API_TOKEN") or getpass.getpass("Client API token: ")
HEADERS = {"Authorization": "Bearer " + TOKEN.strip()}

def get(path, params=None):
    for attempt in range(3):
        response = requests.get(BASE_URL + path, headers=HEADERS,
                                params=params, timeout=30, allow_redirects=False)
        if response.status_code == 429 and attempt < 2:
            time.sleep(max(1, int(response.headers.get("Retry-After", "60"))))
            continue
        response.raise_for_status()
        if response.status_code != 200:
            raise RuntimeError("Expected HTTP 200; check BASE_URL uses the final HTTPS address")
        return response.json()

sites = get("/sites")["sites"]
sensors = get("/sensors")["sensors"]
print("Sites:", len(sites), "Sensors:", len(sensors))
if not sensors:
    raise SystemExit("No sensors configured. Ask the administrator to add one.")

# Choose an enabled sensor. Set JK_SENSOR_ID to use a known active probe.
sensor = next((s for s in sensors if s["enabled"]), sensors[0])
SENSOR_ID = os.environ.get("JK_SENSOR_ID", sensor["id"])
print("Selected sensor ID:", SENSOR_ID)
print("Latest values:", get("/snapshot", {"sensorIds": SENSOR_ID}))

end = datetime.now(timezone.utc)
params = {"sensorIds": SENSOR_ID, "from": (end - timedelta(days=1)).isoformat(),
          "to": end.isoformat(), "resolution": "raw", "limit": 2000}
rows = []
while True:
    page = get("/readings", params)
    rows.extend(page["data"])
    if not page["hasMore"]:
        break
    params["cursor"] = page["nextCursor"]  # Keep the original filters/from/to.

print("Readings in the last 24 hours:", len(rows))
for row in rows[:5]:
    print(row)
if rows:
    with open("readings.csv", "w", newline="", encoding="utf-8") as output:
        writer = csv.DictWriter(output, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)
    print("Saved readings.csv")
else:
    print("No readings in this window. Choose another sensor or a longer period.")

# macOS/Linux, bash or zsh; requires curl and Python 3.
# The token prompt is hidden and the token is not written into shell history.
export JK_ML_API_TOKEN="$(python3 -c 'import getpass; print(getpass.getpass("Client API token: "))')"
export BASE_URL="https://jk.cyberlative.com/api/v1/ml"

# Confirm the service and discover sensor IDs.
curl --silent --show-error "$BASE_URL" | python3 -m json.tool
curl --silent --show-error "$BASE_URL/sensors" \
  --header "Authorization: Bearer $JK_ML_API_TOKEN" | python3 -m json.tool

# Select the first enabled sensor automatically; set SENSOR_ID manually to choose another.
export SENSOR_ID="$(curl --silent --show-error --fail "$BASE_URL/sensors" \
  --header "Authorization: Bearer $JK_ML_API_TOKEN" | python3 -c 'import json,sys; s=json.load(sys.stdin)["sensors"]; print(next((x["id"] for x in s if x["enabled"]), s[0]["id"] if s else ""))')"
# If SENSOR_ID is empty, configure a sensor before continuing.
export FROM="$(python3 -c 'from datetime import datetime,timedelta,timezone; print((datetime.now(timezone.utc)-timedelta(days=1)).isoformat())')"
export TO="$(python3 -c 'from datetime import datetime,timezone; print(datetime.now(timezone.utc).isoformat())')"

// Node.js 20+. Save as demo.mjs and run: node demo.mjs
// First set JK_ML_API_TOKEN with the terminal setup above.
const base = process.env.JK_API_BASE_URL || 'https://jk.cyberlative.com/api/v1/ml';
const token = process.env.JK_ML_API_TOKEN;
if (!token) throw new Error('Set JK_ML_API_TOKEN before running.');
async function get(path, params = {}) {
  const url = new URL(base.replace(/\/$/, '') + path);
  url.search = new URLSearchParams(params).toString();
  const response = await fetch(url, {
    headers: { Authorization: 'Bearer ' + token },
    signal: AbortSignal.timeout(30000), redirect: 'error',
  });
  if (!response.ok) throw new Error('HTTP ' + response.status + ': ' + await response.text());
  return response.json();
}
const { sensors } = await get('/sensors');
if (!sensors.length) throw new Error('No sensors configured.');
const sensor = sensors.find(s => s.enabled) || sensors[0];
console.log(await get('/snapshot', { sensorIds: sensor.id }));
const to = new Date();
const params = { sensorIds: sensor.id, from: new Date(to.getTime() - 86400000).toISOString(),
  to: to.toISOString(), limit: '2000', resolution: 'raw' };
const rows = [];
while (true) {
  const page = await get('/readings', params);
  rows.push(...page.data);
  if (!page.hasMore) break;
  params.cursor = page.nextCursor;
}
console.log('Rows:', rows.length, 'First five:', rows.slice(0, 5));

# Run after the Python quickstart; reuses get() and SENSOR_ID.
# Stop with Ctrl+C / the Colab stop button. No device writes are made.
import time
from datetime import datetime, timezone
params = {"sensorIds": SENSOR_ID, "since": datetime.now(timezone.utc).isoformat(),
          "resolution": "raw", "limit": 2000}
cursor = None
try:
    while True:
        while True:
            page_params = {**params, **({"cursor": cursor} if cursor else {})}
            page = get("/readings", page_params)
            for row in page["data"]:
                print(row)
            # In production, commit processed rows and this checkpoint together.
            # Keep the original since value even across empty polls.
            cursor = page["nextCursor"]
            if not page["hasMore"]:
                break
        time.sleep(30)
except KeyboardInterrupt:
    print("Polling stopped.")