# Project export: AdAtlas AI
This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.
## Project metadata
- Hackathon: Cal Hacks 12.0
- Tagline: AdAtlas AI — Turning Ad Creatives into Measurable Intelligence
- Devpost: https://devpost.com/software/adatlas-ai
- GitHub: https://github.com/2006-sk/backcal
- Video: https://www.youtube.com/embed/KgI6P70oCVk?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — 2006-sk (5 commits)
## Devpost submission (written by the team)
### Inspiration
Ad performance is often measured by surface-level metrics — impressions, clicks, conversions — but those only reveal outcomes, not causes. We were driven by one question: what if ad platforms could understand why certain creatives succeed? Most systems measure audience behavior, not creative quality. We wanted to quantify the emotional tone, color balance, and pacing of ads, and link these to measurable performance signals such as CTR (Click-Through Rate) and virality. This inspired us to build AdAtlas AI, a platform that fuses vision, audio, and marketing intelligence into a unified creative understanding framework.
### What it does
AdAtlas AI interprets ad creatives across multiple modalities visual, audio, and emotional and extracts performance-related insights. In under five minutes, it analyzes any video or image ad and provides a breakdown of its creative effectiveness. It identifies: Emotional tone and mood progression Emotional tone and mood progression Object and brand element detection Object and brand element detection Speech pacing and sentiment alignment Speech pacing and sentiment alignment Color and visual consistency Color and visual consistency Predicted CTR and virality scores (via Reka QuickTag) Predicted CTR and virality scores (via Reka QuickTag) Cross-creative similarity trends (via ChromaDB) Cross-creative similarity trends (via ChromaDB) These insights can directly inform recommendation models, ad ranking algorithms, and creative benchmarking dashboards.
### How we built it
Backend — FastAPI + Asynchronous AI Pipeline Our backend, built with FastAPI, uses an asynchronous event-driven pipeline for parallelized multimodal processing. FFmpeg extracts frames and audio tracks from videos, ensuring consistent temporal sampling. OpenCV computes color-tone averages, motion intensity, and scene transitions. Gemini 2.5 Flash/Pro performs frame-level object detection, mood classification, and narrative summarization. Groq Whisper transcribes and embeds speech, generating semantic and tonal representations. Reka QuickTag predicts CTR, virality, and emotional tone distribution, linking creative elements to marketing performance. ChromaDB Cloud stores and indexes embeddings for similarity search and trend comparison. Each subsystem runs concurrently, with data merging at the orchestration layer. Average analysis time: 2.5 minutes per creative. Frontend — React + Modern UI Frameworks The frontend was built with React.js, HTML, Tailwind CSS, and Vanilla JavaScript for speed and responsiveness. We combined multiple UI frameworks ShadCNn UI, Acertinity UI, Cult UI, Magic UI, and Material UI to create a consistent design system. Frontend highlights: Upload ad creatives (video or image) for instant processing Upload ad creatives (video or image) for instant processing Real-time progress visualization Real-time progress visualization Emotion, tone, and visual object summaries Emotion, tone, and visual object summaries CTR and virality graphs CTR and virality graphs Interactive similarity map powered by embedding distances Interactive similarity map powered by embedding distances Color and narrative progression charts
### Challenges we ran into
Color and narrative progression charts Challenges we ran into Multimodal Synchronization: Aligning visual, auditory, and semantic data streams in real time. Multimodal Synchronization: Aligning visual, auditory, and semantic data streams in real time. Latency Management: Balancing asynchronous API calls across Gemini, Groq, and Reka while preserving throughput. Latency Management: Balancing asynchronous API calls across Gemini, Groq, and Reka while preserving throughput. Visualization Overload: Presenting dense analytical data in an intuitive, non-technical interface. Visualization Overload: Presenting dense analytical data in an intuitive, non-technical interface. Embedding Storage: Optimizing ChromaDB queries for fast creative similarity results. Embedding Storage: Optimizing ChromaDB queries for fast creative similarity results.
### Accomplishments we're proud of
Engineered a fully asynchronous multimodal analysis pipeline integrating five AI systems. Engineered a fully asynchronous multimodal analysis pipeline integrating five AI systems. Reduced total processing time to under 5 minutes per ad. Reduced total processing time to under 5 minutes per ad. Designed embedding-based creative similarity using ChromaDB Cloud. Designed embedding-based creative similarity using ChromaDB Cloud. Developed a responsive, elegant frontend dashboard using seven integrated UI frameworks. Developed a responsive, elegant frontend dashboard using seven integrated UI frameworks. Delivered a project that fits five major tracks AppLovin, Reka, Groq, Chroma, and Y Combinator with clear value for each. Delivered a project that fits five major tracks AppLovin, Reka, Groq, Chroma, and Y Combinator with clear value for each.
### What we learned
Orchestrating asynchronous multimodal AI systems at scale requires strict dependency management. Orchestrating asynchronous multimodal AI systems at scale requires strict dependency management. Reka QuickTag’s marketing insight layer can predict engagement potential from creative tone alone. Reka QuickTag’s marketing insight layer can predict engagement potential from creative tone alone. Groq Whisper’s low-latency transcription accelerates real-time analysis. Groq Whisper’s low-latency transcription accelerates real-time analysis. ChromaDB embeddings are powerful for clustering and creative discovery. ChromaDB embeddings are powerful for clustering and creative discovery. True ad intelligence lies not in metrics, but in interpretation, understanding emotional and narrative structure. True ad intelligence lies not in metrics, but in interpretation, understanding emotional and narrative structure.
### What's next
We are expanding AdAtlas AI into a full-scale creative intelligence SaaS platform designed for marketing teams, ad networks, and recommendation engines. Planned Developments Public API for creative benchmarking and real-time ad insight generation Public API for creative benchmarking and real-time ad insight generation Integration with Bright Data for live web trend enrichment Integration with Bright Data for live web trend enrichment Elastic-based creative search engine for large-scale ad clustering Elastic-based creative search engine for large-scale ad clustering Correlation studies between tone, narrative pacing, and engagement metrics Correlation studies between tone, narrative pacing, and engagement metrics Y Combinator incubation track: evolving AdAtlas into a scalable startup offering insight-driven creative analytics Y Combinator incubation track: evolving AdAtlas into a scalable startup offering insight-driven creative analytics Our long-term goal is to make creativity measurable transforming subjective design into interpretable, actionable data.
## README (from the GitHub repository)
# AdAtlas AI — Feature Extractor
*A FastAPI backend that turns video and image ad creatives into rich, structured features using a multi-model AI pipeline.*
## 📖 Overview
AdAtlas AI is the backend service (`adatlas-backend`) for analyzing advertising creatives. Given an uploaded video or image, it extracts keyframes and runs them through several AI models in parallel to produce a single, unified JSON describing the creative — vision features, an audio transcript, creative-performance signals, and vector embeddings.
The pipeline combines classic computer vision (OpenCV/FFmpeg for frame extraction, color-tone and scene-count estimation) with hosted LLMs: **Google Gemini** for vision analysis and captioning, **Groq Whisper** for audio transcription, **Reka** for creative scoring, and **OpenAI** for embedding generation (plus a vision fallback). Resulting embeddings can optionally be persisted to **ChromaDB** for similarity search. Every external model is opt-in via environment flags, so the service degrades gracefully when a provider is not configured.
## ✨ Features
- **File upload** — accepts videos (`.mp4`, `.mov`, `.mkv`, `.avi`, `.m4v`) and images (`.jpg`, `.jpeg`, `.png`, `.webp`), stored with a generated ID.
- **Unified multi-model analysis** (`/analyze`) — for videos, runs Gemini vision, Reka QuickTag, Groq audio transcription, and visual embedding generation concurrently, then merges everything into one response.
- **Image analysis** (`/analyze_image`) — Gemini-based vision analysis and visual embeddings for still images.
- **Frame & scene understanding** — FFmpeg keyframe extraction (with an OpenCV fallback), warm/cool color-tone detection via HSV, and scene-count estimation via histogram correlation.
- **Audio transcription** — extracts audio and transcribes it with Groq Whisper (`whisper-large-v3`).
- **Creative scoring** — Reka QuickTag features such as expected CTR, virality score, keywords, and mood/tone.
- **Vector embeddings** — audio and visual embeddings that can be stored in ChromaDB (ChromaDB Cloud client) when enabled.
- **Batch processing** (`/batch`) — analyze multiple uploaded files in parallel.
- **Chatbot indexing** (`/chatbot/index_video`) — index a video with Reka for downstream chatbot use.
- **Health & debug endpoints** — liveness/readiness probes plus debug routes for benchmarking and testing individual services.
## 🛠️ Tech Stack
| Area | Technology |
| --- | --- |
| Language | Python 3.11 |
| Web framework | FastAPI + Uvicorn (ASGI) |
| Config / validation | Pydantic v2, pydantic-settings |
| Media processing | ffmpeg-python, OpenCV (`opencv-python-headless`), Pillow |
| Vision & captioning | Google Gemini (`google-generativeai`) |
| Audio transcription | Groq (`whisper-large-v3`) |
| Creative scoring | Reka Vision Agent API |
| Embeddings / fallback | OpenAI |
| Vector store | ChromaDB (optional) |
| Numerics | NumPy, scikit-learn |
| HTTP clients | httpx, requests |
## 🚀 Getting Started
### Prerequisites
- Python 3.11
- [FFmpeg](https://ffmpeg.org/) installed and available on your `PATH`
- API keys for the providers you intend to enable (Groq, Gemini, Reka, OpenAI, ChromaDB)
### Installation
```bash
git clone https://github.com/2006-sk/backcal.git
cd backcal/adatlas-backend
python3.11 -m venv .venv
source .venv/bin/activate
pip install -r Requirements.txt
# ChromaDB is imported by the app but not pinned in Requirements.txt; install it too:
pip install chromadb
```
### Configuration
The app reads settings from a `.env` file in the `adatlas-backend` directory (see `app/core/config.py`). Every external service is **disabled by default** — enable each with its `USE_*` flag *and* provide the matching key:
```env
# Groq (audio transcription)
USE_GROQ=true
GROQ_API_KEY=your_groq_key
# Google Gemini (vision + captions)
USE_GEMINI=true
GEMINI_API_KEY=your_gemini_key
GEMINI_MODEL=gemini-2.5-flash # or gemini-2.5-pro
# Reka (creative scoring / QuickTag)
USE_REKA=true
REKA_API_KEY=your_reka_key
# OpenAI (embedding generation / vision fallback)
OPENAI_API_KEY=your_openai_key
# ChromaDB (optional embedding storage — ChromaDB Cloud)
USE_CHROMA=true
CHROMA_API_KEY=your_chroma_key
CHROMA_TENANT=your_tenant
CHROMA_DATABASE=your_database
CHROMA_COLLECTION_NAME=adatlas_embeddings
```
### Usage
From the `adatlas-backend` directory, start the API server:
```bash
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
```
Then upload a file and analyze it:
```bash
# 1) Upload
curl -F "file=@sample.mp4" http://localhost:8000/upload
# -> { "id": "", "file_name": "...", "saved_path": "..." }
# 2) Analyze
curl -X POST http://localhost:8000/analyze \
-H "Content-Type: application/json" \
-d '{"file_id": ""}'
```
Interactive API docs are available at `http://localhost:8000/docs`.
## 📁 Project Structure
```
backcal/
└── adatlas-backend/
├── app/
│ ├── main.py # FastAPI app + all route handlers
│ ├── core/config.py # Settings (env vars, model selection)
│ ├── models/schemas.py # Pydantic request/response models
│ ├── services/ # Gemini, Groq, Reka, OpenAI, Chroma,
│ │ # audio & visual embedding helpers
│ └── utils/ # timing, unified output, Reka client
├── Requirements.txt
├── run.sh
├── API_ROUTES.md # Endpoint reference
└── test_*.py # Standalone integration/model tests
```
## 📚 API Endpoints
| Method | Path | Description |
| --- | --- | --- |
| `GET` | `/`, `/health`, `/health/live`, `/health/ready` | Health / liveness / readiness checks |
| `POST` | `/upload` | Upload a video or image file |
| `POST` | `/analyze` | Full multi-model analysis (video or image) |
| `POST` | `/analyze_image` | Image-only Gemini vision analysis |
| `POST` | `/batch` | Analyze multiple uploaded files in parallel |
| `POST` | `/chatbot/index_video` | Index a video with Reka for chatbot use |
| `GET` | `/debug/*` | Demo, benchmark, and per-service test routes |
See [`adatlas-backend/API_ROUTES.md`](adatlas-backend/API_ROUTES.md) for full request/response examples.
## Detected evidence (automated analysis)
Indexed codebase: 27 recognized source files, 177 KB.
- FastAPI (technology) — detected in the code
- OpenAI (technology) — detected in the code
- Python (language) — detected in the code
- Google Gemini (technology) — claimed on Devpost, not found in the code
## Codebase structure (from repository index)
### Files (31 of 31)
```
.DS_Store
.gitignore
adatlas-backend/.gitignore
adatlas-backend/API_RESPONSE_FORMAT.md
adatlas-backend/API_ROUTES.md
adatlas-backend/app/core/__init__.py
adatlas-backend/app/core/config.py
adatlas-backend/app/main.py
adatlas-backend/app/models/__init__.py
adatlas-backend/app/models/schemas.py
adatlas-backend/app/services/__init__.py
adatlas-backend/app/services/audio_embeddings.py
adatlas-backend/app/services/chroma_service.py
adatlas-backend/app/services/gemini_helper.py
adatlas-backend/app/services/groq_helper.py
adatlas-backend/app/services/openai/openai_helper.py
adatlas-backend/app/services/reka_helper.py
adatlas-backend/app/services/storage.py
adatlas-backend/app/services/visual_embeddings.py
adatlas-backend/app/utils/__init__.py
adatlas-backend/app/utils/reka_client.py
adatlas-backend/app/utils/timing.py
adatlas-backend/app/utils/unified_output.py
adatlas-backend/MODEL_COMPARISON.md
adatlas-backend/postman_examples.md
adatlas-backend/Requirements.txt
adatlas-backend/run.sh
adatlas-backend/test_all_models.py
adatlas-backend/test_gemini_embeddings.py
adatlas-backend/test_gemini_pro.py
adatlas-backend/test_integration.py
```
### Dependencies
- adatlas-backend/Requirements.txt: fastapi@==0.115.2, ffmpeg-python@==0.2.0, google-generativeai@==0.8.3, groq@==0.33.0, httpx@==0.27.0, numpy@==1.24.3, openai@==1.51.0, opencv-python-headless@==4.12.0.88, Pillow@==10.0.0, pydantic@==2.9.2, pydantic-settings@==2.5.2, python-multipart@==0.0.20, requests@==2.31.0, scikit-learn@==1.3.0, uvicorn[standard]@==0.30.6
### Recent commits (newest first)
- Reka
- Reka
- Reka
- Reka
- Initial clean commit (history reset)
## Key source files (fetched from GitHub, selected and truncated for size)
### adatlas-backend/MODEL_COMPARISON.md
```markdown
# Gemini Model Comparison
## Speed vs Accuracy Trade-off
### Gemini 2.5 Flash (FAST)
- **Speed**: ~2-5 seconds per request
- **First token latency**: ~1-2 seconds
- **Accuracy**: Good (found 3/5 faces)
- **Cost**: Free tier
- **Best for**: Fast responses, high volume
### Gemini 2.5 Pro (ACCURATE)
- **Speed**: ~20-40 seconds per request
- **First token latency**: ~36 seconds
- **Accuracy**: Excellent (should find 5/5 faces)
- **Cost**: Still free tier
- **Best for**: Maximum accuracy, detailed analysis
## Recommendation
Since you're analyzing **ALL frames** from videos:
- Each video has ~30-60 frames
- Pro model will be VERY slow (could take 10-30 minutes per video)
- Flash is fast (30 seconds - 2 minutes per video)
## Options
1. **Keep Pro**: Better accuracy but slower
2. **Use Flash**: Fast but less accurate
3. **Hybrid**: Use Pro for key frames only, Flash for others
```
### adatlas-backend/API_RESPONSE_FORMAT.md
```markdown
# AdAtlas API Response Format
## POST /analyze
### Request Body (JSON)
```json
{
"file_id": "abc123-def456-ghi789",
"file_path": null
}
```
or
```json
{
"file_id": null,
"file_path": "/path/to/video.mp4"
}
```
### Response Format
```json
{
"gemini": {
"file_name": "video.mp4",
"duration": 15.5,
"scene_count": 3,
"color_tone": "warm",
"vision_features": {
"faces_detected": 1,
"objects": ["Amazon package", "knife block set", "vegetables"],
"text_present": true,
"scene_description": "Video shows unboxing and food preparation",
"mood": "creative, promotional",
"color_scheme": "warm tones with vibrant vegetables",
"activity": "unboxing → food prep → artistic plating",
"motion_style": "smooth transitions",
"cta_present": false,
"notes": "Educational cooking content"
},
"video_summary": "A creative video showing unboxing → food prep → artistic plating.",
"error": null
},
"reka_features": {
"expected_ctr": 1.5,
"virality_score": 65,
"keywords": ["stress", "wellness", "brand products", "female individuals"],
"mood_tone": ["happy", "relaxing", "relief", "relaxed"],
"raw_data": { /* Reka API response */ },
"method": "quicktag",
"error": null
},
"audio_embedding": {
"transcript": "Welcome to this cooking tutorial...",
"embedding_dimensions": 768,
"error": null
},
"visual_embedding": {
"captions": [
"A woman unboxes an Amazon package",
"Kitchen counter with colorful vegetables",
"Artistic food plating on wooden table"
],
"embedding_dimensions": 768,
"frame_count": 12,
"error": null
},
"unified": {
/* Combines all above fields */
"file_name": "video.mp4",
"duration": 15.5,
"scene_count": 3,
"color_tone": "warm",
"vision_features": { /* ... */ },
"video_summary": "A creative video...",
"reka": { /* Reka features */ },
"audio": { /* Audio results */ },
"visual": { /* Visual results */ }
}
}
```
### Fields Description
#### `gemini` (Gemini vision analysis)
- `file_name`: Name of the analyzed file
- `duration`: Video duration in seconds
- `scene_count`: Estimated number of scenes
- `color_tone`: "warm" or "cool"
- `vision_features`: Detailed vision analysis
- `faces_detected`: Number of faces
- `objects`: List of detected objects
- `text_present`: Boolean
- `scene_description`: Overall scene description
- `mood`: Detected mood
- `color_scheme`: Color analysis
- `activity`: Sequence of activities
- `motion_style`: Motion characteristics
- `cta_present`: Call-to-action detected
- `notes`: Additional insights
- `video_summary`: High-level summary
- `error`: Error message if failed
#### `reka_features` (Reka QuickTag)
- `expected_ctr`: Expected click-through rate
- `virality_score`: Virality prediction score
- `keywords`: Extracted keywords
- `mood_tone`: Detected moods
- `raw_data`: Complete Reka response
[truncated — 483 more characters]
```
### adatlas-backend/Requirements.txt
```
fastapi==0.115.2
uvicorn[standard]==0.30.6
pydantic==2.9.2
pydantic-settings==2.5.2
ffmpeg-python==0.2.0
python-multipart==0.0.20
Pillow==10.0.0
opencv-python-headless==4.12.0.88
groq==0.33.0
openai==1.51.0
requests==2.31.0
httpx==0.27.0
numpy==1.24.3
scikit-learn==1.3.0
google-generativeai==0.8.3
```
### adatlas-backend/app/main.py
```python
from fastapi import FastAPI, UploadFile, File, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
from pathlib import Path
from typing import Optional
import asyncio
import cv2
import numpy as np
import base64
import json
from app.core.config import settings
from app.models.schemas import AnalyzeResult, UploadResponse, BatchAnalyzeRequest, AnalyzeRequest
from app.services.storage import Storage
from app.services.groq_helper import GroqHelper
from app.services.gemini_helper import GeminiHelper
from app.services.reka_helper import RekaHelper
from app.utils.timing import stopwatch
from app.utils.unified_output import build_unified_json
from app.services.chroma_service import chroma_service
# OpenAI fallback (moved to backup folder)
try:
from app.services.openai.openai_helper import OpenAIHelper as OpenAIHelperFallback
except ImportError:
OpenAIHelperFallback = None
app = FastAPI(title=settings.APP_NAME, version=settings.APP_VERSION)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], allow_credentials=True,
allow_methods=["*"], allow_headers=["*"],
)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
"""Custom validation error handler to prevent UTF-8 decode errors."""
errors = []
for error in exc.errors():
errors.append({
"loc": str(error.get("loc", [])),
"msg": str(error.get("msg", "")),
"type": str(error.get("type", ""))
})
return JSONResponse(
status_code=422,
content={"detail": errors}
)
# Initialize services
storage = Storage()
groq_helper = GroqHelper()
gemini_helper = GeminiHelper()
reka_helper = RekaHelper()
# Fallback to OpenAI if Gemini not available
if not gemini_helper.enabled() and OpenAIHelperFallback:
openai_helper = OpenAIHelperFallback()
print("[Warning] Using OpenAI fallback - configure Gemini for better experience")
else:
openai_helper = None
# File extensions
VIDEO_EXT = {".mp4", ".mov", ".mkv", ".avi", ".m4v"}
IMAGE_EXT = {".jpg", ".jpeg", ".png", ".webp"}
# Health endpoints
@app.get("/")
def root_health():
return {"status": "ok"}
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/health/live")
def liveness():
return {"live": True}
@app.get("/health/ready")
def readiness():
return {"ready": True}
# Core API endpoints
@app.post("/upload", response_model=UploadResponse)
async def upload(file: UploadFile = File(...)):
"""Upload a video or image file for analysis."""
if not file.filename:
raise HTTPException(400, "Missing filename")
fid, path = storage.save_upload(file.file, file.filename)
return UploadResponse(id=fid, file_name=Path(path).name, saved_path=str(path))
def sanitize_response(data):
"""Convert numpy arrays and other non-serializable types to JSON-safe formats."""
if isinstance(data, dict):
return {k: sanitize_response(v) for k, v in data.items()}
elif isinstance(data, list):
return [sanitize_response(item) for item in data]
elif isinstance(data, np.ndarray):
return data.tolist()
elif isinstance(data, (np.integer, np.floating)):
return float(data)
elif isinstance(data, bytes):
return data.decode('utf-8', errors='ignore')
elif isinstance(data, (int, float, str, bool, type(None))):
return data
else:
return str(data)
@app.post("/analyze")
async def analyze(request: AnalyzeRequest):
"""
Analyze a video or image file using Gemini + Groq + Reka.
Request body should contain:
- file_id: ID from /upload endpoint (optional)
- file_path: Direct path to file (optional)
Example JSON:
{
"file_id": "abc123"
}
or
{
"file_path": "/path/to/video.mp4"
}
"""
# Resolve file path
path = None
if request.file_id:
path = storage.get_path(request.file_id)
elif request.file_path:
p = Path(request.file_path)
path = p if p.exists() else None
if not path:
raise HTTPException(404, "File not found. Provide file_id from /upload or a valid file_path.")
ext = Path(path).suffix.lower()
with stopwatch("analyze_total"):
if ext in VIDEO_EXT:
# Video analysis with PARALLEL Gemini + Reka processing
try:
print(f"[Parallel] Starting analysis for {Path(path).name}")
# Step 1: Extract keyframes (parallel with all other operations)
print(f"[Parallel] Starting ALL operations simultaneously...")
# Create ALL tasks immediately
tasks = []
# Frame extraction task
tasks.append(
asyncio.get_event_loop().run_in_executor(
None,
gemini_helper.extract_keyframes,
path,
0.5
)
)
# Reka task (fast QuickTag, no frame extraction needed)
if settings.USE_REKA:
tasks.append(reka_helper.get_reka_features(str(path)))
else:
tasks.append(asyncio.sleep(0))
# Audio embeddings task
tasks.append(
asyncio.get_event_loop().run_in_executor(
None,
groq_helper.analyze_audio_with_groq,
path
)
)
# Wait for frame extraction first (needed for other tasks)
frame_data_task = tasks[0]
[truncated — 25765 more characters]
```
### adatlas-backend/run.sh
```shell
#!/bin/bash
# Run script for AdAtlas Backend
# Activate virtual environment
source ../venv/bin/activate
# Run the application with uvicorn
cd adatlas-backend
python -m uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
```
### adatlas-backend/test_gemini_pro.py
```python
#!/usr/bin/env python3
"""Test Gemini Pro model."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from app.core.config import settings
import google.generativeai as genai
def test_gemini_pro():
print("Testing Gemini 2.5 Pro model...")
if not settings.GEMINI_API_KEY:
print("ERROR: GEMINI_API_KEY not configured")
return False
try:
genai.configure(api_key=settings.GEMINI_API_KEY)
model = genai.GenerativeModel('gemini-2.5-pro')
# Test text generation
print("\n1. Testing text generation...")
response = model.generate_content("Say hello in one word")
print(f" Response: {response.text}")
print("\nSUCCESS: Gemini 2.5 Pro is working!")
return True
except Exception as e:
print(f"\nERROR: {e}")
return False
if __name__ == "__main__":
success = test_gemini_pro()
sys.exit(0 if success else 1)
```
### adatlas-backend/test_all_models.py
```python
#!/usr/bin/env python3
"""Test speed of different Gemini models."""
import sys
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from app.core.config import settings
import google.generativeai as genai
def test_model_speed(model_name):
"""Test a model's speed."""
print(f"\nTesting {model_name}...")
try:
genai.configure(api_key=settings.GEMINI_API_KEY)
model = genai.GenerativeModel(model_name)
start = time.time()
response = model.generate_content("Say hello in one word")
elapsed = time.time() - start
print(f" Response: {response.text}")
print(f" Time: {elapsed:.2f}s")
return elapsed
except Exception as e:
print(f" ERROR: {e}")
return None
if __name__ == "__main__":
models = [
'gemini-2.5-flash',
'gemini-2.0-flash',
'gemini-2.5-pro',
'gemini-flash-latest'
]
results = {}
for model in models:
elapsed = test_model_speed(model)
if elapsed:
results[model] = elapsed
print("\n=== Speed Summary ===")
for model, time in sorted(results.items(), key=lambda x: x[1]):
print(f"{model}: {time:.2f}s")
```
### adatlas-backend/test_gemini_embeddings.py
```python
#!/usr/bin/env python3
"""Test Gemini embeddings API."""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from app.core.config import settings
import google.generativeai as genai
def test_gemini_embeddings():
print("Testing Gemini Embeddings API...")
print(f"USE_GEMINI: {settings.USE_GEMINI}")
print(f"API Key: {settings.GEMINI_API_KEY[:20]}..." if settings.GEMINI_API_KEY else "API Key: Not configured")
if not settings.USE_GEMINI or not settings.GEMINI_API_KEY:
print("ERROR: Gemini not configured")
return False
try:
# Configure Gemini
genai.configure(api_key=settings.GEMINI_API_KEY)
# Test embedding generation
print("\n1. Testing text embedding...")
test_text = "A video showing a person walking in a park"
result = genai.embed_content(
model='models/text-embedding-004',
content=test_text,
task_type='retrieval_document'
)
embedding = result['embedding']
print(f" SUCCESS: Generated embedding")
print(f" Dimensions: {len(embedding)}")
print(f" First 5 values: {embedding[:5]}")
# Test multiple embeddings
print("\n2. Testing multiple embeddings...")
texts = [
"A video showing a person walking",
"A video showing a person running",
"A video showing a person sitting"
]
embeddings = []
for i, text in enumerate(texts):
result = genai.embed_content(
model='models/text-embedding-004',
content=text,
task_type='retrieval_document'
)
embeddings.append(result['embedding'])
print(f" Generated embedding {i+1}/{len(texts)}")
# Test averaging
import numpy as np
avg_embedding = np.mean(embeddings, axis=0).tolist()
print(f" Averaged embedding dimensions: {len(avg_embedding)}")
print("\nSUCCESS: Gemini embeddings are working!")
return True
except Exception as e:
print(f"\nERROR: Gemini embeddings failed: {e}")
import traceback
traceback.print_exc()
return False
if __name__ == "__main__":
success = test_gemini_embeddings()
sys.exit(0 if success else 1)
```
### adatlas-backend/test_integration.py
```python
#!/usr/bin/env python3
"""
Integration tests for AdAtlas AI FastAPI backend.
Tests ffmpeg + OpenCV + Groq logic.
"""
import os
import sys
import tempfile
from pathlib import Path
import cv2
import numpy as np
import ffmpeg
# Add app to path
sys.path.insert(0, str(Path(__file__).parent / "app"))
from services.groq_helper import GroqHelper
from core.config import settings
def test_ffmpeg_installation():
"""Test 1: Verify ffmpeg is installed and working."""
print("🧪 Test 1: FFmpeg Installation")
try:
result = ffmpeg.probe('pipe:', format='lavfi', f='testsrc=duration=1:size=320x240:rate=1')
print("✅ FFmpeg is working correctly")
return True
except Exception as e:
print(f"❌ FFmpeg test failed: {e}")
return False
def test_opencv_functions():
"""Test 2: Verify OpenCV color tone and scene count functions."""
print("\n🧪 Test 2: OpenCV Functions")
groq_helper = GroqHelper()
# Create test frames
frames = []
# Frame 1: Warm colors (red/orange)
warm_frame = np.zeros((100, 100, 3), dtype=np.uint8)
warm_frame[:, :, 0] = 10 # Low hue (red)
frames.append(warm_frame)
# Frame 2: Cool colors (blue/green)
cool_frame = np.zeros((100, 100, 3), dtype=np.uint8)
cool_frame[:, :, 0] = 120 # High hue (blue)
frames.append(cool_frame)
# Frame 3: Different scene (different histogram)
different_frame = np.ones((100, 100, 3), dtype=np.uint8) * 50
frames.append(different_frame)
try:
# Test color tone
color_tone = groq_helper.average_color_tone(frames)
print(f"✅ Color tone analysis: {color_tone}")
# Test scene count
scene_count = groq_helper.scene_count_estimate(frames)
print(f"✅ Scene count analysis: {scene_count}")
return True
except Exception as e:
print(f"❌ OpenCV test failed: {e}")
return False
def test_groq_mock():
"""Test 3: Mock Groq API call with dummy key."""
print("\n🧪 Test 3: Groq Mock Analysis")
# Set dummy API key
os.environ['GROQ_API_KEY'] = 'dummy_key_for_testing'
os.environ['USE_GROQ'] = 'true'
groq_helper = GroqHelper()
try:
# Test enabled check
enabled = groq_helper.enabled()
print(f"✅ Groq enabled check: {enabled}")
# Test graceful handling when Groq is not properly configured
result = groq_helper.analyze_video_with_groq(Path("nonexistent.mp4"))
print(f"✅ Graceful error handling: {result.get('error', 'No error')}")
return True
except Exception as e:
print(f"❌ Groq mock test failed: {e}")
return False
def test_full_pipeline():
"""Test 4: Full analyze pipeline with sample video."""
print("\n🧪 Test 4: Full Pipeline Test")
# Create a simple test video using ffmpeg
try:
test_video_path = Path("data/sample.mp4")
test_video_path.parent.mkdir(exist_ok=True)
# Create a 3-second test video
(
ffmpeg
.input('testsrc=duration=3:size=320x240:rate=1', f='lavfi')
.output(str(test_video_path), vcodec='libx264', pix_fmt='yuv420p')
.overwrite_output()
.run(quiet=True)
)
print(f"✅ Created test video: {test_video_path}")
# Test analysis
groq_helper = GroqHelper()
result = groq_helper.analyze_video_with_groq(test_video_path)
print(f"✅ Analysis result:")
print(f" Duration: {result.get('duration', 'N/A')}s")
print(f" FPS: {result.get('fps', 'N/A')}")
print(f" Total frames: {result.get('total_frames', 'N/A')}")
print(f" Color tone: {result.get('color_tone', 'N/A')}")
print(f" Scene count: {result.get('scene_count', 'N/A')}")
# Clean up
test_video_path.unlink()
print("✅ Cleaned up test video")
return True
except Exception as e:
print(f"❌ Full pipeline test failed: {e}")
return False
def main():
"""Run all integration tests."""
print("🚀 AdAtlas AI Integration Tests")
print("=" * 50)
tests = [
test_ffmpeg_installation,
test_opencv_functions,
test_groq_mock,
test_full_pipeline
]
passed = 0
total = len(tests)
for test in tests:
if test():
passed += 1
print(f"\n📊 Results: {passed}/{total} tests passed")
if passed == total:
print("🎉 All tests passed! The backend is ready.")
else:
print("⚠️ Some tests failed. Check the output above.")
return passed == total
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)
```
### adatlas-backend/app/core/__init__.py
```python
# Core package
```
[18 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]