feat: add camera integration examples and documentation

- Add comprehensive camera integration examples in examples/python/camera/
  - camera-basic.py: Single frame capture and analysis
  - camera-continuous.py: Continuous monitoring system
  - camera-multi-agent.py: Multi-agent parallel analysis
  - camera-video-analysis.py: Video recording and temporal analysis
  - README.md: Complete documentation for all examples

- Add camera integration documentation in docs/features/camera-integration.mdx
  - Complete integration patterns and use cases
  - Configuration options and best practices
  - Troubleshooting guide and security considerations
  - Links to example code

- Update docs/mint.json to include camera integration in navigation

Addresses issue #238 regarding multimodel agent camera input

Co-authored-by: MervinPraison <[email protected]>
This commit is contained in:
claude[bot]andMervinPraison authored and GitHub committed 2025-05-28 16:22:02 +00:00
1 parent a2d7629492
commit 955a05be59
7 files changed
+1212 -1

No files matched your search

+173
View File
@@ -0,0 +1,173 @@
# Camera Integration Examples
This directory contains examples demonstrating how to integrate camera feeds with PraisonAI multimodal agents for real-time visual analysis.
## 📋 Overview
PraisonAI supports visual input through the `images` parameter in Tasks. While there's no built-in camera capture, you can integrate camera feeds by capturing frames/videos and passing them to vision agents.
## 🛠️ Setup
### Requirements
```bash
pip install praisonaiagents opencv-python
```
### Environment Variables
```bash
export OPENAI_API_KEY=your_openai_api_key
```
## 📁 Examples
### 1. `camera-basic.py` - Basic Camera Capture
- **Purpose**: Capture a single frame and analyze it
- **Use Case**: Quick analysis, testing camera setup
- **Features**:
- Single frame capture
- Basic object and scene analysis
- Automatic cleanup
**Usage:**
```bash
python camera-basic.py
```
### 2. `camera-continuous.py` - Continuous Monitoring
- **Purpose**: Continuous camera monitoring with periodic analysis
- **Use Case**: Security monitoring, surveillance systems
- **Features**:
- Configurable analysis intervals
- Real-time monitoring
- Security-focused analysis
- Graceful shutdown with Ctrl+C
**Usage:**
```bash
python camera-continuous.py
```
### 3. `camera-multi-agent.py` - Multi-Agent Analysis
- **Purpose**: Multiple specialized agents analyzing the same camera feed
- **Use Case**: Comprehensive analysis from different perspectives
- **Features**:
- Security analysis agent
- Object detection agent
- Scene analysis agent
- Parallel processing
**Usage:**
```bash
python camera-multi-agent.py
```
### 4. `camera-video-analysis.py` - Video Recording & Analysis
- **Purpose**: Record video segments and analyze temporal events
- **Use Case**: Activity analysis, event detection
- **Features**:
- Video recording with configurable duration
- Temporal event analysis
- Timeline extraction
- Automatic cleanup
**Usage:**
```bash
python camera-video-analysis.py
```
## 🎯 Supported Input Types
- ✅ **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 Settings
```python
camera_id = 0 # Default camera (change to 1, 2, etc. for other cameras)
```
### Analysis Intervals
```python
analysis_interval = 10 # Seconds between analyses for continuous monitoring
```
### Recording Duration
```python
recording_duration = 15 # Seconds for video recording
```
## 💡 Integration Patterns
### 1. Real-time Processing
```python
# Capture frames periodically and process them as separate tasks
def capture_and_analyze():
cap = cv2.VideoCapture(0)
ret, frame = cap.read()
if ret:
cv2.imwrite("temp_capture.jpg", frame)
# Process with PraisonAI agent
cap.release()
```
### 2. Batch Analysis
```python
# Process multiple camera angles simultaneously
task = Task(
description="Analyze multiple camera feeds",
agent=vision_agent,
images=["cam1.jpg", "cam2.jpg", "cam3.jpg"]
)
```
### 3. Video Analysis
```python
# Save video segments and analyze them
task = Task(
description="Analyze this video for activities",
agent=vision_agent,
images=["security_footage.mp4"] # Video files work too
)
```
## 🔒 Security Considerations
- **Camera Permissions**: Ensure your application has camera access
- **Privacy**: Be mindful of privacy when recording/analyzing
- **Storage**: Clean up temporary files to avoid storage issues
- **Access Control**: Implement proper access controls for camera systems
## 🐛 Troubleshooting
### Camera Not Found
```python
# Check available cameras
for i in range(4): # Check first 4 camera indices
cap = cv2.VideoCapture(i)
if cap.isOpened():
print(f"Camera {i} is available")
cap.release()
```
### Permission Issues
- On Linux: Add user to `video` group
- On macOS: Grant camera permissions in System Preferences
- On Windows: Check camera privacy settings
### Performance Optimization
- Reduce frame size for faster processing
- Adjust analysis intervals based on requirements
- Use parallel processing for multiple agents
## 📚 Related Documentation
- [Multimodal Features](../../docs/features/multimodal.mdx)
- [Agent Documentation](../../docs/concepts/agents.mdx)
- [Task Documentation](../../docs/concepts/tasks.mdx)
## 🤝 Contributing
Feel free to contribute additional camera integration examples or improvements to existing ones!
+102
View File
@@ -0,0 +1,102 @@
"""
Basic Camera Integration Example
This example demonstrates how to capture a single frame from a camera
and analyze it using a PraisonAI vision agent.
Requirements:
pip install praisonaiagents opencv-python
Usage:
python camera-basic.py
Environment Variables:
OPENAI_API_KEY=your_openai_api_key
"""
import cv2
import os
from praisonaiagents import Agent, Task, PraisonAIAgents
def capture_and_analyze():
"""Capture a single frame from camera and analyze it"""
# 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 with computer vision skills",
llm="gpt-4o-mini"
)
# Initialize camera (0 for default camera)
cap = cv2.VideoCapture(0)
if not cap.isOpened():
print("Error: Could not open camera")
return None
try:
# Capture frame
ret, frame = cap.read()
if ret:
# Save frame temporarily
image_path = "temp_camera_capture.jpg"
cv2.imwrite(image_path, frame)
print(f"Frame captured and saved as {image_path}")
# Create analysis task
task = Task(
name="analyze_camera_feed",
description="Analyze what you see in this camera capture. Identify objects, people, and activities.",
expected_output="Detailed analysis of the camera content including objects, people, activities, and overall scene description",
agent=vision_agent,
images=[image_path]
)
# Run analysis
agents = PraisonAIAgents(
agents=[vision_agent],
tasks=[task],
process="sequential",
verbose=1
)
result = agents.start()
# Clean up temporary file
if os.path.exists(image_path):
os.remove(image_path)
return result
else:
print("Error: Could not capture frame")
return None
finally:
cap.release()
cv2.destroyAllWindows()
def main():
"""Main function"""
print("📸 Starting basic camera analysis...")
print("Make sure your camera is connected and accessible")
result = capture_and_analyze()
if result:
print("\n🔍 Analysis Results:")
for task_id, task_result in result["task_results"].items():
print(f"\nTask {task_id}:")
print(task_result.raw)
else:
print("❌ Failed to capture and analyze camera feed")
if __name__ == "__main__":
main()
+142
View File
@@ -0,0 +1,142 @@
"""
Continuous Camera Monitoring Example
This example demonstrates continuous camera monitoring with periodic analysis
using PraisonAI vision agents. Perfect for security monitoring or real-time surveillance.
Requirements:
pip install praisonaiagents opencv-python
Usage:
python camera-continuous.py
Environment Variables:
OPENAI_API_KEY=your_openai_api_key
"""
import cv2
import time
import os
from datetime import datetime
from praisonaiagents import Agent, Task, PraisonAIAgents
class ContinuousCameraMonitor:
"""Continuous camera monitoring system"""
def __init__(self, camera_id=0, analysis_interval=10):
"""
Initialize camera monitor
Args:
camera_id (int): Camera ID (0 for default)
analysis_interval (int): Seconds between analyses
"""
self.camera_id = camera_id
self.analysis_interval = analysis_interval
self.running = False
# Create vision agent
self.vision_agent = Agent(
name="SecurityMonitor",
role="Security Camera Analyst",
goal="Monitor camera feeds for activities and security events",
backstory="Expert security analyst with computer vision skills specialized in real-time monitoring",
llm="gpt-4o-mini"
)
def capture_frame(self, cap):
"""Capture a single frame from camera"""
ret, frame = cap.read()
if ret:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"camera_capture_{timestamp}.jpg"
cv2.imwrite(filename, frame)
return filename
return None
def analyze_frame(self, image_path):
"""Analyze captured frame"""
task = Task(
name="monitor_camera_feed",
description="""Monitor this camera feed for:
1. Any people or unusual activities
2. Objects that seem out of place
3. Movement or changes in the scene
4. Security-relevant events
Provide a brief but detailed security report.""",
expected_output="Security analysis report with timestamp and observations",
agent=self.vision_agent,
images=[image_path]
)
agents = PraisonAIAgents(
agents=[self.vision_agent],
tasks=[task],
process="sequential",
verbose=0 # Reduced verbosity for continuous monitoring
)
return agents.start()
def start_monitoring(self):
"""Start continuous monitoring"""
print(f"🎥 Starting continuous camera monitoring...")
print(f"📊 Analysis interval: {self.analysis_interval} seconds")
print("🛑 Press Ctrl+C to stop")
cap = cv2.VideoCapture(self.camera_id)
if not cap.isOpened():
print(f"❌ Error: Could not open camera {self.camera_id}")
return
self.running = True
try:
while self.running:
# Capture frame
image_path = self.capture_frame(cap)
if image_path:
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"\n⏰ {timestamp} - Analyzing camera feed...")
# Analyze frame
result = self.analyze_frame(image_path)
# Display results
if result and "task_results" in result:
for task_id, task_result in result["task_results"].items():
print(f"📋 Security Report: {task_result.raw}")
# Clean up temporary file
if os.path.exists(image_path):
os.remove(image_path)
# Wait for next analysis
time.sleep(self.analysis_interval)
except KeyboardInterrupt:
print("\n🛑 Stopping camera monitoring...")
finally:
self.running = False
cap.release()
cv2.destroyAllWindows()
print("✅ Camera monitoring stopped")
def main():
"""Main function"""
# Configuration
camera_id = 0 # Default camera
analysis_interval = 15 # Analyze every 15 seconds
# Create and start monitor
monitor = ContinuousCameraMonitor(camera_id, analysis_interval)
monitor.start_monitoring()
if __name__ == "__main__":
main()
@@ -0,0 +1,164 @@
"""
Multi-Agent Camera Analysis Example
This example demonstrates using multiple specialized agents to analyze
camera feeds from different perspectives (security, object detection, etc.).
Requirements:
pip install praisonaiagents opencv-python
Usage:
python camera-multi-agent.py
Environment Variables:
OPENAI_API_KEY=your_openai_api_key
"""
import cv2
import os
from datetime import datetime
from praisonaiagents import Agent, Task, PraisonAIAgents
def capture_camera_frame(camera_id=0):
"""Capture a frame from camera"""
cap = cv2.VideoCapture(camera_id)
if not cap.isOpened():
print(f"Error: Could not open camera {camera_id}")
return None
ret, frame = cap.read()
cap.release()
if ret:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
image_path = f"camera_frame_{timestamp}.jpg"
cv2.imwrite(image_path, frame)
return image_path
return None
def analyze_with_multiple_agents():
"""Analyze camera feed using multiple specialized agents"""
# Capture frame
image_path = capture_camera_frame()
if not image_path:
print("Failed to capture camera frame")
return None
print(f"📸 Captured frame: {image_path}")
# Create specialized agents
security_agent = Agent(
name="SecurityExpert",
role="Security Specialist",
goal="Identify security threats and suspicious activities",
backstory="Expert security analyst with years of experience in surveillance and threat detection",
llm="gpt-4o-mini"
)
object_detector = Agent(
name="ObjectDetector",
role="Object Recognition Specialist",
goal="Identify and catalog all objects in the scene",
backstory="Computer vision expert specializing in object detection and classification",
llm="gpt-4o-mini"
)
scene_analyst = Agent(
name="SceneAnalyst",
role="Scene Understanding Expert",
goal="Provide comprehensive scene analysis and context",
backstory="Environmental analyst expert in understanding spatial relationships and scene context",
llm="gpt-4o-mini"
)
# Create specialized tasks
security_task = Task(
name="security_analysis",
description="""Analyze this camera feed for security concerns:
1. Identify any people and their activities
2. Look for suspicious behavior or unauthorized access
3. Check for potential security threats
4. Assess overall security status of the area""",
expected_output="Detailed security assessment with threat level and recommendations",
agent=security_agent,
images=[image_path]
)
object_task = Task(
name="object_detection",
description="""Identify and catalog all objects in this image:
1. List all visible objects with their approximate locations
2. Identify any vehicles, electronics, or valuable items
3. Note any objects that seem out of place
4. Provide a comprehensive inventory""",
expected_output="Detailed object inventory with locations and descriptions",
agent=object_detector,
images=[image_path]
)
scene_task = Task(
name="scene_analysis",
description="""Provide comprehensive scene analysis:
1. Describe the overall environment and setting
2. Analyze lighting conditions and time of day
3. Assess the general activity level and atmosphere
4. Provide context about what type of location this appears to be""",
expected_output="Complete scene description with environmental context",
agent=scene_analyst,
images=[image_path]
)
# Run multi-agent analysis
agents = PraisonAIAgents(
agents=[security_agent, object_detector, scene_analyst],
tasks=[security_task, object_task, scene_task],
process="parallel", # Run all analyses in parallel
verbose=1
)
result = agents.start()
# Clean up temporary file
if os.path.exists(image_path):
os.remove(image_path)
return result
def main():
"""Main function"""
print("🎯 Starting multi-agent camera analysis...")
print("📋 This will analyze the camera feed from multiple perspectives")
result = analyze_with_multiple_agents()
if result:
print("\n" + "="*60)
print("📊 MULTI-AGENT ANALYSIS RESULTS")
print("="*60)
# Display results by agent
agent_names = {
"security_analysis": "🛡️ SECURITY ANALYSIS",
"object_detection": "🔍 OBJECT DETECTION",
"scene_analysis": "🌍 SCENE ANALYSIS"
}
for task_id, task_result in result["task_results"].items():
print(f"\n{agent_names.get(task_id, f'Task {task_id}')}:")
print("-" * 40)
print(task_result.raw)
print("\n" + "="*60)
print("✅ Multi-agent analysis complete!")
else:
print("❌ Failed to complete camera analysis")
if __name__ == "__main__":
main()
@@ -0,0 +1,206 @@
"""
Camera Video Recording and Analysis Example
This example demonstrates recording video segments from camera
and analyzing them using PraisonAI vision agents.
Requirements:
pip install praisonaiagents opencv-python
Usage:
python camera-video-analysis.py
Environment Variables:
OPENAI_API_KEY=your_openai_api_key
"""
import cv2
import os
import time
from datetime import datetime
from praisonaiagents import Agent, Task, PraisonAIAgents
class CameraVideoAnalyzer:
"""Camera video recording and analysis system"""
def __init__(self, camera_id=0):
"""
Initialize video analyzer
Args:
camera_id (int): Camera ID (0 for default)
"""
self.camera_id = camera_id
# Create video analysis agent
self.video_agent = Agent(
name="VideoAnalyst",
role="Video Content Analyzer",
goal="Analyze video content for activities, objects, and events",
backstory="Expert in video analysis and temporal event detection with computer vision expertise",
llm="gpt-4o-mini"
)
def record_video_segment(self, duration_seconds=10):
"""
Record a video segment from camera
Args:
duration_seconds (int): Length of video to record
Returns:
str: Path to recorded video file or None if failed
"""
cap = cv2.VideoCapture(self.camera_id)
if not cap.isOpened():
print(f"Error: Could not open camera {self.camera_id}")
return None
# Get video properties
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))
# Setup video writer
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
video_path = f"camera_recording_{timestamp}.mp4"
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter(video_path, fourcc, fps, (width, height))
print(f"🎬 Recording video for {duration_seconds} seconds...")
print(f"📹 Resolution: {width}x{height}, FPS: {fps}")
start_time = time.time()
frame_count = 0
try:
while (time.time() - start_time) < duration_seconds:
ret, frame = cap.read()
if ret:
out.write(frame)
frame_count += 1
# Show recording indicator
if frame_count % fps == 0: # Update every second
elapsed = int(time.time() - start_time)
remaining = duration_seconds - elapsed
print(f"⏱️ Recording... {remaining}s remaining")
else:
break
finally:
cap.release()
out.release()
cv2.destroyAllWindows()
if frame_count > 0:
print(f"✅ Video recorded: {video_path} ({frame_count} frames)")
return video_path
else:
print("❌ Failed to record video")
if os.path.exists(video_path):
os.remove(video_path)
return None
def analyze_video(self, video_path):
"""
Analyze recorded video
Args:
video_path (str): Path to video file
Returns:
dict: Analysis results
"""
task = Task(
name="analyze_video_content",
description="""Analyze this video recording and provide:
1. Summary of main activities and events observed
2. Timeline of key moments or changes
3. Identification of people, objects, and movements
4. Any notable interactions or behaviors
5. Overall assessment of what happened in the video
Focus on providing a comprehensive temporal analysis.""",
expected_output="Detailed video analysis with timeline and event description",
agent=self.video_agent,
images=[video_path] # PraisonAI supports video files in images parameter
)
agents = PraisonAIAgents(
agents=[self.video_agent],
tasks=[task],
process="sequential",
verbose=1
)
return agents.start()
def record_and_analyze(self, duration=10):
"""
Record video and analyze it
Args:
duration (int): Recording duration in seconds
"""
# Record video
video_path = self.record_video_segment(duration)
if not video_path:
return None
print(f"\n🔍 Analyzing recorded video...")
# Analyze video
result = self.analyze_video(video_path)
# Clean up video file
if os.path.exists(video_path):
os.remove(video_path)
print(f"🗑️ Cleaned up temporary video file")
return result
def main():
"""Main function"""
print("🎥 Camera Video Recording and Analysis")
print("=" * 50)
# Configuration
camera_id = 0
recording_duration = 15 # seconds
print(f"📹 Camera: {camera_id}")
print(f"⏱️ Recording Duration: {recording_duration} seconds")
print("\nMake sure your camera is connected and accessible")
print("The system will record a video segment and then analyze it")
input("\nPress Enter to start recording...")
# Create analyzer and run
analyzer = CameraVideoAnalyzer(camera_id)
result = analyzer.record_and_analyze(recording_duration)
if result:
print("\n" + "="*60)
print("📊 VIDEO ANALYSIS RESULTS")
print("="*60)
for task_id, task_result in result["task_results"].items():
print(f"\nTask: {task_id}")
print("-" * 40)
print(task_result.raw)
print("\n" + "="*60)
print("✅ Video analysis complete!")
else:
print("❌ Failed to record and analyze video")
if __name__ == "__main__":
main()