Mikshi Python SDK

pip install videosdk

import time
from videosdk import Client

with Client(
    api_key="sk_AnFZjh95_...",
    base_url="https://api.your-host.com",
) as client:
    # 1. Upload — chunks the file, PUTs to S3, completes, and kicks off
    # transcoding + HLS generation. Returns immediately with a video_id.
    job = client.uploads.upload_file("clip.mp4", name="My Clip")
    print(f"uploaded {job.video_id}")

    # 2. Wait for upload processing (transcode + HLS) to finish.
    while True:
        status = client.uploads.get_status(job.video_id)
        derived = status.derived_status()
        if derived == "completed":
            break
        if derived == "failed":
            raise RuntimeError(status.error_message or "upload processing failed")
        time.sleep(2)

    # 3. Run analyse + search-process. These two pipelines are what make the
    # video searchable and chat-ready — search/visualize/ask all require them.
    client.videos.initiate_analyse(job.video_id)
    client.videos.initiate_search_process(job.video_id)
    while True:
        a = client.videos.get_analyse_status(job.video_id)
        sp = client.videos.get_search_process_status(job.video_id)
        if a.is_terminal() and sp.is_terminal():
            if a.is_failed() or sp.is_failed():
                raise RuntimeError(f"pipeline failed: analyse={a.status}, search={sp.status}")
            break
        time.sleep(2)

    # 4. Vector search — ranked clip-level hits with pre-signed thumbnails and
    # the auth-gated HLS URL for each segment.
    hits = client.search.search(
        "person near the door",
        video_ids=[job.video_id],
        top_k=3,
    )
    for hit in hits:
        print(f"  #{hit.rank}  [{hit.start} - {hit.end}]  {hit.hls_playlist_url}")

    # 5. Chat — natural-language Q&A over the same video.
    result = client.chat.ask(
        "Summarize what happens in the video.",
        video_id=job.video_id,
    )
    print(result.answer)

from videosdk import Client

client = Client(
    api_key="sk_...",          # required
    base_url="https://...",    # optional; defaults to http://localhost:8001
    timeout=30.0,              # optional; per-request timeout in seconds
    http_client=None,          # optional; pass a pre-configured httpx.Client
)

job = client.uploads.upload_file("movie.mp4", name="My Movie")
print(job.video_id)   # poll get_status(video_id) for processing progress

def progress(uploaded, total):
    print(f"  {uploaded}/{total} bytes ({uploaded * 100 // total}%)")

job = client.uploads.upload_file(
    "movie.mp4",
    name="My Movie",
    chunk_size=25 * 1024 * 1024,  # 25 MB chunks (default is 10 MB)
    on_progress=progress,
)

job = client.uploads.direct("movie.mp4", name="My Movie")
print(job.video_id)

job = client.uploads.import_youtube("https://www.youtube.com/watch?v=dQw4w9WgXcQ")

from videosdk import YouTubeImportError

try:
    client.uploads.import_youtube(url)
except YouTubeImportError as e:
    print(f"YouTube import rejected: {e.code}")
    if e.code == "YOUTUBE_VIDEO_PRIVATE":
        ...  # skip

import httpx

init = client.uploads.initiate_multipart("movie.mp4", "video/mp4")

parts = []
with open("movie.mp4", "rb") as f:
    part_number = 1
    while chunk := f.read(10 * 1024 * 1024):  # 10 MB chunks
        p = client.uploads.get_part_url(init.s3_object_name, init.upload_id, part_number)
        r = httpx.put(p.url, content=chunk)
        r.raise_for_status()
        parts.append({"part_number": part_number, "etag": r.headers["ETag"]})
        part_number += 1

client.uploads.complete_multipart(init.s3_object_name, init.upload_id, parts)
job = client.uploads.preprocess(init.s3_object_name, init.video_id, name="My Movie")

import time

while True:
    status = client.uploads.get_status(job.video_id)
    derived = status.derived_status()
    if derived in ("completed", "failed"):
        break
    print(
        f"  transcode={getattr(status.transcoding_status, 'percentage', 0)}%  "
        f"hls={getattr(status.hls_generation_status, 'percentage', 0)}%"
    )
    time.sleep(2)

