Read your data from external software

export API_KEY="imk_..."          # shown once, at creation
export API_BASE_URL="https://iot.luminatti.online"
export PROJECT_ID="..."           # from the dashboard URL
export DEVICE_ID="..."            # the device's id, not its MQTT username

"""01 - Authenticate with an Integration API key.

External software reads IoT Manager data with an Integration API key, NOT
with a device credential and NOT with a user login:

    header:  X-API-Key: imk_...

Deliberately not `Authorization: Bearer` - a session token, a device key
and an integration key must never be interchangeable on one header.

Where to get the key: IoT Manager dashboard -> your account -> API keys
-> create. The secret is shown once, at creation. Put it in the
`API_KEY` environment variable; never commit it.

What the key can do: read-only, and only the endpoints published in the
Private Developer API - projects, devices, datastreams, alerts and the
telemetry read endpoints. Anything else answers 403, including creating
more keys.

There is no dedicated "validate this key" endpoint. The cheapest check is
listing projects: 200 means the key is live, 401 means unknown or
revoked.

Run:
    export API_KEY="imk_..."          # Windows PowerShell: $env:API_KEY="imk_..."
    python 01_auth_api_key.py

Expected result: "key OK - N project(s) visible".
"""

from __future__ import annotations

import os
import sys

import requests

API_BASE_URL = os.getenv("API_BASE_URL", "https://iot.luminatti.online")


def api_key() -> str:
    key = os.getenv("API_KEY", "").strip()
    if not key:
        raise SystemExit("Set API_KEY to an Integration API key from the dashboard (it starts with imk_).")
    return key


def main() -> int:
    resp = requests.get(
        f"{API_BASE_URL}/api/v1/projects",
        headers={"X-API-Key": api_key()},
        params={"limit": 1},
        timeout=15,
    )

    if resp.status_code == 401:
        print("401 - the key is unknown or has been revoked.", file=sys.stderr)
        return 1
    if resp.status_code == 403:
        print("403 - the key is valid but not allowed on this endpoint.", file=sys.stderr)
        return 1
    if resp.status_code == 429:
        print("429 - this key is over its request limit (120/minute). Back off and retry.", file=sys.stderr)
        return 1
    resp.raise_for_status()

    print(f"key OK - {resp.json()['total']} project(s) visible")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""02 - Walk the inventory: projects -> devices -> datastreams.

Three read endpoints, in the order you need them to discover the ids the
telemetry endpoints ask for:

    GET /api/v1/projects                              -> project ids
    GET /api/v1/projects/{project_id}/devices         -> device ids
    GET /api/v1/projects/{project_id}/datastreams     -> signal keys

`limit`/`offset` paginate the first two (devices clamps `limit` to
1-500). Telemetry is queried by `deviceId` + `signalKey`, so this is how
you turn human-readable names into the ids the rest of the samples use.

Run:
    export API_KEY="imk_..."
    export PROJECT_ID="..."      # optional; omit to walk every project
    python 02_list_projects_devices.py

