Skip to main content

Code Examples

This page provides working code examples for common operations with the Valossa Core API and metadata.

API Operations​

Submit a Video for Analysis​

curl​

curl -X POST \
  -H "Content-Type: application/json" \
  -d '{
    "api_key": "YOUR_API_KEY",
    "media": {
      "video": {
        "url": "https://example.com/video.mp4"
      }
    }
  }' \
  https://api-eu.valossa.com/core/1.0/new_job

Python​

import requests

response = requests.post(
    "https://api-eu.valossa.com/core/1.0/new_job",
    json={
        "api_key": "YOUR_API_KEY",
        "media": {
            "video": {"url": "https://example.com/video.mp4"}
        }
    }
)
job_id = response.json()["job_id"]
print(f"Job ID: {job_id}")

JavaScript​

const response = await fetch("https://api-eu.valossa.com/core/1.0/new_job", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    api_key: "YOUR_API_KEY",
    media: { video: { url: "https://example.com/video.mp4" } }
  })
});
const { job_id } = await response.json();
console.log("Job ID:", job_id);

Check Job Status​

curl​

curl "https://api-eu.valossa.com/core/1.0/job_status?api_key=YOUR_API_KEY&job_id=JOB_ID"

Python​

status = requests.get(
    "https://api-eu.valossa.com/core/1.0/job_status",
    params={"api_key": "YOUR_API_KEY", "job_id": job_id}
).json()
print(f"Status: {status['status']}")

Download Results​

curl​

curl "https://api-eu.valossa.com/core/1.0/job_results?api_key=YOUR_API_KEY&job_id=JOB_ID" -o results.json

Python​

metadata = requests.get(
    "https://api-eu.valossa.com/core/1.0/job_results",
    params={"api_key": "YOUR_API_KEY", "job_id": job_id}
).json()

Reading Metadata​

Loop Over All Detections by Type​

Python​

import json

with open("core_metadata.json", "r") as f:
    metadata = json.load(f)

for det_type, det_ids in metadata["detection_groupings"]["by_detection_type"].items():
    print(f"\n{'='*40}")
    print(f"Detection type: {det_type} ({len(det_ids)} detections)")
    print(f"{'='*40}")

    for det_id in det_ids[:5]:  # First 5 per type
        detection = metadata["detections"][det_id]
        print(f"  [{det_id}] {detection['label']}")

        # Print occurrences if available
        if "occs" in detection:
            for occ in detection["occs"]:
                c = occ.get("c_max", occ.get("c", "N/A"))
                print(f"    {occ['ss']:.1f}s - {occ['se']:.1f}s (confidence: {c})")

JavaScript​

const fs = require("fs");
const metadata = JSON.parse(fs.readFileSync("core_metadata.json", "utf-8"));

for (const [detType, detIds] of Object.entries(metadata.detection_groupings.by_detection_type)) {
  console.log(`\n${"=".repeat(40)}`);
  console.log(`Detection type: ${detType} (${detIds.length} detections)`);

  for (const detId of detIds.slice(0, 5)) {
    const detection = metadata.detections[detId];
    console.log(`  [${detId}] ${detection.label}`);
  }
}

Read Face Detections with Identity​

Python​

face_ids = metadata["detection_groupings"]["by_detection_type"].get("human.face", [])

for det_id in face_ids:
    detection = metadata["detections"][det_id]
    attrs = detection.get("a", {})

    gender_value = attrs.get("gender", {}).get("value", "unknown")
    gender_conf = attrs.get("gender", {}).get("c", 0)
    print(f"Face {det_id}: gender={gender_value} ({gender_conf:.2f})")

    if "similar_to" in attrs:
        for match in attrs["similar_to"]:
            print(f"  Identified as: {match['name']} (confidence: {match['c']:.2f})")

Read External Ontology References​

Python​

visual_ids = metadata["detection_groupings"]["by_detection_type"].get("visual.context", [])

for det_id in visual_ids[:10]:
    detection = metadata["detections"][det_id]
    refs = detection.get("ext_refs", {})

    wikidata_id = refs.get("wikidata", {}).get("id", "N/A")
    gkg_id = refs.get("gkg", {}).get("id", "N/A")
    print(f"{detection['label']}: Wikidata={wikidata_id}, Google KG={gkg_id}")

Read IAB Categories​

Python​

for det_type, det_ids in metadata["detection_groupings"]["by_detection_type"].items():
    if det_type.startswith("topic.iab"):
        for det_id in det_ids:
            detection = metadata["detections"][det_id]
            iab = detection["ext_refs"]["iab"]
            print(f"IAB: {detection['label']}")
            print(f"  ID: {iab['id']}")
            print(f"  Hierarchy: {' > '.join(iab['labels_hierarchy'])}")

Extract Audio Keywords​

Python​

for det_type, det_ids in metadata["detection_groupings"]["by_detection_type"].items():
    if det_type.startswith("audio.keyword."):
        print(f"\n{det_type}:")
        for det_id in det_ids:
            detection = metadata["detections"][det_id]
            print(f"  {detection['label']}")

Time-Based Access (What Happens at Each Second)​

Python​

for sec_index, sec_data in enumerate(metadata["detection_groupings"]["by_second"]):
    if sec_index > 10:  # First 10 seconds only
        break

    print(f"\n--- Second {sec_index} ---")
    for item in sec_data:
        det_id = item["d"]
        detection = metadata["detections"][det_id]
        confidence = item.get("c", "N/A")
        print(f"  {detection['t']}: {detection['label']} (c={confidence})")

Complete End-to-End Example​

Python: Analyze and Extract Summary​

import requests
import time
import json

API_KEY = "YOUR_API_KEY"
VIDEO_URL = "https://example.com/video.mp4"

# Step 1: Submit
print("Submitting video...")
job = requests.post(
    "https://api-eu.valossa.com/core/1.0/new_job",
    json={"api_key": API_KEY, "media": {"video": {"url": VIDEO_URL}}}
).json()
job_id = job["job_id"]
print(f"Job ID: {job_id}")

# Step 2: Wait
print("Waiting for analysis...")
while True:
    status = requests.get(
        "https://api-eu.valossa.com/core/1.0/job_status",
        params={"api_key": API_KEY, "job_id": job_id}
    ).json()
    print(f"  Status: {status['status']}")
    if status["status"] == "finished":
        break
    if status["status"] == "error":
        print("Error occurred!")
        exit(1)
    time.sleep(15)

# Step 3: Download
print("Downloading results...")
metadata = requests.get(
    "https://api-eu.valossa.com/core/1.0/job_results",
    params={"api_key": API_KEY, "job_id": job_id}
).json()

# Step 4: Summarize
duration = metadata["media_info"]["technical"]["duration_s"]
print(f"\nVideo duration: {duration:.0f}s")

for det_type in ["visual.context", "audio.context", "human.face", "topic.iab"]:
    ids = metadata["detection_groupings"]["by_detection_type"].get(det_type, [])
    if ids:
        labels = [metadata["detections"][d]["label"] for d in ids[:5]]
        print(f"{det_type}: {', '.join(labels)}")