if derived == "failed":
    raise RuntimeError(status.error_message or "processing failed")

# List with filters (only set what you need; everything is keyword-only)
videos = client.videos.list(status="processed", limit=50, sort_order="asc")

# Single-video details (includes video_metadata)
details = client.videos.get(video_id)
print(details.name, details.video_metadata)

# Kick off the analyse pipeline, then poll
client.videos.initiate_analyse(video_id)
while True:
    s = client.videos.get_analyse_status(video_id)
    if s.is_terminal():
        break
    print(f"  analyse {s.percentage or 0}%")  # percentage is None on DB fallback
    time.sleep(2)

# Permanent delete (also removes derived S3 assets)
client.videos.delete(video_id)

result = client.chat.ask(
    "Describe what happens in this video.",
    video_id="a249266f-052d-4148-bc2f-3efa76dc4269",
)
print(result.answer)

VIDEO_ID = "a249266f-052d-4148-bc2f-3efa76dc4269"
history: list[dict] = []

def turn(question: str) -> str:
    result = client.chat.ask(question, video_id=VIDEO_ID, history=history)
    history.append({"role": "user", "content": question})
    history.append({"role": "assistant", "content": result.answer})
    return result.answer

print(turn("Describe what happens in this video."))
print(turn("How many people are involved?"))
print(turn("What was my first question?"))

client.chat.ask(
    "Walk me through every event in chronological order.",
    video_id=VIDEO_ID,
    top_k=20,
)

import sys
from videosdk import AskTokenEvent, AskDoneEvent, AskErrorEvent

for event in client.chat.ask_stream(
    "How many people are involved?",
    video_id=VIDEO_ID,
):
    if isinstance(event, AskTokenEvent):
        sys.stdout.write(event.content)
        sys.stdout.flush()
    elif isinstance(event, AskDoneEvent):
        print()                       # full answer is also on event.answer
    elif isinstance(event, AskErrorEvent):
        print(f"\n[error] {event.detail}")
        break

hits = client.search.search(
    "person walking near the gate",
    video_ids=["b2f1...-1111", "b2f1...-2222"],
    top_k=5,
)

for hit in hits:
    print(
        f"#{hit.rank}  [{hit.start} - {hit.end}]  "
        f"video={hit.video_id}\n"
        f"    hls:   {hit.hls_playlist_url}\n"
        f"    thumb: {hit.thumbnail_url}"
    )

hits = client.search.search("forklift incident", collection_id=coll.collection_id)

viz = client.search.visualize(collection_id="3fa8...")

print(len(viz.videos), "videos,", len(viz.points), "points")

for video_id, info in viz.videos.items():
    print(video_id, "color=", info.color)

for p in viz.points[:5]:
    print(f"  ({p.x:.4f}, {p.y:.4f})  segment={p.segment_id}  start={p.start}")

class VisualizationVideo:
    video_id: UUID
    color: Optional[str]            # e.g. "hsl(217, 70%, 55%)"
    hls_playlist_url: Optional[str]
    video_url: Optional[str]

class VisualizationPoint:
    segment_id: str
    video_id: UUID
    x: float                # 0.0 - 1.0
    y: float                # 0.0 - 1.0
    start: str              # "HH:MM:SS"
    end: str                # "HH:MM:SS"
    thumbnail_url: Optional[str]

class Visualization:
    videos: Dict[str, VisualizationVideo]   # keyed by video_id (str)
    points: List[VisualizationPoint]

resp = client.embeddings.video("a249266f-052d-4148-bc2f-3efa76dc4269")

print(resp.status, resp.id, len(resp.data), "segments")
for seg in resp.data:
    print(
        f"  [{seg.start} - {seg.end}]  "
        f"option={seg.embedding_option}  scope={seg.embedding_scope}  "
        f"dims={len(seg.embedding)}"
    )

resp = client.embeddings.text("chain wearing man")
[vector] = resp.data
print(f"dims={len(vector.embedding)}  status={resp.status}")

import uuid
from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key="...")
pc.create_index(
    name="video-segments",
    dimension=1024,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1"),
)
index = pc.Index("video-segments")