Expected result: an indented inventory tree.
"""

from __future__ import annotations

import os
from typing import Any

import requests

API_BASE_URL = os.getenv("API_BASE_URL", "https://iot.luminatti.online")
PAGE_SIZE = 100


def headers() -> dict[str, str]:
    key = os.getenv("API_KEY", "").strip()
    if not key:
        raise SystemExit("Set API_KEY to an Integration API key from the dashboard.")
    return {"X-API-Key": key}


def get(path: str, **params: Any) -> Any:  # noqa: ANN401 - endpoint shapes differ
    """GET one endpoint. Most return an object; datastreams returns a bare array."""
    resp = requests.get(f"{API_BASE_URL}{path}", headers=headers(), params=params, timeout=15)
    resp.raise_for_status()
    return resp.json()


def list_projects() -> list[dict[str, Any]]:
    """Page through every project. `total` is what bounds the loop."""
    projects: list[dict[str, Any]] = []
    offset = 0
    while True:
        page = get("/api/v1/projects", limit=PAGE_SIZE, offset=offset)
        projects.extend(page["projects"])
        offset += PAGE_SIZE
        if offset >= page["total"]:
            return projects


def list_devices(project_id: str) -> list[dict[str, Any]]:
    devices: list[dict[str, Any]] = []
    offset = 0
    while True:
        page = get(f"/api/v1/projects/{project_id}/devices", limit=PAGE_SIZE, offset=offset)
        devices.extend(page["devices"])
        offset += PAGE_SIZE
        if offset >= page["total"]:
            return devices


def main() -> int:
    project_id = os.getenv("PROJECT_ID", "").strip()
    projects = [get(f"/api/v1/projects/{project_id}")] if project_id else list_projects()

    for project in projects:
        print(f"{project['name']}  ({project['id']})")
        # This endpoint answers with a bare JSON array, not a wrapper object.
        datastreams = get(f"/api/v1/projects/{project['id']}/datastreams")
        keys_by_device: dict[str, list[str]] = {}
        for ds in datastreams:
            keys_by_device.setdefault(ds["device_id"], []).append(ds["key"])

        for device in list_devices(project["id"]):
            signals = ", ".join(keys_by_device.get(device["id"], [])) or "(no datastreams)"
            print(f"  - {device['name']:<24} {device['id']}  [{device['status']}]")
            print(f"      signals: {signals}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""03 - Read the most recent values of a device.

    GET /api/v1/telemetry/latest?deviceId=...&signalKey=...&limit=...

* `deviceId` is required (422 without it).
* With `signalKey`: up to `limit` recent readings of THAT signal.
* Without `signalKey`: the `limit` most recent rows across all of the
  device's signals, ordered by time and NOT deduplicated per signal - a
  chatty signal can crowd the others out, so pass `signalKey` when you
  want one specific value.
* `limit` defaults to 50 and is not capped server-side.

Each item: `signalKey`, `value`, `unit`, `observedAt`, `quality`,
`sequence`.

Environment:
    API_KEY     Integration API key (imk_...) from the dashboard.
    DEVICE_ID   The device's id, from the dashboard URL or sample 02.
                NOTE: this is the device's own id, not the MQTT username
                that the device samples also call DEVICE_ID.
    SIGNAL_KEY  Optional; omit to see the latest across all signals.

Run:
    python 03_latest_telemetry.py

Expected result: one line per reading, newest first.
"""

from __future__ import annotations

import os

import requests

API_BASE_URL = os.getenv("API_BASE_URL", "https://iot.luminatti.online")


def main() -> int:
    api_key = os.getenv("API_KEY", "").strip()
    device_id = os.getenv("DEVICE_ID", "").strip()
    if not api_key or not device_id:
        raise SystemExit("Set API_KEY and DEVICE_ID (see this file's docstring).")

    params: dict[str, str | int] = {"deviceId": device_id, "limit": 20}
    signal_key = os.getenv("SIGNAL_KEY", "").strip()
    if signal_key:
        params["signalKey"] = signal_key

    resp = requests.get(
        f"{API_BASE_URL}/api/v1/telemetry/latest",
        headers={"X-API-Key": api_key},
        params=params,
        timeout=15,
    )
    resp.raise_for_status()
    data = resp.json()

    print(f"{data['total']} reading(s)")
    for reading in data["measurements"]:
        unit = f" {reading['unit']}" if reading["unit"] else ""
        print(
            f"{reading['observedAt']}  {reading['signalKey']:<16} "
            f"{reading['value']}{unit}  [{reading['quality']}] seq={reading['sequence']}"
        )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""04 - Read a long history, one window at a time.

    GET /api/v1/telemetry/query?deviceId=...&signalKey=...&range=24h

Two things surprise people here:

1. There is no `offset`. The history endpoint paginates by TIME, not by
   row number: pass `since` and `until` (ISO 8601 instants) to select an
   exact window, and walk the window backwards to read further. When
   both are given they override `range`.

2. The result is BUCKETED by default. The step comes from the window's
   duration alone - 1 s for <= 1h, 1 m above 1h and under 7 d, 1 h from
   7 d to under 30 d, 1 d from 30 d - and is echoed back as
   `query.bucketSeconds`. So a 30-day window returns ~30 points, not
   every reading. Pass `raw=true` to opt out of bucketing and get raw
   readings instead, capped at `limit` (default 1000, max 5000).

Points come back as `[timestamp_ms, value]` pairs.

Environment: API_KEY, DEVICE_ID, SIGNAL_KEY (see sample 03), plus
    DAYS   how far back to walk (default 7).

Run:
    python 04_history_pagination.py

