Video Analysis & Spatial Intelligence

Key terms — defined before you touch the API

When to use Video Indexer vs Spatial Analysis

Trial vs ARM-Connected — the most commonly tested distinction

What Video Indexer extracts from your video

All insights are returned in the JSON index. Knowing what each insight is and what limitations apply is exam-tested.

Upload, index, and query — the API workflow

Key API endpoint reference

Real-time people detection in physical spaces

Deployment architecture

Spatial Analysis operations — the five you must know

Operation selection — exam decision guide

Deployment manifest — how operations are configured

Operations are configured in an IoT Edge deployment manifest JSON file. Each operation specifies the camera RTSP URL, the operation type, and the zone/line geometry as polygon coordinates.

// Simplified IoT Edge deployment manifest for Spatial Analysis
// Full manifest also includes module image, resources, routes to IoT Hub
{
  "spatialanalysis": {
    "operationId": "cognitiveservices.vision.spatialanalysis-personcrossingline",
    "version": 1,
    "enabled": true,
    "parameters": {
      // RTSP stream URL from the IP camera
      "VIDEO_URL": "rtsp://192.168.1.10:554/stream1",
      "VIDEO_SOURCE_ID": "entrance-camera-01",
      // Line defined as two points in image coordinates (x,y as fraction of frame)
      "DETECTOR_NODE_CONFIG": "{ \"lines\": [{ \"name\": \"store_entrance\", \"line\": { \"start\": {\"x\": 0.1, \"y\": 0.5}, \"end\": {\"x\": 0.9, \"y\": 0.5}}}] }",
      "SPACEANALYTICS_CONFIG": "{ \"zones\": [] }"
    }
  }
}

Upload, index, and query a video — every line explained

Video Indexer has no dedicated Python SDK — you use the requests library directly against the REST API. Install: pip install requests

import requests
import time
import os

# ── Connection details ─────────────────────────────────────────────────
# Find these in the Azure portal: your Video Indexer resource → Keys and Endpoint
# Also available at videoindexer.ai → Account Settings
ACCOUNT_ID       = os.environ.get("VIDEOINDEXER_ACCOUNT_ID")     # GUID format
ACCOUNT_LOCATION = os.environ.get("VIDEOINDEXER_LOCATION")       # e.g. "trial" or "eastus"
API_KEY          = os.environ.get("VIDEOINDEXER_API_KEY")         # Management API key (subscription key)

# Base URL for the Video Indexer API
BASE_URL = f"https://api.videoindexer.ai/{ACCOUNT_LOCATION}/Accounts/{ACCOUNT_ID}"
AUTH_URL = f"https://api.videoindexer.ai/Auth/{ACCOUNT_LOCATION}/Accounts/{ACCOUNT_ID}"


# ══════════════════════════════════════════════════════════════════════
# PART 1 — AUTHENTICATION: get an account-level access token
# ══════════════════════════════════════════════════════════════════════

def get_access_token(allow_edit: bool = True) -> str:
    """Get a JWT access token. allow_edit=True for management operations."""
    response = requests.get(
        f"{AUTH_URL}/AccessToken",
        params={"allowEdit": str(allow_edit).lower()},
        headers={"Ocp-Apim-Subscription-Key": API_KEY}
        # "Ocp-Apim-Subscription-Key" is the header name for the Management API key
        # This is different from the access token — you pass the API key HERE
        # to GET the short-lived access token that you use for all other calls
    )
    response.raise_for_status()
    # Response body is a JSON string — the token itself (with surrounding quotes)
    # .strip('"') removes those surrounding quotes
    return response.json()  # Returns the JWT string

access_token = get_access_token(allow_edit=True)
print("Access token obtained.")


# ══════════════════════════════════════════════════════════════════════
# PART 2 — UPLOAD: Submit a video for indexing
# ══════════════════════════════════════════════════════════════════════

# Option A — Upload from a URL (video is publicly accessible)
upload_response = requests.post(
    f"{BASE_URL}/Videos",
    params={
        "name"       : "quarterly-review-q1",   # Display name for the video
        "videoUrl"   : "https://example.com/videos/q1-review.mp4",  # Public URL
        "language"   : "en-US",               # Hint for speech-to-text; auto-detect if omitted
        "indexingPreset": "Default",          # "Default" (all insights) or "AudioOnly" or "VideoOnly"
        "accessToken": access_token           # Account-level token in query param (not header)
    }
)
upload_response.raise_for_status()
video = upload_response.json()
video_id = video["id"]             # GUID — needed for all subsequent calls on this video
print(f"Video submitted. ID: {video_id}  State: {video['state']}")

# Option B — Upload a local file (multipart POST)
# with open("meeting.mp4", "rb") as f:
#     upload_response = requests.post(
#         f"{BASE_URL}/Videos",
#         params={"name": "meeting", "accessToken": access_token},
#         files={"file": ("meeting.mp4", f, "video/mp4")}  # multipart file upload
#     )


# ══════════════════════════════════════════════════════════════════════
# PART 3 — POLL: Wait for indexing to complete
# ══════════════════════════════════════════════════════════════════════