# 1. Ingest — pull segment embeddings from our SDK, upsert into Pinecone.
resp = client.embeddings.video(video_id)
index.upsert(vectors=[
    (
        str(uuid.uuid5(uuid.NAMESPACE_URL, f"{resp.id}|{seg.start}|{seg.end}")),
        seg.embedding,
        {"video_id": str(resp.id), "start": seg.start, "end": seg.end},
    )
    for seg in resp.data
])

# 2. Query — embed the user's text with our SDK, search Pinecone.
query = client.embeddings.text("man wearing striped shirt")
[qv] = query.data
hits = index.query(vector=qv.embedding, top_k=5, include_metadata=True)
for m in hits.matches:
    print(m.score, m.metadata["video_id"], m.metadata["start"])

import uuid
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, PointStruct, VectorParams

qd = QdrantClient(url="http://localhost:6333")
qd.recreate_collection(
    collection_name="video-segments",
    vectors_config=VectorParams(size=1024, distance=Distance.COSINE),
)

# 1. Ingest.
resp = client.embeddings.video(video_id)
qd.upsert(
    collection_name="video-segments",
    points=[
        PointStruct(
            id=str(uuid.uuid5(uuid.NAMESPACE_URL, f"{resp.id}|{seg.start}|{seg.end}")),
            vector=seg.embedding,
            payload={"video_id": str(resp.id), "start": seg.start, "end": seg.end},
        )
        for seg in resp.data
    ],
)

# 2. Query.
query = client.embeddings.text("man wearing striped shirt")
[qv] = query.data
hits = qd.search(
    collection_name="video-segments",
    query_vector=qv.embedding,
    limit=5,
    with_payload=True,
)
for h in hits:
    print(h.score, h.payload["video_id"], h.payload["start"])

import weaviate
from weaviate.classes.config import Configure, Property, DataType
from weaviate.classes.query import MetadataQuery

client_wv = weaviate.connect_to_local()
client_wv.collections.create(
    name="VideoSegment",
    vectorizer_config=Configure.Vectorizer.none(),   # we supply our own vectors
    properties=[
        Property(name="video_id", data_type=DataType.TEXT),
        Property(name="start",    data_type=DataType.TEXT),
        Property(name="end",      data_type=DataType.TEXT),
    ],
)
coll = client_wv.collections.get("VideoSegment")

# 1. Ingest — Weaviate accepts our vector directly via `vector=`.
resp = client.embeddings.video(video_id)
with coll.batch.dynamic() as batch:
    for seg in resp.data:
        batch.add_object(
            properties={
                "video_id": str(resp.id),
                "start":    seg.start,
                "end":      seg.end,
            },
            vector=seg.embedding,
        )

# 2. Query.
query = client.embeddings.text("man wearing striped shirt")
[qv] = query.data
results = coll.query.near_vector(
    near_vector=qv.embedding,
    limit=5,
    return_metadata=MetadataQuery(distance=True),
)
for obj in results.objects:
    print(obj.metadata.distance, obj.properties["video_id"], obj.properties["start"])

import uuid
import chromadb

chroma = chromadb.PersistentClient(path="./chroma-db")
coll = chroma.get_or_create_collection(
    name="video-segments",
    metadata={"hnsw:space": "cosine"},
)

# 1. Ingest.
resp = client.embeddings.video(video_id)
coll.upsert(
    ids=[
        str(uuid.uuid5(uuid.NAMESPACE_URL, f"{resp.id}|{seg.start}|{seg.end}"))
        for seg in resp.data
    ],
    embeddings=[seg.embedding for seg in resp.data],
    metadatas=[
        {"video_id": str(resp.id), "start": seg.start, "end": seg.end}
        for seg in resp.data
    ],
)

# 2. Query.
query = client.embeddings.text("man wearing striped shirt")
[qv] = query.data
hits = coll.query(query_embeddings=[qv.embedding], n_results=5)
for id_, dist, meta in zip(hits["ids"][0], hits["distances"][0], hits["metadatas"][0]):
    print(dist, meta["video_id"], meta["start"])

import uuid
import psycopg
from pgvector.psycopg import register_vector

conn = psycopg.connect("dbname=video")
register_vector(conn)