Expected result: one line per day-window with its point count, then a
total.
"""

from __future__ import annotations

import os
from datetime import UTC, datetime, timedelta
from typing import Any

import requests

API_BASE_URL = os.getenv("API_BASE_URL", "https://iot.luminatti.online")
WINDOW = timedelta(days=1)


def fetch_window(
    api_key: str, device_id: str, signal_key: str, since: datetime, until: datetime
) -> list[list[Any]]:
    """One window of RAW readings. `raw=true` disables bucketing; `limit`
    is what actually bounds a raw response, so keep windows small enough
    that a single one cannot hit the 5000-row ceiling."""
    resp = requests.get(
        f"{API_BASE_URL}/api/v1/telemetry/query",
        headers={"X-API-Key": api_key},
        params={
            "deviceId": device_id,
            "signalKey": signal_key,
            "since": since.isoformat().replace("+00:00", "Z"),
            "until": until.isoformat().replace("+00:00", "Z"),
            "raw": "true",
            "limit": 5000,
        },
        timeout=30,
    )
    resp.raise_for_status()
    series = resp.json()["series"]
    return series[0]["points"] if series else []


def main() -> int:
    api_key = os.getenv("API_KEY", "").strip()
    device_id = os.getenv("DEVICE_ID", "").strip()
    signal_key = os.getenv("SIGNAL_KEY", "temperature").strip()
    if not api_key or not device_id:
        raise SystemExit("Set API_KEY and DEVICE_ID (see sample 03's docstring).")

    days = int(os.getenv("DAYS", "7"))
    until = datetime.now(UTC)
    total = 0

    # Walk backwards a window at a time. Each request is independent, so a
    # failure only costs you that window, and memory stays flat.
    for _ in range(days):
        since = until - WINDOW
        points = fetch_window(api_key, device_id, signal_key, since, until)
        total += len(points)
        print(f"{since:%Y-%m-%d %H:%M} .. {until:%Y-%m-%d %H:%M}  {len(points):>5} point(s)")
        if len(points) >= 5000:
            print("  (hit the raw-row cap - narrow the window to avoid losing readings)")
        until = since

    print(f"total: {total} point(s) over {days} day(s)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""05 - Numbers for a chart or a report: stats, and several signals at once.

Two endpoints:

    GET  /api/v1/telemetry/stats        one signal, descriptive statistics
    POST /api/v1/telemetry/query-batch  many signals' history, one request

`stats` needs `deviceId` AND `signalKey` (omitting the signal quietly
returns `{"count": 0, "message": "No data in range."}` rather than an
error), plus `range` or `since`/`until`. It is computed over the same
BUCKETED points `/telemetry/query` would return for that window, so
min/max/percentiles are over bucket averages - pass `raw=true` for exact
extremes over raw readings.

Watch the response shape: when there is no data you get `count` and
`message` only; when there is, you get the full object with
min/max/mean/median/p95/p99/stddev/first/last/delta/unit. Branch on
`count`, do not assume the keys exist.

`query-batch` takes up to 100 `{deviceId, signalKey}` pairs and returns a
`series` entry per pair, in the order you asked. One unknown deviceId
fails the WHOLE request with 404 - there is no per-item error reporting.

Environment: API_KEY, DEVICE_ID (see sample 03), plus
    SIGNAL_KEYS   comma-separated signal keys (default "temperature").

Run:
    SIGNAL_KEYS=temperature,humidity python 05_stats_for_charts.py

Expected result: a stats block per signal, then a point count per signal
from the single batch call.
"""

from __future__ import annotations

import os
from typing import Any

import requests

API_BASE_URL = os.getenv("API_BASE_URL", "https://iot.luminatti.online")
RANGE = os.getenv("RANGE", "24h")  # 1h, 6h, 24h, 7d or 30d


