#!/usr/bin/env python3
"""
Backend Verification Script
Runs comprehensive health checks on the backend system
Usage:
python verify_backend.py
"""
import sys
import asyncio
from pathlib import Path
# Color codes for terminal output
GREEN = '\033[92m'
RED = '\033[91m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
RESET = '\033[0m'
def print_header(text):
"""Print section header"""
print(f"\n{BLUE}{'='*60}{RESET}")
print(f"{BLUE}{text}{RESET}")
print(f"{BLUE}{'='*60}{RESET}")
def print_success(text):
"""Print success message"""
print(f"{GREEN}✅ {text}{RESET}")
def print_error(text):
"""Print error message"""
print(f"{RED}❌ {text}{RESET}")
def print_warning(text):
"""Print warning message"""
print(f"{YELLOW}⚠️ {text}{RESET}")
def check_file_structure():
"""Verify all required files exist"""
print_header("File Structure Verification")
required_files = [
'main.py',
'requirements.txt',
'__init__.py',
'agents/__init__.py',
'agents/base_agent.py',
'agents/vision_agent.py',
'agents/planner_agent.py',
'services/__init__.py',
'services/database_service.py',
'services/action_executor.py',
'services/video_processor.py',
'config/__init__.py',
'config/settings.py',
'api/__init__.py',
'api/routes.py',
'utils/__init__.py',
'utils/logger.py',
]
all_exist = True
for file in required_files:
file_path = Path(file)
if file_path.exists():
print_success(f"{file}")
else:
print_error(f"{file} - NOT FOUND")
all_exist = False
return all_exist
def check_dependencies():
"""Verify required packages are installed"""
print_header("Dependency Verification")
required_packages = [
'fastapi',
'uvicorn',
'pydantic',
'google.generativeai',
'cv2',
'numpy',
'aiosqlite',
]
all_installed = True
for package in required_packages:
try:
if package == 'cv2':
__import__('cv2')
elif package == 'google.generativeai':
__import__('google.generativeai')
else:
__import__(package)
print_success(f"{package}")
except ImportError:
print_error(f"{package} - NOT INSTALLED")
all_installed = False
return all_installed
def check_configuration():
"""Verify configuration is valid"""
print_header("Configuration Verification")
try:
from config.settings import settings
# Check API key
if settings.GEMINI_API_KEY and settings.GEMINI_API_KEY != 'your_gemini_api_key_here':
print_success("GEMINI_API_KEY is set")
else:
print_error("GEMINI_API_KEY not set or using placeholder")
return False
# Check other settings
print_success(f"API Host: {settings.API_HOST}")
print_success(f"API Port: {settings.API_PORT}")
print_success(f"Frame Rate: {settings.FRAME_SAMPLE_RATE}s")
print_success(f"Model: {settings.GEMINI_MODEL}")
return True
except Exception as e:
print_error(f"Configuration error: {e}")
return False
def check_database():
"""Verify database can be initialized"""
print_header("Database Verification")
try:
from services.database_service import DatabaseService
import tempfile
# Create temp database
temp_file = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
temp_path = Path(temp_file.name)
temp_file.close()
db = DatabaseService(temp_path)
print_success("Database initialized")
# Test save incident
incident_id = db.save_incident({
'timestamp': '2024-01-01T12:00:00',
'type': 'test',
'severity': 'low',
'confidence': 75,
'reasoning': 'Verification test',
'subjects': [],
'evidence_path': '',
'response_plan': []
})
if incident_id > 0:
print_success(f"Incident saved (ID: {incident_id})")
# Test retrieve
incident = db.get_incident_by_id(incident_id)
if incident:
print_success("Incident retrieved")
# Test stats
stats = db.get_statistics()
if stats:
print_success("Statistics retrieved")
# Cleanup
temp_path.unlink()
return True
except Exception as e:
print_error(f"Database error: {e}")
return False
async def check_agents():
"""Verify agents can be initialized and work"""
print_header("Agent Verification")
try:
from agents.vision_agent import VisionAgent
from agents.planner_agent import PlannerAgent
import numpy as np
# Test VisionAgent
vision = VisionAgent()
print_success("VisionAgent initialized")
# Test with sample frame
frame = np.zeros((480, 640, 3), dtype=np.uint8)
result = await vision.process(frame=frame, frame_number=1)
if result and 'incident' in result:
print_success(f"VisionAgent analysis: {result['type']}")
else:
print_warning("VisionAgent returned unexpected result")
# Test PlannerAgent
planner = PlannerAgent()
print_success("PlannerAgent initialized")
sample_incident = {
'type': 'test',
'severity': 'medium',
'confidence': 80,
'reasoning': 'Test incident'
}
plan = await planner.process(sample_incident)
if plan and isinstance(plan, list):
print_success(f"PlannerAgent created plan ({len(plan)} steps)")
return True
except Exception as e:
print_error(f"Agent error: {e}")
import traceback
traceback.print_exc()
return False
def check_api():
"""Verify API can be imported"""
print_header("API Verification")
try:
from main import app
print_success("FastAPI app imported")
from api.routes import router
print_success("API router imported")
return True
except Exception as e:
print_error(f"API error: {e}")
return False
async def run_all_checks():
"""Run all verification checks"""
print(f"\n{BLUE}🛡️ AegisAI Backend Verification{RESET}")
print(f"{BLUE}{'='*60}{RESET}\n")
results = {
'File Structure': check_file_structure(),
'Dependencies': check_dependencies(),
'Configuration': check_configuration(),
'Database': check_database(),
'Agents': await check_agents(),
'API': check_api(),
}
# Summary
print_header("Verification Summary")
passed = sum(results.values())
total = len(results)
for check, result in results.items():
if result:
print_success(f"{check}")
else:
print_error(f"{check}")
print(f"\n{BLUE}Results: {passed}/{total} checks passed{RESET}")
if passed == total:
print(f"\n{GREEN}✅ ALL CHECKS PASSED!{RESET}")
print(f"\n{GREEN}🚀 Backend is ready to run:{RESET}")
print(f"{BLUE} python main.py{RESET}\n")
return 0
else:
print(f"\n{RED}❌ SOME CHECKS FAILED{RESET}")
print(f"\n{YELLOW}📝 Fix the issues above and run again{RESET}\n")
return 1
def main():
"""Main entry point"""
try:
exit_code = asyncio.run(run_all_checks())
sys.exit(exit_code)
except KeyboardInterrupt:
print(f"\n{YELLOW}Verification cancelled{RESET}\n")
sys.exit(1)
except Exception as e:
print_error(f"Unexpected error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == '__main__':
main()