conn.execute("""
    CREATE TABLE IF NOT EXISTS video_segments (
        segment_id  uuid       PRIMARY KEY,
        video_id    uuid       NOT NULL,
        start_ts    text       NOT NULL,
        end_ts      text       NOT NULL,
        embedding   vector(1024)
    );
    CREATE INDEX IF NOT EXISTS video_segments_vec_idx
        ON video_segments USING hnsw (embedding vector_cosine_ops);
""")

# 1. Ingest.
resp = client.embeddings.video(video_id)
with conn.cursor() as cur:
    for seg in resp.data:
        cur.execute(
            """
            INSERT INTO video_segments (segment_id, video_id, start_ts, end_ts, embedding)
            VALUES (%s, %s, %s, %s, %s)
            ON CONFLICT (segment_id) DO UPDATE SET embedding = EXCLUDED.embedding
            """,
            (
                uuid.uuid5(uuid.NAMESPACE_URL, f"{resp.id}|{seg.start}|{seg.end}"),
                resp.id,
                seg.start,
                seg.end,
                seg.embedding,
            ),
        )
conn.commit()

# 2. Query — `<=>` is cosine distance (smaller = closer).
query = client.embeddings.text("man wearing striped shirt")
[qv] = query.data
with conn.cursor() as cur:
    cur.execute(
        """
        SELECT segment_id, video_id, start_ts, end_ts, embedding <=> %s AS distance
        FROM video_segments
        ORDER BY embedding <=> %s
        LIMIT 5
        """,
        (qv.embedding, qv.embedding),
    )
    for segment_id, video_id, start_ts, end_ts, distance in cur.fetchall():
        print(distance, video_id, start_ts)

def hit_to_playable(video_id: str, start: str):
    details = client.videos.get(video_id)
    return f"{details.hls_playlist_path}#t={start}"

from videosdk import Client, ConflictError, RateLimitError
import time

try:
    client.collections.create(name="already-taken")
except ConflictError:
    print("name is taken")
except RateLimitError as e:
    if e.retry_after:
        time.sleep(e.retry_after)

class UploadJob:
    video_id: UUID

class TaskStatus:
    status: str          # "pending" | "in_progress" | "completed" | "failed"
    percentage: int = 0
    error: Optional[str]

class UploadStatus:
    video_id: UUID
    thumbnail_generation_status: Optional[TaskStatus]
    transcoding_status: Optional[TaskStatus]
    hls_generation_status: Optional[TaskStatus]
    error_message: Optional[str]

    def derived_status(self) -> str:
        """Returns 'pending' | 'in_progress' | 'completed' | 'failed' —
        'failed' if any pipeline failed, 'completed' if all three did,
        'in_progress' if any has started, 'pending' otherwise."""

class MultipartInitiate:
    upload_id: str
    s3_object_name: str
    video_id: UUID

class PartUrl:
    url: str
    part_number: int

class UploadedPart:
    part_number: int
    etag: str

class CompletedMultipart:
    s3_object_name: str
    upload_id: str
    etag: str

class PipelineEnqueued:
    video_id: UUID
    status: str          # "enqueued"

class PipelineStatus:
    video_id: UUID
    status: str          # see note below — not a fixed enum
    percentage: Optional[int]   # None on DB fallback (no Redis hit)
    error: Optional[str]

    def is_done(self) -> bool: ...      # status in {"done", "success", "completed"}
    def is_failed(self) -> bool: ...    # status in {"failed", "error"}
    def is_terminal(self) -> bool: ...  # is_done() or is_failed()

class PaginatedCollections:
    data: List[Collection]
    page: int
    page_size: int
    total: int

class PaginatedCollectionVideos:
    data: List[Video]
    page: int
    page_size: int
    total: int

class AskResponse:
    answer: str

class SearchHit:
    rank: int
    start: str              # "HH:MM:SS"
    end: str                # "HH:MM:SS"
    video_id: UUID
    thumbnail_url: Optional[str]
    hls_playlist_url: Optional[str]
    video_url: Optional[str]

class VisualizationVideo:
    video_id: UUID
    color: Optional[str]
    hls_playlist_url: Optional[str]
    video_url: Optional[str]

class VisualizationPoint:
    segment_id: str
    video_id: UUID
    x: float
    y: float
    start: str              # "HH:MM:SS"
    end: str                # "HH:MM:SS"
    thumbnail_url: Optional[str]

class Visualization:
    videos: Dict[str, VisualizationVideo]
    points: List[VisualizationPoint]

class VideoSegmentEmbedding:
    embedding: List[float]
    embedding_option: Optional[str]   # e.g. "visual"
    embedding_scope: Optional[str]    # e.g. "clip"
    start: Optional[str]              # "HH:MM:SS"
    end: Optional[str]                # "HH:MM:SS"

class VideoEmbeddingsResponse:
    id: UUID                          # echo of the requested video_id
    status: str                       # "ready"
    data: List[VideoSegmentEmbedding]

class TextEmbedding:
    embedding: List[float]

class TextEmbeddingResponse:
    status: str                       # "ready"
    data: List[TextEmbedding]

from videosdk import Client

client = Client(api_key="sk_...", base_url="https://api.your-host.com")

import time

def upload_and_wait(path: str, name: str, poll_interval: float = 2.0):
    job = client.uploads.upload_file(path, name=name)
    print(f"queued {job.video_id}")

    while True:
        status = client.uploads.get_status(job.video_id)
        derived = status.derived_status()
        if derived == "completed":
            break
        if derived == "failed":
            raise RuntimeError(f"upload failed: {status.error_message}")
        print(
            f"  transcode={getattr(status.transcoding_status, 'percentage', 0)}%  "
            f"hls={getattr(status.hls_generation_status, 'percentage', 0)}%"
        )
        time.sleep(poll_interval)

    return client.videos.get(job.video_id)

video = upload_and_wait("clip.mp4", name="My Clip")
print("HLS playlist:", video.hls_playlist_path)

import json
import httpx
from pathlib import Path

CHUNK = 10 * 1024 * 1024  # 10 MB
STATE_FILE = Path(".upload-state.json")

def resume_or_start(local_path: str, name: str):
    file_path = Path(local_path)
    if STATE_FILE.exists():
        state = json.loads(STATE_FILE.read_text())
        print(f"resuming upload {state['upload_id']}")
    else:
        init = client.uploads.initiate_multipart(file_path.name, "video/mp4")
        state = {
            "upload_id": init.upload_id,
            "s3_object_name": init.s3_object_name,
            "video_id": str(init.video_id),
        }
        STATE_FILE.write_text(json.dumps(state))

    already = client.uploads.list_parts(state["s3_object_name"], state["upload_id"])
    done_part_numbers = {p.part_number for p in already}
    parts = [p.model_dump() for p in already]

    with file_path.open("rb") as f, httpx.Client(timeout=300.0) as s3:
        part_number = 1
        while chunk := f.read(CHUNK):
            if part_number not in done_part_numbers:
                p = client.uploads.get_part_url(
                    state["s3_object_name"], state["upload_id"], part_number
                )
                r = s3.put(p.url, content=chunk)
                r.raise_for_status()
                parts.append({"part_number": part_number, "etag": r.headers["ETag"]})
            part_number += 1

    parts.sort(key=lambda p: p["part_number"])
    client.uploads.complete_multipart(
        state["s3_object_name"], state["upload_id"], parts
    )
    job = client.uploads.preprocess(
        state["s3_object_name"], state["video_id"], name=name
    )
    STATE_FILE.unlink()
    return job

resume_or_start("big-movie.mp4", "Big Movie")

from videosdk import YouTubeImportError

def safe_youtube_import(url: str):
    try:
        return client.uploads.import_youtube(url)
    except YouTubeImportError as e:
        if e.code == "YOUTUBE_VIDEO_PRIVATE":
            print("video is private, skipping")
            return None
        if e.code == "YOUTUBE_VIDEO_AGE_RESTRICTED":
            print("age-restricted, skipping")
            return None
        raise

from concurrent.futures import ThreadPoolExecutor

def backfill_pipelines():
    pending = [
        v for v in client.videos.list(limit=500)
        if v.transcoding_status == "completed"
        and (v.analyse_status != "done" or v.search_process_status != "done")
    ]
    print(f"backfilling {len(pending)} videos")

    def kick_off(v):
        if v.analyse_status != "done":
            client.videos.initiate_analyse(v.video_id)
        if v.search_process_status != "done":
            client.videos.initiate_search_process(v.video_id)

    with ThreadPoolExecutor(max_workers=8) as pool:
        list(pool.map(kick_off, pending))

    return [v.video_id for v in pending]

backfill_pipelines()

from datetime import datetime, timezone, timedelta

def recent_videos(limit_per_call: int = 100):
    week_ago = datetime.now(tz=timezone.utc) - timedelta(days=7)
    skip = 0
    while True:
        batch = client.videos.list(
            start_time=week_ago,
            sort_by="created_at",
            sort_order="desc",
            skip=skip,
            limit=limit_per_call,
        )
        if not batch:
            return
        yield from batch
        if len(batch) < limit_per_call:
            return
        skip += limit_per_call

for video in recent_videos():
    print(video.video_id, video.name, video.created_at)

import time
from pathlib import Path

def wait_pipelines(video_id):
    """Block until both analyse and search-process finish."""
    client.videos.initiate_analyse(video_id)
    client.videos.initiate_search_process(video_id)
    while True:
        a = client.videos.get_analyse_status(video_id)
        s = client.videos.get_search_process_status(video_id)
        if a.is_terminal() and s.is_terminal():
            if a.is_failed() or s.is_failed():
                raise RuntimeError(f"pipeline failed: analyse={a.status}, search={s.status}")
            return
        time.sleep(2)

def bulk_upload_to_collection(folder: str, collection_name: str):
    coll = client.collections.create(name=collection_name)
    print(f"created collection {coll.collection_id}")

    for path in sorted(Path(folder).glob("*.mp4")):
        video = upload_and_wait(str(path), name=path.stem)   # see "Upload a file and play it back"
        wait_pipelines(video.video_id)
        client.collections.add_video(coll.collection_id, video.video_id)
        print(f"  added {path.name}")

    return coll

bulk_upload_to_collection("./footage", "Warehouse - April")

import time
from videosdk import RateLimitError, ConflictError

def with_retries(call, *, retries: int = 5):
    for attempt in range(retries):
        try:
            return call()
        except RateLimitError as e:
            wait = e.retry_after or (2 ** attempt)
            print(f"rate limited, sleeping {wait}s")
            time.sleep(wait)
    raise RuntimeError("retries exhausted")

def get_or_create_collection(name: str):
    try:
        return with_retries(lambda: client.collections.create(name=name))
    except ConflictError:
        page = client.collections.list(page=1, page_size=100)
        for c in page.data:
            if c.name == name:
                return c
        raise

def all_collections(page_size: int = 100):
    page = 1
    yielded = 0
    while True:
        result = client.collections.list(page=page, page_size=page_size)
        if not result.data:
            return
        for c in result.data:
            yield c
            yielded += 1
        if yielded >= result.total:
            return
        page += 1

from videosdk import APIError

class Conversation:
    def __init__(self, client, *, video_id: str, top_k: int | None = None):
        self.client = client
        self.video_id = video_id
        self.top_k = top_k
        self.history: list[dict] = []

    def ask(self, question: str) -> str:
        result = self.client.chat.ask(
            question,
            video_id=self.video_id,
            top_k=self.top_k,
            history=self.history,
        )
        self.history.append({"role": "user", "content": question})
        self.history.append({"role": "assistant", "content": result.answer})
        return result.answer

chat = Conversation(client, video_id="a249266f-052d-4148-bc2f-3efa76dc4269")
print(chat.ask("Describe what happens."))
print(chat.ask("How many people are involved?"))
print(chat.ask("What was my first question?"))

hits = client.search.search(
    "package being dropped",
    collection_id=coll.collection_id,
    top_k=10,
)
if not hits:
    print("no matches")
else:
    top = hits[0]
    print(f"open {top.hls_playlist_url}#t={top.start}")

from collections import defaultdict

def grouped_visualization(collection_id):
    viz = client.search.visualize(collection_id=collection_id)
    by_video = defaultdict(list)
    for point in viz.points:
        by_video[str(point.video_id)].append(point)

    for video_id, points in by_video.items():
        info = viz.videos[video_id]
        print(f"{video_id}  color={info.color}  segments={len(points)}")
    return viz, by_video

grouped_visualization(coll.collection_id)