Pixeltable and Modal

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

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

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

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

# pxt schema update app.py media
videos = pxt.get_table('media.videos')
videos.insert([{'video': 'lecture.mp4', 'title': 'Machine Learning 101'}])

# Query frames semantically across all stored videos
frames = pxt.get_table('media.frames')
sim = frames.frame.similarity(string='whiteboard diagram')
hits = frames.order_by(sim, asc=False).limit(5).select(frames.pos, frames.title)

import modal

app = modal.App("video-processor")
image = modal.Image.debian_slim().pip_install("torch", "transformers", "opencv-python", "pillow")

@app.function(image=image, gpu="T4", timeout=600)
def process_video_frames(video_url: str):
    import cv2
    from PIL import Image
    from transformers import CLIPProcessor, CLIPModel
    
    # 1. Download and extract frames using OpenCV
    cap = cv2.VideoCapture(video_url)
    frames = []
    # ... frame extraction logic ...
    
    # 2. Run CLIP inference on GPU
    model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32").cuda()
    processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
    # ... compute embeddings ...
    
    # 3. Must save results to external database
    # db.insert_many(...)
    return {"status": "ok", "frames_processed": len(frames)}
# Modal handles GPU execution, but you must build the storage and query layer.

import pixeltable as pxt
import modal

# Reference a Modal function
run_flux = modal.Function.lookup("image-gen", "generate_flux_image")

@pxt.udf
def modal_generate(prompt: str):
    # Executes remotely on a Modal A100 GPU. The UDF returns a PIL image.
    import io
    from PIL import Image
    image_bytes = run_flux.remote(prompt)
    return Image.open(io.BytesIO(image_bytes))

TableModel = pxt.model_base()

class Prompts(TableModel, name='prompts'):
    prompt: pxt.String
    generated_image = modal_generate(prompt)

# Inserting a row triggers remote Modal GPU execution,
# and Pixeltable stores and versions the resulting image.

# In modal_app.py
import modal

app = modal.App("image-gen")
image = modal.Image.debian_slim().pip_install("diffusers", "torch")

@app.function(image=image, gpu="A100")
def generate_flux_image(prompt: str) -> bytes:
    # Runs on Modal's A100 GPU infrastructure
    from diffusers import FluxPipeline
    import io
    
    pipe = FluxPipeline.from_pretrained("black-forest-labs/FLUX.1-schnell", torch_dtype=torch.bfloat16).to("cuda")
    out = pipe(prompt, num_inference_steps=4).images[0]
    buf = io.BytesIO()
    out.save(buf, format="PNG")
    return buf.getvalue()