def wait_for_indexing(video_id: str, token: str, poll_interval: int = 30) -> dict:
    """Poll the index endpoint until state is 'Processed' or 'Failed'."""
    while True:
        response = requests.get(
            f"{BASE_URL}/Videos/{video_id}/Index",
            params={"accessToken": token}
        )
        response.raise_for_status()
        index = response.json()
        state = index["state"]  # "Uploaded" | "Processing" | "Processed" | "Failed"

        if state == "Processed":
            print("Indexing complete!")
            return index  # The full index JSON — all insights inside
        elif state == "Failed":
            raise RuntimeError(f"Indexing failed: {index.get('failureMessage', 'unknown error')}")
        else:
            # Still processing — print progress if available
            progress = index.get("processingProgress", "unknown")
            print(f"  State: {state}  Progress: {progress}  (checking again in {poll_interval}s)")
            time.sleep(poll_interval)

index_data = wait_for_indexing(video_id, access_token)


# ══════════════════════════════════════════════════════════════════════
# PART 4 — EXTRACT INSIGHTS: Parse the index JSON
# ══════════════════════════════════════════════════════════════════════

# The full index is a large JSON object — insights are inside "videos[0].insights"
# (A single upload can contain multiple video tracks — usually just one)
insights = index_data["videos"][0]["insights"]

# ── Transcript ─────────────────────────────────────────────────────────
print("\n=== TRANSCRIPT (first 3 segments) ===")
for segment in insights.get("transcript", [])[:3]:
    # Each segment: text, confidence, speakerId, and instances (start/end times)
    text  = segment["text"]
    conf  = segment["confidence"]
    start = segment["instances"][0]["start"]  # e.g. "0:00:02.5"
    end   = segment["instances"][0]["end"]
    print(f"  [{start} → {end}] ({conf:.0%})  {text}")

# ── Keywords ───────────────────────────────────────────────────────────
print("\n=== KEYWORDS ===")
for kw in insights.get("keywords", [])[:5]:
    # Each keyword: text, confidence, instances (list of times it appeared)
    print(f"  {kw['text']}  ({kw['confidence']:.0%})")

# ── Topics ─────────────────────────────────────────────────────────────
print("\n=== TOPICS ===")
for topic in insights.get("topics", []):
    # Each topic: name (the category), confidence, iptcName (IPTC taxonomy label)
    print(f"  {topic['name']}  (IPTC: {topic.get('iaptcName', 'N/A')})  {topic['confidence']:.0%}")

# ── Scenes and Shots ───────────────────────────────────────────────────
print("\n=== SCENES ===")
for scene in insights.get("scenes", []):
    scene_id = scene["id"]
    start    = scene["instances"][0]["start"]
    end      = scene["instances"][0]["end"]
    print(f"  Scene {scene_id}: {start} → {end}")

# ── Named Entities ─────────────────────────────────────────────────────
print("\n=== NAMED ENTITIES ===")
for entity in insights.get("namedEntities", []):
    # type: "Person", "Location", "Organization", "Brand"
    print(f"  [{entity['type']}] {entity['name']}  ({entity['confidence']:.0%})")

# ── Faces (if Limited Access granted) ─────────────────────────────────
print("\n=== FACES ===")
for face in insights.get("faces", []):
    # name: identified person name (or "Unknown" if unrecognised)
    # thumbnailId: use to retrieve the face crop image via the Thumbnails endpoint
    print(f"  {face['name']}  confidence={face['confidence']:.0%}  seenFor={face['seenDuration']:.1f}s")


# ══════════════════════════════════════════════════════════════════════
# PART 5 — SEARCH: Find videos containing a term
# ══════════════════════════════════════════════════════════════════════

search_response = requests.get(
    f"{BASE_URL}/Videos/Search",
    params={
        "query"      : "quarterly revenue",     # Search across transcript, OCR, labels, topics
        "accessToken": access_token,
        "pageSize"   : 10,                    # Number of results per page
        "skip"       : 0                       # Pagination offset
    }
)
search_results = search_response.json()

print("\n=== SEARCH RESULTS ===")
for result in search_results.get("results", []):
    vid_name = result["name"]
    vid_id   = result["id"]
    # matchedText shows WHERE in the video the query was found (with context)
    print(f"  {vid_name} ({vid_id})")
    for match in result.get("matchedText", [])[:2]:
        print(f"    → at {match.get('start', '?')}: \"{match.get('text', '')}\"")


# ══════════════════════════════════════════════════════════════════════
# PART 6 — WIDGETS: Get embeddable iframe URLs
# ══════════════════════════════════════════════════════════════════════

# Get a video-level token (scoped to one video) for embedding widgets
video_token_response = requests.get(
    f"https://api.videoindexer.ai/Auth/{ACCOUNT_LOCATION}/Accounts/{ACCOUNT_ID}/Videos/{video_id}/AccessToken",
    params={"allowEdit": "false"},         # Read-only token for embedding in a web page
    headers={"Ocp-Apim-Subscription-Key": API_KEY}
)
video_token = video_token_response.json()

# Player widget — embeddable video player with transcript highlighting
player_widget = requests.get(
    f"{BASE_URL}/Videos/{video_id}/PlayerWidget",
    params={"accessToken": video_token}
)
print(f"\nPlayer widget URL: {player_widget.json()}")
# Embed in HTML: <iframe src="{player_widget.json()}" ...></iframe>

# Insights widget — side panel with faces, labels, topics timeline
insights_widget = requests.get(
    f"{BASE_URL}/Videos/{video_id}/InsightsWidget",
    params={"accessToken": video_token, "widgetType": "Keywords"}
    # widgetType: "Keywords", "Faces", "Sentiments", "Emotions", "Topics", "Labels"
    # Omit widgetType for all insights combined
)
print(f"Insights widget URL: {insights_widget.json()}")

Video Analysis — what the exam tests

Test your understanding