Developer Quickstart

import json
import os
import sys

from archetypeai.api_client import ArchetypeAI

MODEL = "Newton::c2_6_8b_fp8_260424d7a55d5e"


def make_client():
    # The /v0.5 version path is required — without it every call 404s.
    endpoint = os.environ.get(
        "ATAI_API_ENDPOINT", "https://api.u1.archetypeai.app/v0.5").rstrip("/")
    if not endpoint.endswith("/v0.5"):
        endpoint += "/v0.5"
    return ArchetypeAI(os.environ["ATAI_API_KEY"], api_endpoint=endpoint)


def main():
    client = make_client()

    # Analyze the sample video, or pass your own .mp4 as an argument:
    #   python fusion_example.py your_video.mp4
    video_path = sys.argv[1] if len(sys.argv) > 1 else "delivery.mp4"

    # Upload the video. The returned file_id is the filename — reference
    # the file by this (not the fil_... file_uid) in your query.
    upload = client.files.local.upload(video_path)
    file_id = upload["file_id"]

    body = {
        "model": MODEL,
        "query": "Describe the actions in the video.",
        "instruction_prompt": "You are a perception agent analyzing activity "
                              "in front of a house. Reply in 2-3 sentences.",
        "file_ids": [file_id],
        "max_frames": 32,   # frames sampled server-side (default 32, max 64)
        "max_new_tokens": 256,
    }
    result = client.requests_post(
        f"{client.api_endpoint}/query",
        data_payload=json.dumps(body),
        additional_headers={"Content-Type": "application/json"},
    )

    # The model's answer is at response.response[0].
    print(result["response"]["response"][0])


if __name__ == "__main__":
    main()

import csv
import json
import math
import os
from concurrent.futures import ThreadPoolExecutor

from archetypeai.api_client import ArchetypeAI

MODEL = "OmegaEncoder::omega_embeddings_1_4"
WINDOW = 1024      # timesteps per window (the encoder handles 16-1024)
CHANNEL = "bearing_1"  # the bearing that fails in this experiment
TEST_WINDOWS = 20  # held-out windows to classify (~1000 available)


def make_client():
    # The /v0.5 version path is required — without it every call 404s.
    endpoint = os.environ.get(
        "ATAI_API_ENDPOINT", "https://api.u1.archetypeai.app/v0.5").rstrip("/")
    if not endpoint.endswith("/v0.5"):
        endpoint += "/v0.5"
    return ArchetypeAI(os.environ["ATAI_API_KEY"], api_endpoint=endpoint)


def read_column(csv_path, column, cast=float):
    """Read one column from a CSV into a list."""
    with open(csv_path, newline="") as csv_file:
        reader = csv.reader(csv_file)
        index = next(reader).index(column)
        return [cast(row[index]) for row in reader]


def embed(client, values):
    """Embed one window (16-1024 timesteps) into a 768-d vector."""
    body = {
        "model": MODEL,
        "query": "",
        "normalize_input": False,
        "events": [
            {"type": "data.numeric_array", "event_data": {"contents": [values]}}
        ],
    }
    result = client.requests_post(
        f"{client.api_endpoint}/query",
        data_payload=json.dumps(body),
        additional_headers={"Content-Type": "application/json"},
    )
    return result["response"]["response"]


def knn_predict(embedding, library, k=3):
    """Majority vote among the k nearest labeled library embeddings."""
    distances = sorted(
        (math.dist(embedding, lib_embedding), label)
        for lib_embedding, label in library
    )
    votes = [label for _, label in distances[:k]]
    return max(set(votes), key=votes.count)


def main():
    client = make_client()

    # 1. Build a small n-shot library from the labeled shot files.
    #    Normalize every window with ONE scaler fit on the library pool —
    #    per-window normalization would erase the fault's amplitude signal.
    healthy = read_column("bearing_healthy.csv", CHANNEL)
    degraded = read_column("bearing_degraded.csv", CHANNEL)
    pool = healthy + degraded
    mean = sum(pool) / len(pool)
    std = (sum((value - mean) ** 2 for value in pool) / len(pool)) ** 0.5

    def scale(values):
        return [(value - mean) / std for value in values]

    starts = range(0, len(healthy) - WINDOW + 1, 256)  # overlapping shots
    shots = [(scale(healthy[s:s + WINDOW]), "healthy") for s in starts]
    shots += [(scale(degraded[s:s + WINDOW]), "degraded") for s in starts]

    with ThreadPoolExecutor(max_workers=8) as pool_executor:
        vectors = list(pool_executor.map(
            lambda shot: embed(client, shot[0]), shots))
    library = [(vector, label) for vector, (_, label) in zip(vectors, shots)]
    print(f"n-shot library: {len(library)} labeled windows")

    # 2. Classify held-out windows the library has never seen.
    #    bearing_inference.csv has no label column, so leakage is impossible.
    timestamps = read_column("bearing_inference.csv", "timestamp", cast=int)
    series = read_column("bearing_inference.csv", CHANNEL)

    # Non-overlapping windows that don't span a gap in the recording,
    # evenly spread across the timeline.
    all_starts = [
        start for start in range(0, len(series) - WINDOW + 1, WINDOW)
        if timestamps[start + WINDOW - 1] - timestamps[start] == WINDOW - 1
    ]
    stride = max(1, len(all_starts) // TEST_WINDOWS)
    test_starts = all_starts[::stride][:TEST_WINDOWS]

    with ThreadPoolExecutor(max_workers=8) as pool_executor:
        predictions = list(pool_executor.map(
            lambda start: knn_predict(
                embed(client, scale(series[start:start + WINDOW])), library),
            test_starts))

    # 3. Score against the ground-truth labels (used ONLY for scoring).
    truth_by_timestamp = dict(zip(
        read_column("bearing_labels.csv", "timestamp", cast=int),
        read_column("bearing_labels.csv", "label", cast=str)))
    truths = [truth_by_timestamp[timestamps[start]] for start in test_starts]

    for start, pred, truth in zip(test_starts, predictions, truths):
        marker = "ok" if pred == truth else "MISS"
        print(f"  ts={timestamps[start]:>8}  predicted={pred:<8} truth={truth:<8} {marker}")

    # Metrics with "degraded" as the positive class: did we catch the fault?
    pairs = list(zip(predictions, truths))
    correct = sum(pred == truth for pred, truth in pairs)
    true_pos = sum(pred == truth == "degraded" for pred, truth in pairs)
    true_neg = sum(pred == truth == "healthy" for pred, truth in pairs)
    false_pos = sum(pred == "degraded" and truth == "healthy" for pred, truth in pairs)
    false_neg = sum(pred == "healthy" and truth == "degraded" for pred, truth in pairs)
    precision = true_pos / (true_pos + false_pos) if true_pos + false_pos else 0.0
    recall = true_pos / (true_pos + false_neg) if true_pos + false_neg else 0.0
    f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0.0
    print("confusion matrix (rows: truth, cols: predicted)")
    print(f"{'':>16}{'healthy':>9}{'degraded':>10}")
    print(f"{'healthy':>16}{true_neg:>9}{false_pos:>10}")
    print(f"{'degraded':>16}{false_neg:>9}{true_pos:>10}")
    print(f"accuracy : {correct}/{len(pairs)} = {correct / len(pairs):.0%}")
    print(f"precision: {precision:.3f}  recall: {recall:.3f}  f1: {f1:.3f}"
          f"  (positive class: degraded)")


if __name__ == "__main__":
    main()