SDK Documentation

# Single local file
result = client.upload("video.mp4")

# Single video with metadata
result = client.upload(("video.mp4", "video.json"))

# Multiple local files
batch = client.upload(["a.mp4", "b.mp4"])

# Multiple videos with mixed metadata
batch = client.upload([
    ("video1.mp4", "video1.json"),     # Video with metadata
    "video2.mp4",                       # Video without metadata
    ("video3.mp4", "video3.json")      # Another video with metadata
])

# Public GCS URL (any accessible HTTPS URL)
remote = client.upload("https://storage.googleapis.com/my-bucket/videos/demo.mp4")

# Cloud Buckets. Must provide full gs:// URIs.
batch = client.upload([
  "gs://drive-monitor/uploads/trip-042/front.mp4",
  "gs://drive-monitor/uploads/trip-042/rear.mp4",
],
)


# With folder organization
result = client.upload("video.mp4", folder="my_folder")

# With metadata JSON file and folder
result = client.upload(("dashcam.mp4", "dashcam.json"), folder="fleet_videos")

# Organization scope
result = client.upload("launch.mp4", folder="robotics_org", scope="org")

# Single-video prompt
client.analyze(
    "abc123",
    analysis_type=AnalysisType.ASK,
    custom_event="vehicles parked on sidewalk"
)

# With thumbnails (creates annotated bounding boxes)
client.analyze(
    "abc123",
    analysis_type=AnalysisType.ASK,
    custom_event="delivery vans double parked",
    is_thumbnail=True
)

# With overlay extraction for telemetry data
from nomadicml import OverlayMode

client.analyze(
    "abc123",
    analysis_type=AnalysisType.ASK,
    custom_event="speeding events",
    overlay_mode=OverlayMode.CUSTOM  # Extracts custom fields from uploaded metadata
)

# Batch Ask: analyze multiple IDs at once
client.analyze(
    ["abc123", "def456"],
    analysis_type=AnalysisType.ASK,
    custom_event="jaywalking near intersections"
)

# Batch Ask: analyze every video in a folder
client.analyze(
    folder="fleet_uploads",
    analysis_type=AnalysisType.ASK,
    custom_event="jaywalking near intersections"
)

# General agent (edge-case detection)
client.analyze(
    "abc123",
    analysis_type=AnalysisType.GENERAL_AGENT,
)

# Lane change specialist
client.analyze(
    "abc123",
    analysis_type=AnalysisType.LANE_CHANGE,
)

# Batch agent run
client.analyze(
    ["abc123", "def456"],
    analysis_type=AnalysisType.LANE_CHANGE
)

# Batch agent run using a folder
client.analyze(
    folder="san_francisco_midday",
    analysis_type=AnalysisType.LANE_CHANGE
)

from nomadicml import NomadicML, DEFAULT_STRUCTURED_ODD_COLUMNS

client = NomadicML(api_key="your_api_key")

# Use the default schema or customise it before calling the export.
columns = [
    {
        "name": "timestamp",
        "prompt": "Log the timestamp in ISO 8601 format (placeholder date 2024-01-01).",
        "type": "YYYY-MM-DDTHH:MM:SSZ",
    },
    {
        "name": "scenery.road.type",
        "prompt": "The type of road the vehicle is on.",
        "type": "categorical",
        "literals": ["motorway", "rural", "urban_street", "parking_lot", "unpaved", "unknown"],
    },
]

odd = client.generate_structured_odd(
    video_id="VIDEO_ID",
    columns=columns or DEFAULT_STRUCTURED_ODD_COLUMNS,
)

print(odd["csv"])
print(odd.get("share_url"))

# List every integration visible to your user/org
client.cloud_integrations.list()

# Filter by provider
client.cloud_integrations.list(type="gcs")

# Add a new GCS integration using a service account JSON file
client.cloud_integrations.add(
    type="gcs",
    name="Fleet bucket",
    bucket="drive-monitor",
    prefix="uploads/",
    credentials="service-account.json",  # path or dict/bytes
)

# Add a new S3 integration 
client.cloud_integrations.add(
    type="s3",
    name="AWS archive",
    bucket="drive-archive",
    prefix="raw/",
    region="us-east-1",
    credentials={
        "accessKeyId": "...",
        "secretAccessKey": "...",
        "sessionToken": "...",  # optional
    },
)