Anomaly Detection

import csv
import io
import time
import requests

from mixpeek import Mixpeek

client = Mixpeek(api_key="YOUR_API_KEY", namespace="inspection")
API = "https://api.mixpeek.com/v1"
HEADERS = {"Authorization": "Bearer YOUR_API_KEY", "X-Namespace": "inspection"}

# 1. A collection that embeds every inspection photo
bucket = client.buckets.create(
    bucket_name="inspection-photos",
    bucket_schema={"properties": {"photo": {"type": "image"}}},
)
collection = client.collections.create(
    collection_name="inspection-photos",
    source={"type": "bucket", "bucket_ids": [bucket["bucket_id"]]},
    feature_extractor={
        "feature_extractor_name": "multimodal_extractor",
        "version": "v1",
    },
)
client.buckets.upload(
    bucket["bucket_id"],
    blobs=[{"property": "photo", "type": "image", "data": "s3://your-bucket/inspection/line-4/frame-0001.jpg"}],
)
client.collections.trigger(collection["collection_id"])

# 2. HDBSCAN puts photos that fit no dense group into noise; those are the anomalies.
#    The SDK has no clusters resource, so this part is REST.
cluster = requests.post(API + "/clusters", headers=HEADERS, json={
    "cluster_name": "inspection-baseline",
    "collection_ids": [collection["collection_id"]],
    "cluster_type": "vector",
    "vector_config": {
        "feature_uris": ["mixpeek://multimodal_extractor@v1/vertex_multimodal_embedding"],
        "clustering_method": "hdbscan",
        "algorithm_params": {"min_cluster_size": 20},
    },
}).json()
task = requests.post(API + "/clusters/" + cluster["cluster_id"] + "/execute", headers=HEADERS, json={}).json()
while client.tasks.get(task["task_id"])["status"] not in ("COMPLETED", "COMPLETED_WITH_ERRORS", "FAILED"):
    time.sleep(15)

# 3. The latest run reports the noise ratio, and its CSV export lists every member
run = requests.get(API + "/clusters/" + cluster["cluster_id"] + "/executions", headers=HEADERS).json()
print("noise ratio:", (run.get("metrics") or {}).get("noise_ratio"))
export = requests.get(
    API + "/clusters/" + cluster["cluster_id"] + "/executions/" + run["run_id"] + "/export",
    headers=HEADERS,
    params={"format": "csv"},
)
rows = csv.DictReader(io.StringIO(export.text))
outliers = [r["document_id"] for r in rows if r["cluster_id"] in ("-1", "cl_-1") and r["is_centroid"] != "True"]
print(len(outliers), "anomalous photos:", outliers[:10])