Camera Integration
PraisonAI supports camera integration for real-time visual analysis through multimodal agents. While there’s no built-in camera capture, you can easily integrate camera feeds by capturing frames or videos and passing them to vision agents.
Overview
Camera integration works by:
- Capturing frames/videos from camera using OpenCV
- Saving temporarily to disk
- Passing file paths to agents via the
images
parameter
- Cleaning up temporary files after analysis
Quick Start
Prerequisites
pip install praisonaiagents opencv-python
export OPENAI_API_KEY=your_openai_api_key
Basic Camera Capture
import cv2
from praisonaiagents import Agent, Task, PraisonAIAgents
def capture_and_analyze():
# Create vision agent
vision_agent = Agent(
name="CameraAnalyst",
role="Camera Feed Analyzer",
goal="Analyze camera captures in real-time",
backstory="Expert in real-time visual analysis",
llm="gpt-4o-mini"
)
# Capture from camera
cap = cv2.VideoCapture(0) # 0 for default camera
ret, frame = cap.read()
if ret:
# Save frame temporarily
cv2.imwrite("temp_capture.jpg", frame)
cap.release()
# Create analysis task
task = Task(
description="Analyze what you see in this camera feed",
expected_output="Detailed analysis of camera content",
agent=vision_agent,
images=["temp_capture.jpg"]
)
# Run analysis
agents = PraisonAIAgents(
agents=[vision_agent],
tasks=[task]
)
return agents.start()
# Run analysis
result = capture_and_analyze()
Integration Patterns
1. Single Frame Analysis
Perfect for quick snapshots and one-time analysis:
def single_frame_analysis():
cap = cv2.VideoCapture(0)
ret, frame = cap.read()
if ret:
image_path = "snapshot.jpg"
cv2.imwrite(image_path, frame)
# Analyze with your agent
# ... (agent setup and task creation)
cap.release()
2. Continuous Monitoring
Ideal for security systems and real-time monitoring:
def continuous_monitoring(interval=10):
vision_agent = Agent(
name="SecurityMonitor",
role="Security Camera Analyst",
goal="Monitor for security events",
backstory="Expert security analyst",
llm="gpt-4o-mini"
)
cap = cv2.VideoCapture(0)
while True:
ret, frame = cap.read()
if ret:
timestamp = int(time.time())
filename = f"capture_{timestamp}.jpg"
cv2.imwrite(filename, frame)
# Analyze frame
task = Task(
description="Monitor for unusual activities",
agent=vision_agent,
images=[filename]
)
agents = PraisonAIAgents(
agents=[vision_agent],
tasks=[task]
)
result = agents.start()
print(f"Analysis: {result}")
time.sleep(interval)
3. Multi-Agent Analysis
Use multiple specialized agents for comprehensive analysis:
def multi_agent_camera_analysis():
# Create specialized agents
security_agent = Agent(
name="SecurityExpert",
role="Security Specialist",
goal="Identify security threats",
backstory="Expert in surveillance and threat detection",
llm="gpt-4o-mini"
)
object_detector = Agent(
name="ObjectDetector",
role="Object Recognition Specialist",
goal="Identify and catalog objects",
backstory="Computer vision expert",
llm="gpt-4o-mini"
)
scene_analyst = Agent(
name="SceneAnalyst",
role="Scene Understanding Expert",
goal="Provide comprehensive scene analysis",
backstory="Environmental analysis expert",
llm="gpt-4o-mini"
)
# Capture frame
cap = cv2.VideoCapture(0)
ret, frame = cap.read()
cap.release()
if ret:
cv2.imwrite("analysis_frame.jpg", frame)
# Create specialized tasks
security_task = Task(
description="Analyze for security threats",
agent=security_agent,
images=["analysis_frame.jpg"]
)
object_task = Task(
description="Identify all objects in scene",
agent=object_detector,
images=["analysis_frame.jpg"]
)
scene_task = Task(
description="Provide comprehensive scene analysis",
agent=scene_analyst,
images=["analysis_frame.jpg"]
)
# Run parallel analysis
agents = PraisonAIAgents(
agents=[security_agent, object_detector, scene_analyst],
tasks=[security_task, object_task, scene_task],
process="parallel"
)
return agents.start()
4. Video Recording & Analysis
Analyze temporal events and activities:
def record_and_analyze_video(duration=10):
cap = cv2.VideoCapture(0)
# Setup video writer
fps = int(cap.get(cv2.CAP_PROP_FPS)) or 30
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter('recording.mp4', fourcc, fps, (width, height))
# Record video
start_time = time.time()
while (time.time() - start_time) < duration:
ret, frame = cap.read()
if ret:
out.write(frame)
cap.release()
out.release()
# Analyze video
video_agent = Agent(
name="VideoAnalyst",
role="Video Content Analyzer",
goal="Analyze video content for activities",
backstory="Expert in video analysis",
llm="gpt-4o-mini"
)
task = Task(
description="Analyze this video for activities and events",
agent=video_agent,
images=["recording.mp4"] # Video files work too!
)
agents = PraisonAIAgents(
agents=[video_agent],
tasks=[task]
)
return agents.start()
PraisonAI accepts various visual input formats:
- ✅ Local Images:
"camera_shot.jpg"
, "webcam_capture.png"
- ✅ Local Videos:
"security_feed.mp4"
, "recording.avi"
- ✅ Image URLs:
"https://example.com/live_feed.jpg"
- ✅ Multiple Sources:
["cam1.jpg", "cam2.jpg", "video.mp4"]
Configuration Options
Camera Selection
# Different camera sources
camera_id = 0 # Default camera
camera_id = 1 # External USB camera
camera_id = "rtsp://camera-ip/stream" # IP camera
Analysis Intervals
# For continuous monitoring
analysis_interval = 5 # Every 5 seconds (high frequency)
analysis_interval = 30 # Every 30 seconds (moderate)
analysis_interval = 300 # Every 5 minutes (low frequency)
Video Recording
# Recording duration
duration = 10 # 10 seconds
duration = 60 # 1 minute
duration = 300 # 5 minutes
Use Cases
Security Monitoring
- Real-time threat detection
- Unauthorized access alerts
- Suspicious activity identification
- Perimeter monitoring
Retail Analytics
- Customer behavior analysis
- Inventory monitoring
- Queue management
- Theft prevention
Industrial Automation
- Quality control inspection
- Safety compliance monitoring
- Equipment status verification
- Process optimization
Smart Home
- Activity recognition
- Elderly care monitoring
- Pet monitoring
- Energy usage optimization
Best Practices
-
Frame Rate Management
# Reduce analysis frequency for better performance
time.sleep(5) # Analyze every 5 seconds instead of continuously
-
Image Size Optimization
# Resize frames for faster processing
frame = cv2.resize(frame, (640, 480))
-
Parallel Processing
# Use parallel process for multiple agents
process="parallel"
Memory Management
-
Clean Up Temporary Files
import os
if os.path.exists(image_path):
os.remove(image_path)
-
Limit Video Duration
# Keep recordings short to avoid large files
max_duration = 30 # seconds
Security Considerations
-
Camera Permissions
- Ensure application has camera access
- Handle permission errors gracefully
-
Privacy Protection
- Implement data retention policies
- Secure transmission of camera data
- User consent for recording
-
Access Control
- Authenticate camera access
- Implement role-based permissions
Troubleshooting
Camera Not Found
# Check available cameras
for i in range(4):
cap = cv2.VideoCapture(i)
if cap.isOpened():
print(f"Camera {i} is available")
cap.release()
else:
print(f"Camera {i} not available")
Permission Issues
Linux:
# Add user to video group
sudo usermod -a -G video $USER
macOS:
- Grant camera permissions in System Preferences > Security & Privacy
Windows:
- Check camera privacy settings in Windows Settings
# Reduce frame size
frame = cv2.resize(frame, (320, 240))
# Lower analysis frequency
time.sleep(10) # Analyze every 10 seconds
# Use threading for non-blocking capture
import threading
Examples
Complete working examples are available in the repository: