Pixeltable and Mixpeek

import pixeltable as pxt
from pixeltable.functions.huggingface import clip
from pixeltable.functions.video import frame_iterator
from pixeltable.serving import FastAPIRouter

TableModel = pxt.model_base()
clip_embed = clip.using(model_id='openai/clip-vit-base-patch32')

class Videos(TableModel, name='videos'):
    title: pxt.String
    video: pxt.Video

class Frames(
    TableModel,
    name='frames',
    base=Videos,
    iterator=frame_iterator(video=Videos.video, fps=1),
):
    __indexes__ = [pxt.EmbeddingIndex(frame, embedding=clip_embed)]

@pxt.query
def search_video(query: str, limit: int = 5):
    sim = Frames.frame.similarity(string=query)
    return (
        Frames.order_by(sim, asc=False)
        .limit(limit)
        .select(title=Videos.title, frame_attrs=Frames.frame_attrs, score=sim)
    )

api = FastAPIRouter(name='video_lib')
api.add_insert_route(
    Videos, path='/videos', inputs=[Videos.title], uploadfile_inputs=[Videos.video]
)
api.add_query_route(path='/search', query=search_video)

import os
from mixpeek import Mixpeek

client = Mixpeek(api_key=os.environ['MIXPEEK_API_KEY'])
ns = client.namespaces.create(namespace_name='video-demo')
client.namespace = ns['namespace_id']

bucket = client.buckets.create(
    bucket_name='demo-videos',
    bucket_schema={'properties': {'video': {'type': 'video'}}},
)
# This quickstart creates a Mixpeek bucket and uploads into it.
# An existing S3, GCS, or Azure bucket is the separate FAQ path.
col = client.collections.create(
    collection_name='video-scenes',
    source={'type': 'bucket', 'bucket_ids': [bucket['bucket_id']]},
    features=['video_search'],
)
client.buckets.upload(
    bucket['bucket_id'],
    blobs=[{
        'property': 'video',
        'type': 'video',
        'data': 'https://example.com/lecture.mp4',
    }],
)
result = client.collections.trigger(col['collection_id'])
task = client.tasks.get(task_id=result['task_id'])
# Repeat tasks.get until task['status'] is COMPLETED or COMPLETED_WITH_ERRORS.
# Do not call collections.trigger again to poll.
ret = client.retrievers.create(
    retriever_name='video-search',
    input_schema={'query_text': {'type': 'text', 'required': True}},
    collection_identifiers=[col['collection_id']],
    stages=[{
        'stage_name': 'search',
        'stage_type': 'filter',
        'config': {
            'stage_id': 'feature_search',
            'parameters': {
                'searches': [{
                    'feature_uri': 'mixpeek://universal_extractor@v1/gemini-embedding-2',
                    'query': {'input_mode': 'text', 'value': '{{INPUT.query_text}}'},
                    'top_k': 50,
                }],
                'final_top_k': 20,
            },
        },
    }],
)
results = client.retrievers.execute(
    ret['retriever_id'],
    inputs={'query_text': 'person at a whiteboard'},
)
for doc in results['documents']:
    print(doc['document_id'], doc.get('start_time'), doc['score'])