def fetch_stats(api_key: str, device_id: str, signal_key: str) -> dict[str, Any]:
    resp = requests.get(
        f"{API_BASE_URL}/api/v1/telemetry/stats",
        headers={"X-API-Key": api_key},
        params={"deviceId": device_id, "signalKey": signal_key, "range": RANGE},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()


def fetch_batch(api_key: str, device_id: str, signal_keys: list[str]) -> dict[str, Any]:
    resp = requests.post(
        f"{API_BASE_URL}/api/v1/telemetry/query-batch",
        headers={"X-API-Key": api_key, "Content-Type": "application/json"},
        json={
            "signals": [{"deviceId": device_id, "signalKey": key} for key in signal_keys],
            "range": RANGE,
        },
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()


def main() -> int:
    api_key = os.getenv("API_KEY", "").strip()
    device_id = os.getenv("DEVICE_ID", "").strip()
    if not api_key or not device_id:
        raise SystemExit("Set API_KEY and DEVICE_ID (see sample 03's docstring).")
    signal_keys = [key.strip() for key in os.getenv("SIGNAL_KEYS", "temperature").split(",") if key.strip()]

    for key in signal_keys:
        stats = fetch_stats(api_key, device_id, key)
        if not stats.get("count"):
            print(f"{key}: no data in the last {RANGE}")
            continue
        unit = stats.get("unit", "")
        print(
            f"{key}: n={stats['count']} min={stats['min']}{unit} max={stats['max']}{unit} "
            f"mean={stats['mean']:.2f}{unit} p95={stats['p95']}{unit} delta={stats['delta']:.2f}{unit}"
        )

    batch = fetch_batch(api_key, device_id, signal_keys)
    bucket = batch["query"]["bucketSeconds"]
    print(f"\nbatch window {batch['query']['from']} .. {batch['query']['to']} (bucketSeconds={bucket})")
    for series in batch["series"]:
        print(f"  {series['signalKey']:<16} {len(series['points'])} point(s)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""06 - Export telemetry to a CSV file.

    GET /api/v1/telemetry/export.csv?signals=deviceId:signalKey&range=30d

* `signals` is a REPEATED query parameter, each value a
  `deviceId:signalKey` pair. 1 to 8 pairs; a 9th is a 422.
* `range`, or `since`/`until`, selects the window - same rules as
  /telemetry/query.
* Rows from several signals are interleaved chronologically, not grouped
  per signal.
* The response is STREAMED. Use `stream=True` and write it out in chunks
  so a multi-month export never has to fit in memory.

Two different row shapes, and the file tells you which in a leading `#`
comment line:

    raw       timestamp,deviceKey,signalKey,unit,value,observedUntil,sampleCount
    bucketed  bucketStart,deviceKey,signalKey,unit,valueAvg,valueMin,valueMax,sampleCount

A raw row can stand for a RUN of identical consecutive readings:
`observedUntil` is when that run ended and `sampleCount` is how many
readings it represents. Wide windows switch to bucketed rows instead.

Environment: API_KEY, DEVICE_ID (see sample 03), plus
    SIGNAL_KEYS   comma-separated (default "temperature"), max 8
    RANGE         1h/6h/24h/7d/30d (default 30d)
    OUT           output path (default telemetry_export.csv)

Run:
    python 06_export_csv.py

Expected result: "wrote N bytes to telemetry_export.csv" and a readable
first line telling you whether the rows are raw or bucketed.
"""

from __future__ import annotations

import os
from pathlib import Path

import requests

API_BASE_URL = os.getenv("API_BASE_URL", "https://iot.luminatti.online")
MAX_SIGNALS = 8


def main() -> int:
    api_key = os.getenv("API_KEY", "").strip()
    device_id = os.getenv("DEVICE_ID", "").strip()
    if not api_key or not device_id:
        raise SystemExit("Set API_KEY and DEVICE_ID (see sample 03's docstring).")

    signal_keys = [key.strip() for key in os.getenv("SIGNAL_KEYS", "temperature").split(",") if key.strip()]
    if len(signal_keys) > MAX_SIGNALS:
        raise SystemExit(f"At most {MAX_SIGNALS} signals per export; got {len(signal_keys)}.")

    out_path = Path(os.getenv("OUT", "telemetry_export.csv"))
    params = [("signals", f"{device_id}:{key}") for key in signal_keys]
    params.append(("range", os.getenv("RANGE", "30d")))

    written = 0
    with requests.get(
        f"{API_BASE_URL}/api/v1/telemetry/export.csv",
        headers={"X-API-Key": api_key},
        params=params,
        timeout=300,
        stream=True,  # never buffer a whole export in memory
    ) as resp:
        resp.raise_for_status()
        with out_path.open("wb") as handle:
            for chunk in resp.iter_content(chunk_size=64 * 1024):
                handle.write(chunk)
                written += len(chunk)

    print(f"wrote {written} bytes to {out_path}")
    with out_path.open(encoding="utf-8") as handle:
        print(handle.readline().rstrip())  # the leading "#" line: raw or bucketed
        print(handle.readline().rstrip())  # the header row
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

{
  "code": "RESOURCE_NOT_FOUND",
  "message": "Not found.",
  "retryable": false,
  "details": {}
}

"""07 - Handle errors and rate limits properly.

Every failed request answers with the same JSON envelope:

    {"code": "RESOURCE_NOT_FOUND", "message": "Not found.",
     "retryable": false, "details": {}}

Branch on `code`, not on the human-readable `message` (it is translated
per the caller's language). `retryable` tells you whether trying again
can possibly help.

The statuses you will actually meet with an Integration API key:

    401  the key is unknown or revoked - stop, do not retry
    403  the key is valid but out of scope for this endpoint - stop
    404  no such project/device - stop
    422  a required parameter is missing or malformed - stop
    429  over the per-key limit of 120 requests/minute - back off
    5xx  server-side; retry with exponential backoff

429 and 5xx are the only ones worth retrying. This sample retries them
with exponential backoff and jitter, and gives up on everything else
immediately - a retry loop over a 401 just burns your allowance.

The per-key limit is 120 requests per minute (2/s sustained), counted per
key. That is far above any useful polling rate: the fastest plan only
produces a measurement every 10 s, so poll on your data's cadence rather
than in a tight loop.

Run:
    python 07_errors_and_rate_limits.py

Expected result: a demonstration of a successful call, then a deliberate
404 and a deliberate 422, each reported without a retry storm.
"""

from __future__ import annotations

import os
import random
import time
from typing import Any

import requests

API_BASE_URL = os.getenv("API_BASE_URL", "https://iot.luminatti.online")
MAX_ATTEMPTS = 5
RETRYABLE_STATUSES = frozenset({429, 500, 502, 503, 504})


class ApiError(Exception):
    """A non-retryable API failure, carrying the machine-readable code."""

    def __init__(self, status: int, code: str, message: str) -> None:
        super().__init__(f"HTTP {status} {code}: {message}")
        self.status = status
        self.code = code


def _error_fields(resp: requests.Response) -> tuple[str, str]:
    """Pull `code`/`message` out of an error body, tolerating a non-JSON
    response (a proxy 502, say, never has the envelope)."""
    try:
        body = resp.json()
    except ValueError:
        return "UNKNOWN", resp.text[:200]
    return str(body.get("code", "UNKNOWN")), str(body.get("message", ""))


def request_with_retry(method: str, path: str, **kwargs: Any) -> Any:  # noqa: ANN401
    api_key = os.getenv("API_KEY", "").strip()
    if not api_key:
        raise SystemExit("Set API_KEY to an Integration API key from the dashboard.")

    delay = 1.0
    for attempt in range(1, MAX_ATTEMPTS + 1):
        resp = requests.request(
            method, f"{API_BASE_URL}{path}", headers={"X-API-Key": api_key}, timeout=30, **kwargs
        )
        if resp.ok:
            return resp.json()

        code, message = _error_fields(resp)
        if resp.status_code not in RETRYABLE_STATUSES or attempt == MAX_ATTEMPTS:
            raise ApiError(resp.status_code, code, message)

        # Honour Retry-After when the server sends one; otherwise back off
        # exponentially with jitter so parallel clients do not resynchronise.
        wait = float(resp.headers.get("Retry-After", delay)) + random.uniform(0, 0.5)
        print(f"  {resp.status_code} {code} - retrying in {wait:.1f}s (attempt {attempt}/{MAX_ATTEMPTS})")
        time.sleep(wait)
        delay = min(delay * 2, 30.0)

    raise ApiError(0, "UNREACHABLE", "retry loop exhausted")


def main() -> int:
    print("1. a call that should work:")
    data = request_with_retry("GET", "/api/v1/projects", params={"limit": 1})
    print(f"   ok - {data['total']} project(s)\n")

    print("2. a device that does not exist (expect RESOURCE_NOT_FOUND, no retry):")
    try:
        request_with_retry("GET", "/api/v1/telemetry/latest", params={"deviceId": "no-such-device"})
    except ApiError as exc:
        print(f"   handled: {exc}\n")

    print("3. a missing required parameter (expect MESSAGE_INVALID, no retry):")
    try:
        request_with_retry("GET", "/api/v1/telemetry/latest")
    except ApiError as exc:
        print(f"   handled: {exc}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())