## What This Does
Converts video footage into depth map image sequences for use in TouchDesigner, After Effects, or other 3D software. Each frame becomes a grayscale depth map where brightness = distance from camera.
## Prerequisites Installation
Open Terminal and run:
### 1. Install Homebrew (if not already installed)
/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"### 2. Install Python dependencies
# Install OpenCV and other required packagespython3 -m pip install opencv-python requests flask --user### 3. Install ngrok
# Install ngrok via Homebrew
brew install ngrok# Create free ngrok account at https://ngrok.com/
# Get your authtoken from the dashboard, then run:
ngrok config add-authtoken YOUR_TOKEN“`
### 4. Get Artificial Studio API Key
1. Go to https://artificialstudio.ai/
2. Create a free account
3. Navigate to your API dashboard
4. Generate an API key
5. Copy it for use in the script
## Running the Script
### Step 1: Save the script
1. Download `depth_processor.py`
2. Place it in a folder with your video files
### Step 2: Edit the configuration
Open `depth_processor.py` and edit the configuration section (around line 280):
# Configuration
API_KEY = “YOUR_API_KEY_HERE” # ⬅️ Paste your Artificial Studio API key
VIDEO_FILE = “cranes_clips/LIGHT SONG_001.mp4” # ⬅️ Your video path
BATCH_DELAY = 20.0 # Seconds between API calls (increase if hitting rate limits)
MAX_FRAMES = None # None = full video, or set a number for testing
### Step 3: Run the script
cd /path/to/your/folderpython3 depth_processor.py“`
## Folder Structure
The script automatically creates organized folders based on your video filename:
your_video_folder/
├── LIGHT SONG_001.mp4 (original video)
└── LIGHT SONG_001/ (auto-created)
├── frames/ (extracted video frames - temporary)
└── depth_maps/ (depth map PNGs - this is what you need!)“`
Each video gets its own folder – no manual organization needed!
## What Happens During Processing
1. **Webhook server starts** (Flask on port 5000)
2. **Ngrok tunnel created** (exposes webhook publicly)
3. **Frames extracted** from your video (saved as PNGs)
4. **Frames submitted** to Artificial Studio API (with delay between each)
5. **Webhooks received** as API completes processing
6. **Depth maps downloaded** automatically to the depth_maps folder
## Output Files
– `depth_maps/` folder contains numbered PNG files: `depth_00000.png`, `depth_00001.png`, etc.
– These are grayscale images where brightness = depth
– Import this image sequence directly into After Effects or TouchDesigner
## Troubleshooting
### “ngrok: command not found”
brew install ngrokngrok config add-authtoken YOUR_TOKEN“`
### “No module named ‘cv2′” or “No module named ‘flask'”
python3 -m pip install opencv-python requests flask --user“`
### “Permission denied” errors during pip install
Add `–user` flag:
python3 -m pip install opencv-python requests flask --user“`
### Rate limiting / “exceeded the limit of requests per minute”
This is normal – the script handles retries automatically. If you see many consecutive failures:
– Increase `BATCH_DELAY` to 30.0 or higher in the script
– Process videos during off-peak hours
### Script hangs with “Queue timeout” messages
The API stopped sending results back. This can happen with:
– Very long videos (>100 frames)
– API rate limits or service issues
– Network interruptions
**Solution**: Press `Ctrl+C` to stop. Check how many depth maps you got:
ls depth_maps/ | wc -l
If you got partial results, you can still use them or reprocess with longer delays.
### No depth maps downloaded (webhook receives but no files)
Check the “WEBHOOK DEBUG” output. If you see webhooks arriving but no “output” URLs, the API is acknowledging jobs but not completing them. Increase delays significantly.
## Monitoring Progress
You’ll see output like:
“`
🚀 Starting ngrok tunnel…
✅ Ngrok tunnel: https://xxxx.ngrok.io
✅ Webhook server running on port 5000
📁 AUTO-GENERATED FOLDER STRUCTURE:
Frames: cranes_clips/LIGHT SONG_001/frames
Depth maps: cranes_clips/LIGHT SONG_001/depth_maps
🎬 Video: 30.0 fps, extracting 150 frames
✅ Extracted 150 frames
✓ Submitted frame 0 (Job: abc123)
✓ Submitted frame 1 (Job: def456)
…
📥 Waiting for 150 depth maps…
📥 WEBHOOK DEBUG:
Job ID: xyz789
Output URL: https://files.artificialstudio.ai/…
⬇️ Downloading depth map for frame 42…
✅ Downloaded depth map 1/150 (frame 42)
“`
## Processing Time Estimates
– **20-second delay** between submissions (recommended for reliability)
– **30-60 seconds** API processing per frame
– **Example**: 100-frame video = ~1.5 hours total
Long videos require patience. This is a set-it-and-forget-it operation – start it before bed or during work.
## Tips for Best Results
1. **Test first**: Set `MAX_FRAMES = 20` to process just the first 20 frames as a test
2. **Don’t close Terminal**: Keep it open the entire time
3. **Stable internet**: Processing can take hours – use reliable WiFi
4. **Start with shorter clips**: 5-10 second videos for initial testing
5. **Check depth map quality**: Open a few PNGs to verify they look reasonable before processing more videos
## Using with TouchDesigner
Import the depth map PNG sequence:
1. Add a **Movie File In TOP**
2. Point it to your `depth_maps/` folder
3. Enable **Image Sequence** mode
4. Set the frame rate to match your original video
5. The grayscale values now represent depth data
Create a point cloud:
– Use **TOP to CHOP** to read pixel values
– Feed into **CHOP to SOP**
– Use depth values for Z-position
– Result: 3D point cloud that animates with the performance
## Using with After Effects
1. Import > File > Import as **Image Sequence**
2. Select the first file in `depth_maps/`
3. Check “Force alphabetical order” and “PNG Sequence”
4. The sequence imports as a single video layer
5. Use as a displacement map or luma matte
## Processing Multiple Videos
Edit the `VIDEO_FILE` line for each video:
“`python
VIDEO_FILE = “cranes_clips/LIGHT SONG_001.mp4” # First video
# When done, change to:
VIDEO_FILE = “cranes_clips/LIGHT SONG_002.mp4” # Second video
“`
Each gets its own auto-generated folder – no overwrites!
## Advanced: Batch Processing
If you want to queue multiple videos (warning: this takes many hours):
“`python
videos = [
“cranes_clips/LIGHT SONG_001.mp4”,
“cranes_clips/LIGHT SONG_002.mp4”,
“cranes_clips/LIGHT SONG_003.mp4”
]
for video in videos:
print(f”\n\n{‘=’*60}”)
print(f”STARTING: {video}”)
print(‘=’*60)
processor.process_video(video, batch_delay=20.0)
“`
## Important Notes
– **API costs**: Check Artificial Studio’s pricing – free tier has limits
– **Rate limits**: The 20-second delay is conservative; you might be able to go faster
– **Partial results**: If the script times out, you can still use whatever depth maps downloaded
– **Brightness adjustment**: Depth maps might look overexposed – adjust in TouchDesigner using Level TOP
– **Remove credentials**: DELETE your API key and ngrok token before sharing the script!
## Technical Background
This workflow uses:
– **Artificial Studio’s ML depth estimation** (trained neural network)
– **Asynchronous webhook architecture** (non-blocking API calls)
– **Ngrok tunneling** (expose localhost to the internet)
– **Flask web server** (receive webhook POST requests)
– **OpenCV** (video frame extraction and manipulation)
The depth estimation model infers 3D structure from 2D images using learned patterns about object scale, perspective, occlusion, and texture. It’s not perfect but remarkably effective for monocular depth estimation.
## Acknowledgments
This script was developed through collaboration with Claude (Anthropic’s AI assistant) to solve the specific challenge of batch-processing video through an async webhook-based API on a local machine.
The workflow combines multiple technologies that don’t typically work together out-of-the-box, making it a good example of creative technical problem-solving for artistic applications.
import cv2
import requests
import os
import time
import base64
import threading
from pathlib import Path
from flask import Flask, request, jsonify
from queue import Queue
import subprocess
import signal
import sys
import json
class WebhookDepthProcessor:
def __init__(self, api_key):
self.api_key = api_key
self.api_url = "https://api.artificialstudio.ai/api/generate"
self.results_queue = Queue()
self.pending_jobs = {}
self.flask_app = Flask(__name__)
self.ngrok_url = None
self.ngrok_process = None
# Setup Flask route
@self.flask_app.route('/webhook', methods=['POST'])
def webhook():
data = request.json
job_id = data.get('id', 'unknown')
output_url = data.get('output', None)
# DEBUG: Show what we received
print(f"\n📥 WEBHOOK DEBUG:")
print(f" Job ID: {job_id}")
print(f" Output URL: {output_url}")
print(f" Full data: {json.dumps(data, indent=2)}")
self.results_queue.put(data)
return jsonify({"status": "received"}), 200
def start_ngrok(self, port=5000):
"""Start ngrok tunnel"""
print("🚀 Starting ngrok tunnel...")
try:
# Start ngrok
self.ngrok_process = subprocess.Popen(
['ngrok', 'http', str(port)],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
# Wait for ngrok to start
time.sleep(3)
# Get ngrok URL from API
response = requests.get('http://127.0.0.1:4040/api/tunnels')
tunnels = response.json()['tunnels']
for tunnel in tunnels:
if tunnel['proto'] == 'https':
self.ngrok_url = tunnel['public_url']
print(f"✅ Ngrok tunnel: {self.ngrok_url}")
return self.ngrok_url
raise Exception("No HTTPS tunnel found")
except Exception as e:
print(f"❌ Ngrok error: {e}")
print("Make sure ngrok is installed: brew install ngrok")
return None
def start_webhook_server(self, port=5000):
"""Start Flask server in background thread"""
def run_server():
self.flask_app.run(port=port, debug=False, use_reloader=False)
thread = threading.Thread(target=run_server, daemon=True)
thread.start()
time.sleep(2)
print(f"✅ Webhook server running on port {port}")
def extract_frames(self, video_path, output_dir, max_frames=None):
"""Extract frames from video"""
os.makedirs(output_dir, exist_ok=True)
cap = cv2.VideoCapture(video_path)
fps = cap.get(cv2.CAP_PROP_FPS)
frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
if max_frames:
frame_count = min(frame_count, max_frames)
print(f"🎬 Video: {fps} fps, extracting first {frame_count} frames (test mode)")
else:
print(f"🎬 Video: {fps} fps, {frame_count} frames")
frames = []
idx = 0
while cap.isOpened() and (max_frames is None or idx < max_frames):
ret, frame = cap.read()
if not ret:
break
frame_path = os.path.join(output_dir, f"frame_{idx:05d}.png")
cv2.imwrite(frame_path, frame)
frames.append(frame_path)
if idx % 10 == 0:
print(f" Extracted frame {idx}/{frame_count}")
idx += 1
cap.release()
print(f"✅ Extracted {len(frames)} frames")
return frames, fps
def submit_frame(self, image_path, frame_index):
"""Submit single frame to API"""
webhook_url = f"{self.ngrok_url}/webhook"
# Read and encode image
with open(image_path, 'rb') as f:
image_bytes = f.read()
image_base64 = base64.b64encode(image_bytes).decode('utf-8')
image_data_url = f"data:image/png;base64,{image_base64}"
payload = {
"model": "image-depth-map-generator",
"input": {
"image": image_data_url
},
"webhook": webhook_url
}
headers = {
'Content-Type': 'application/json',
'Authorization': self.api_key
}
try:
response = requests.post(
self.api_url,
json=payload,
headers=headers,
timeout=30
)
if response.status_code == 200:
result = response.json()
job_id = result.get('id')
self.pending_jobs[job_id] = {
'frame_index': frame_index,
'frame_path': image_path,
'submitted_at': time.time()
}
print(f"✓ Submitted frame {frame_index} (Job: {job_id})")
return job_id
elif response.status_code == 429:
print(f"⚠️ Rate limited on frame {frame_index} - will retry")
return None
else:
print(f"❌ Error submitting frame {frame_index}: {response.status_code} - {response.text}")
return None
except Exception as e:
print(f"❌ Exception submitting frame {frame_index}: {e}")
return None
def download_depth_map(self, url, output_path):
"""Download depth map from result URL"""
try:
response = requests.get(url, timeout=30)
if response.status_code == 200:
with open(output_path, 'wb') as f:
f.write(response.content)
return True
return False
except Exception as e:
print(f"❌ Download error: {e}")
return False
def process_results(self, expected_count, output_dir="depth_maps", timeout_per_frame=600):
"""Process webhook results as they arrive"""
os.makedirs(output_dir, exist_ok=True)
received = 0
start_time = time.time()
print(f"\n📥 Waiting for {expected_count} depth maps...")
print(f" (Timeout: {timeout_per_frame}s per frame)")
while received < expected_count:
elapsed = time.time() - start_time
total_timeout = timeout_per_frame * expected_count
if elapsed > total_timeout:
print(f"\n⚠️ Timeout reached ({elapsed:.0f}s). Stopping collection.")
break
try:
# Wait for result
result = self.results_queue.get(timeout=60)
job_id = result.get('id')
output_url = result.get('output')
if not output_url:
print(f"⚠️ Job {job_id} completed but no output URL provided")
print(f" Status: {result.get('status', 'unknown')}")
continue
if job_id in self.pending_jobs:
job_info = self.pending_jobs[job_id]
frame_index = job_info['frame_index']
# Download depth map
output_path = os.path.join(
output_dir,
f"depth_{frame_index:05d}.png"
)
print(f"\n⬇️ Downloading depth map for frame {frame_index}...")
if self.download_depth_map(output_url, output_path):
received += 1
processing_time = time.time() - job_info['submitted_at']
print(f"✅ Downloaded depth map {received}/{expected_count} (frame {frame_index})")
print(f" Processing time: {processing_time:.1f}s")
else:
print(f"❌ Failed to download depth map for frame {frame_index}")
del self.pending_jobs[job_id]
except Exception as e:
if "Empty" not in str(e):
print(f"⚠️ Queue timeout or error: {e}")
print(f"\n📊 Collection complete: {received}/{expected_count} depth maps received")
# Clean up ngrok
if self.ngrok_process:
self.ngrok_process.terminate()
print("\n" + "=" * 60)
if received > 0:
print(f"✅ COMPLETE!")
print(f" Depth maps saved to: {depth_dir}")
print(f" Frames processed: {received}/{len(frames)}")
print(f"\n💡 Import the depth map PNGs into After Effects or TouchDesigner")
else:
print("❌ No depth maps received - check API limits and webhook messages")
print("=" * 60)
return received
def process_video(self, video_path, batch_delay=20.0, max_frames=None):
"""Complete pipeline"""
# Auto-generate folder structure from video filename
video_name = Path(video_path).stem # e.g., "LIGHT SONG_003"
base_dir = Path(video_path).parent / video_name # e.g., "cranes_clips/LIGHT SONG_003"
frames_dir = str(base_dir / "frames")
depth_dir = str(base_dir / "depth_maps")
print("=" * 60)
print("DEPTH MAP VIDEO PROCESSOR")
if max_frames:
print(f"TEST MODE: Processing first {max_frames} frames only")
print("=" * 60)
print(f"\n📁 AUTO-GENERATED FOLDER STRUCTURE:")
print(f" Frames: {frames_dir}")
print(f" Depth maps: {depth_dir}")
print()
# Step 1: Start webhook infrastructure
self.start_webhook_server()
if not self.start_ngrok():
print("❌ Failed to start ngrok. Exiting.")
return None
# Step 2: Extract frames
print("\n[1/4] Extracting frames...")
frames, fps = self.extract_frames(video_path, frames_dir, max_frames=max_frames)
if not frames:
print("❌ No frames extracted")
return None
# Step 3: Submit all frames with rate limiting
print(f"\n[2/4] Submitting {len(frames)} frames to API...")
print(f" Delay between submissions: {batch_delay}s")
submitted = 0
for idx, frame_path in enumerate(frames):
job_id = self.submit_frame(frame_path, idx)
if job_id:
submitted += 1
else:
print(f"⚠️ Frame {idx} not submitted, waiting extra time...")
time.sleep(batch_delay * 2) # Wait longer if rate limited
# Retry once
job_id = self.submit_frame(frame_path, idx)
if job_id:
submitted += 1
if idx < len(frames) - 1:
time.sleep(batch_delay)
print(f"\n✅ Successfully submitted {submitted}/{len(frames)} frames")
# Step 4: Collect results
print(f"\n[3/4] Collecting depth maps...")
received = self.process_results(submitted, output_dir=depth_dir)
if received == 0:
print("\n❌ No depth maps received. Check:")
print(" 1. API rate limits")
print(" 2. Webhook messages above for errors")
print(" 3. Your API key is valid")
return None
if received < submitted:
print(f"⚠️ Only received {received}/{submitted} depth maps")
print(" You can still use the partial result in After Effects/TouchDesigner")
return received
def signal_handler(sig, frame):
"""Handle Ctrl+C gracefully"""
print('\n\n⚠️ Interrupted! Cleaning up...')
sys.exit(0)
if __name__ == "__main__":
signal.signal(signal.SIGINT, signal_handler)
# Configuration
API_KEY = "ac04e52c79e6aa96266f7781bf31a01a4d444357"
VIDEO_FILE = "cranes_clips/LIGHT SONG_003.mp4" # ⬅️ CHANGE THIS
BATCH_DELAY = 20.0 # Increased to 20s for better reliability
MAX_FRAMES = None # Set to a number for testing, None for full video
print("\n⚙️ CONFIGURATION:")
print(f" Video: {VIDEO_FILE}")
print(f" Delay: {BATCH_DELAY}s between frames (slower = more reliable)")
print(f" Mode: {'TEST (first ' + str(MAX_FRAMES) + ' frames)' if MAX_FRAMES else 'FULL VIDEO'}")
print(f"\n All output will be auto-organized in: {Path(VIDEO_FILE).parent / Path(VIDEO_FILE).stem}/")
print()
# Process video
processor = WebhookDepthProcessor(API_KEY)
processor.process_video(VIDEO_FILE, batch_delay=BATCH_DELAY, max_frames=MAX_FRAMES)
