SDK Documentation

# Single multi-view set
multi = client.upload(
  {
    "front": "https://example.com/trip-042/front.mp4",
    "left": "https://example.com/trip-042/left.mp4",
    "right": "https://example.com/trip-042/right.mp4",
  },
  folder="fleet_uploads",
  scope="org",
)

# Multiple multi-view sets in one call
multi_batch = client.upload([
  {
    "front": "https://example.com/set1/front.mp4",
    "left": "https://example.com/set1/left.mp4",
    "right": "https://example.com/set1/right.mp4",
  },
  {
    "front": "https://example.com/set2/front.mp4",
    "left": "https://example.com/set2/left.mp4",
    "right": "https://example.com/set2/right.mp4",
  },
])

# Single cloud multi-view set
multi = client.upload(
  {
    "front": "s3://drive-monitor/uploads/trip-042/front.mp4",
    "left": "s3://drive-monitor/uploads/trip-042/left.mp4",
    "right": "s3://drive-monitor/uploads/trip-042/right.mp4",
  },
  folder="fleet_uploads",
  scope="org",
)
# Returns: {"import_job_id": "ij_xxx", "status": "importing"}

upload = client.upload(
    [
        "s3://drive-monitor-archive/2024-09-01/front.mp4",
        "s3://drive-monitor-archive/2024-09-01/rear.mp4",
    ],
    wait_for_uploaded=False,
)

job = client.video.get_import_job(upload["import_job_id"])
print(job["job_id"], job["total"])

upload = client.upload(
    [
        "gs://drive-monitor/uploads/trip-042/front.mp4",
        "gs://drive-monitor/uploads/trip-042/rear.mp4",
    ],
    wait_for_uploaded=False,
)

cursor = None  # First page: no cursor
while True:
    result = client.video.get_import_job_videos(
        upload["import_job_id"],
        limit=500,
        cursor=cursor,  # Pass cursor from previous response
    )

    print(result["import_job_id"])
    print(result["video_count"], result["has_more"], result["next_cursor"])  # page_size, more pages?, next anchor
    print(result["videos"][:3])  # [{video_id, status, import_source_uri}, ...]

    if not result["has_more"]:
        break
    cursor = result["next_cursor"]  # Use this value for the next page

import time

# 1. Start the cloud import — returns an import_job_id, NOT a video_id
result = client.upload([
    "s3://my-bucket/videos/clip_001.mp4",
    "s3://my-bucket/videos/clip_002.mp4",
    "s3://my-bucket/videos/clip_003.mp4",
])
print(result)
# {"import_job_id": "ij_a1b2c3d4e5f6", "status": "importing"}

job_id = result["import_job_id"]

# 2. Poll import-job videos until returned IDs reach the requested total
#    and each returned row is terminal.
TERMINAL = {"UPLOADED", "UPLOADING_FAILED"}
seen = {}
while True:
    job = client.video.get_import_job(job_id)
    target_total = int(job.get("total") or 0)
    cursor = None
    while True:
        page = client.video.get_import_job_videos(job_id, limit=500, cursor=cursor)
        for row in page["videos"]:
            seen[row["video_id"]] = row["status"]
        if not page["has_more"]:
            break
        cursor = page["next_cursor"]

    if target_total > 0:
        all_ids_materialized = len(seen) >= target_total
        all_terminal = all(status in TERMINAL for status in seen.values())
        print(f"Import progress: ids={len(seen)}/{target_total}")
        if all_ids_materialized and all_terminal:
            break
    time.sleep(10)

# 3. Use uploaded IDs for analysis
video_ids = [vid for vid, status in seen.items() if status == "UPLOADED"]

print(f"Imported {len(video_ids)} videos: {video_ids[:5]}...")

# 4. Now use the video IDs for analysis
client.analyze(
    video_ids,
    analysis_type=AnalysisType.ASK,
    custom_event="lane departure",
)

# Get all videos
videos = client.my_videos()

# Get videos in specific folder
videos = client.my_videos(folder="my_folder")

# Get videos from a personal folder (when org folder has same name)
videos = client.my_videos(folder="shared_folder", scope="user")

# Get videos from an organization folder
videos = client.my_videos(folder="shared_folder", scope="org")

client.delete_video("video_id")

marketing = client.create_folder("marketing", description="Q1 campaign")

folder = client.get_folder("fleet_uploads")
org_folder = client.get_folder("fleet_uploads", scope="org")

results = client.search(
    query="red pickup truck overtaking",
    folder_name="my_fleet_uploads",
    scope="org",                 # optional, defaults to "user"
)

print(results["summary"])
for thought in results["thoughts"]:
    print("•", thought)

# Get all results from a batch
batch_results = client.get_batch_analysis("batch_id")

# Filter for only approved events
approved_only = client.get_batch_analysis(
    "batch_id",
    filter="approved"
)

# Filter for multiple statuses
pending_and_rejected = client.get_batch_analysis(
    "batch_id",
    filter=["pending", "rejected"]
)

# Return event-level CSV instead of JSON
csv_text = client.get_batch_analysis("batch_id", as_csv=True)

# Filter + CSV (only approved events are included)
approved_csv = client.get_batch_analysis(
    "batch_id",
    filter="approved",
    as_csv=True
)

# Save CSV to disk
with open("batch-results-approved.csv", "w", encoding="utf-8", newline="") as f:
    f.write(approved_csv)

# Access batch metadata
print(batch_results["batch_metadata"]["batch_type"])
print(batch_results["batch_metadata"]["batch_viewer_url"])

# Iterate through video results
for result in batch_results["results"]:
    print(f"Video: {result['video_id']}")
    print(f"Events: {len(result['events'])}")
    for event in result["events"]:
        print(f"  - {event.get('label', '')} at {event.get('t_start', '')}-{event.get('t_end', '')}")

# Add metadata to a batch
client.add_batch_metadata(
    "batch_id",
    {
        "experiment_id": "exp-001",
        "version": 2,
        "model": "Nomadic-VL-XLarge",
        "notes": "Test run with new parameters"
    }
)

# Update existing metadata (new keys will be merged, existing keys overwritten)
client.add_batch_metadata(
    "batch_id",
    {
        "version": 3,
        "status": "completed"
    }
)

# Retrieve metadata later
batch_results = client.get_batch_analysis("batch_id")
metadata = batch_results["batch_metadata"]["metadata"]
print(f"Experiment: {metadata.get('experiment_id')}")
print(f"Version: {metadata.get('version')}")

# Get all thumbnail URLs from an analysis
visuals = client.get_visuals("video_id", "analysis_id")

# Returns list of thumbnail URLs
# ['https://storage.googleapis.com/.../event_0_thumb.jpg',
#  'https://storage.googleapis.com/.../event_1_thumb.jpg']

# Get thumbnail for the first event (index 0)
thumbnail_url = client.get_visual("video_id", "analysis_id", 0)

# Get thumbnail for the third event (index 2)
thumbnail_url = client.get_visual("video_id", "analysis_id", 2)