#!/usr/bin/env python3
"""
๐ฏ EKALAVYA - The Ultimate AI Teaching Assistant
๐ Multi-Modal โข Multi-Lingual โข Memory-Powered
๐ก๏ธ Safe โข ๐ Educational โข ๐ Friendly
"""
from fastapi import FastAPI, HTTPException, UploadFile, File, Form
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from typing import Optional, List, Dict
import json
import os
# Import all modules
from model.teaching import TeachingMode
from model.safety import SafetyRules
from model.memory import MemorySystem
# ๐ฏ Initialize FastAPI with style
app = FastAPI(
title="๐ฏ EKALAVYA API",
description="๐ The Ultimate AI Teaching Assistant - Multi-Modal, Multi-Lingual, Memory-Powered",
version="3.0.0",
docs_url="/docs",
redoc_url="/redoc"
)
# ๐ก๏ธ Initialize safety rules
safety = SafetyRules()
# ๐ฏ Initialize teaching mode
teaching_mode = TeachingMode(style="friend")
# ๐ Data directory
DATA_DIR = "data"
os.makedirs(DATA_DIR, exist_ok=True)
# ๐ฆ Request/Response Models
class TeachingRequest(BaseModel):
"""๐ Teaching request model"""
input_text: str
user_id: str = "default_user"
conversation_style: str = "friend" # friend, teacher, lover, mentor
language: str = "english"
class SafetyCheckRequest(BaseModel):
"""๐ก๏ธ Safety check request model"""
content: str
check_type: str = "all" # all, scam, hacking, privacy, inappropriate
class ProgressRequest(BaseModel):
"""๐ Progress request model"""
user_id: str
class StyleRequest(BaseModel):
"""๐ Conversation style request model"""
style: str # friend, teacher, lover, mentor
user_id: str = "default_user"
# ๐ Root endpoint
@app.get("/", response_class=HTMLResponse)
async def root():
"""๐ฏ Welcome page with emojis"""
return """
๐ฏ EKALAVYA - AI Teaching Assistant
๐ฏ EKALAVYA
๐ The Ultimate AI Teaching Assistant ๐
๐ก๏ธ 100%
Safe & Private
โจ Features
๐ Teaching Mode
Learn English with real-time mistake detection and correction
๐ง Memory System
Remembers your mistakes and tracks your learning progress
๐ Conversation Styles
Choose: Friend ๐ซ, Teacher ๐จโ๐ซ, Lover ๐, or Mentor ๐
๐ก๏ธ Safety First
Scam detection, hacking prevention, privacy protection
๐ Multi-Modal
Supports text, images, video, and audio
๐ 100% Private
All data stays on your device, no tracking
๐ API Endpoints
POST /teach - Start learning session
POST /safety/check - Check content safety
GET /safety/tips - Get safety tips
POST /progress - View learning progress
POST /style - Change conversation style
๐ Quick Links
๐ Interactive API Docs |
๐ Alternative Docs |
๐ฏ HuggingFace Model
๐ฏ Built with โค๏ธ for learners everywhere ๐
"""
# ๐ Teaching endpoint
@app.post("/teach")
async def teach(request: TeachingRequest):
"""๐ Start teaching session with mistake detection"""
try:
# ๐ก๏ธ Safety check first
safety_check = safety.check_content(request.input_text)
if not safety_check['is_safe']:
return {
"status": "๐ก๏ธ safety_warning",
"message": safety_check['warnings'][0],
"suggestions": safety_check['suggestions']
}
# ๐ฏ Process teaching request
result = teaching_mode.process_teaching_request(
user_input=request.input_text,
user_id=request.user_id,
conversation_style=request.conversation_style,
language=request.language
)
return {
"status": "โ
success",
"response": result['response'],
"mistakes_found": result['mistakes'],
"corrections": result['corrections'],
"explanation": result['explanation'],
"encouragement": result['encouragement'],
"next_steps": result['next_steps'],
"emoji": "๐" if result['mistakes'] else "โจ"
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Error: {str(e)}")
# ๐ก๏ธ Safety check endpoint
@app.post("/safety/check")
async def check_safety(request: SafetyCheckRequest):
"""๐ก๏ธ Check if content is safe"""
result = safety.check_content(request.content)
return {
"is_safe": result['is_safe'],
"violations": result['violations'],
"warnings": result['warnings'],
"suggestions": result['suggestions'],
"emoji": "โ
" if result['is_safe'] else "โ ๏ธ",
"message": "โ
Content is safe!" if result['is_safe'] else "โ ๏ธ Safety issues detected"
}
# ๐ก Safety tips endpoint
@app.get("/safety/tips")
async def get_safety_tips():
"""๐ก Get safety tips with emojis"""
tips = safety.get_safety_tips()
emoji_tips = [
f"๐ {tips['privacy_protection'][0]}",
f"๐ก๏ธ {tips['privacy_protection'][1]}",
f"๐ซ {tips['prohibited_actions'][0]}",
f"โ ๏ธ {tips['prohibited_actions'][1]}",
f"๐ {tips['positive_behaviors'][0]}",
f"๐ {tips['positive_behaviors'][1]}",
]
return {
"tips": emoji_tips,
"count": len(emoji_tips),
"emoji": "๐ก",
"message": "๐ก Stay safe with these tips!"
}
# ๐ Progress endpoint
@app.post("/progress")
async def get_progress(request: ProgressRequest):
"""๐ Get user learning progress"""
memory = MemorySystem(user_id=request.user_id)
stats = memory.get_user_stats()
# Calculate learning score
total_attempts = stats['total_attempts']
correct_attempts = stats['correct_attempts']
learning_score = (correct_attempts / total_attempts * 100) if total_attempts > 0 else 0
return {
"user_id": request.user_id,
"stats": stats,
"learning_score": round(learning_score, 2),
"emoji": "๐" if learning_score > 80 else "๐" if learning_score > 50 else "๐ช",
"message": "๐ Excellent progress!" if learning_score > 80 else
"๐ Good progress, keep going!" if learning_score > 50 else
"๐ช Keep practicing, you'll improve!"
}
# ๐ Style change endpoint
@app.post("/style")
async def change_style(request: StyleRequest):
"""๐ Change conversation style"""
styles = {
"friend": "๐ซ",
"teacher": "๐จโ๐ซ",
"lover": "๐",
"mentor": "๐"
}
if request.style not in styles:
raise HTTPException(
status_code=400,
detail=f"โ Invalid style. Choose from: {', '.join(styles.keys())}"
)
teaching_mode.set_style(request.style, request.user_id)
return {
"status": "โ
success",
"style": request.style,
"emoji": styles[request.style],
"message": f"{styles[request.style]} Now talking as your {request.style}!"
}
# ๐ Languages endpoint
@app.get("/languages")
async def get_languages():
"""๐ Get supported languages with flags"""
languages = {
"english": {"name": "English", "flag": "๐ฌ๐ง", "emoji": "๐"},
"hindi": {"name": "Hindi", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"bengali": {"name": "Bengali", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"telugu": {"name": "Telugu", "flag": "๐ฎ๐ณ", "emoji": "โ๏ธ"},
"tamil": {"name": "Tamil", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"marathi": {"name": "Marathi", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"gujarati": {"name": "Gujarati", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"kannada": {"name": "Kannada", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"malayalam": {"name": "Malayalam", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"odia": {"name": "Odia", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"punjabi": {"name": "Punjabi", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"assamese": {"name": "Assamese", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"urdu": {"name": "Urdu", "flag": "๐ต๐ฐ", "emoji": "๐"},
"maithili": {"name": "Maithili", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"santali": {"name": "Santali", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"kashmiri": {"name": "Kashmiri", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"nepali": {"name": "Nepali", "flag": "๐ณ๐ต", "emoji": "๐"},
"sindhi": {"name": "Sindhi", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"konkani": {"name": "Konkani", "flag": "๐ฎ๐ณ", "emoji": "โ๏ธ"},
"dogri": {"name": "Dogri", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"manipuri": {"name": "Manipuri", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"bodo": {"name": "Bodo", "flag": "๐ฎ๐ณ", "emoji": "๐"},
"sanskrit": {"name": "Sanskrit", "flag": "๐ฎ๐ณ", "emoji": "๐"}
}
return {
"languages": languages,
"count": len(languages),
"emoji": "๐",
"message": f"๐ Supporting {len(languages)} languages!"
}
# ๐ฅ Health check endpoint
@app.get("/health")
async def health_check():
"""๐ฅ Health check with status"""
return {
"status": "โ
healthy",
"service": "๐ฏ EKALAVYA",
"version": "๐ฆ 3.0.0",
"emoji": "๐ข",
"message": "๐ข All systems operational!"
}
# ๐ฏ Main entry point
if __name__ == "__main__":
import uvicorn
print("""
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ โ
โ ๐ฏ EKALAVYA - AI Teaching Assistant โ
โ โ
โ ๐ Multi-Modal โข Multi-Lingual โข Memory-Powered โ
โ โ
โ ๐ก๏ธ Safe โข ๐ Educational โข ๐ Friendly โ
โ โ
โ ๐ API Docs: http://localhost:8000/docs โ
โ โ
โ ๐ Supporting 23 Indian Languages โ
โ โ
โ ๐ง 1M Token Context Window โ
โ โ
โ ๐ 100% Private & Secure โ
โ โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
""")
uvicorn.run(app, host="0.0.0.0", port=8000)
# ๐ฅ VIDEO AI ENDPOINTS
from model.video_ai import VideoAnalyzer, CameraProcessor
# ๐ฌ Initialize video AI
video_analyzer = VideoAnalyzer()
camera_processor = CameraProcessor()
class VideoRequest(BaseModel):
"""๐ฅ Video processing request"""
video_path: str
operation: str = "analyze" # analyze, enhance, stabilize, remove_objects, extract_info
class RealTimeVideoRequest(BaseModel):
"""๐บ Real-time video request"""
source: int = 0 # 0 for camera, other for screen
duration: int = 30 # seconds
# ๐ฌ Video Analysis endpoint
@app.post("/video/analyze")
async def analyze_video(request: VideoRequest):
"""๐ฌ Analyze video content with AI"""
try:
result = video_analyzer.analyze_video_content(request.video_path)
return {
"status": "โ
success",
"analysis": result,
"emoji": "๐ฌ",
"message": "๐ฌ Video analysis complete!"
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Video analysis failed: {str(e)}")
# ๐จ Video Enhancement endpoint
@app.post("/video/enhance")
async def enhance_video(request: VideoRequest):
"""๐ฏ Enhance video quality"""
try:
output_path = f"enhanced_{request.video_path.split('/')[-1]}"
result = video_analyzer.enhance_video_quality(request.video_path, output_path)
return {
"status": "โ
success",
"output_file": output_path,
"enhancements": result["enhancements"],
"emoji": "๐ฏ",
"message": "๐ฏ Video enhanced successfully!"
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Video enhancement failed: {str(e)}")
# ๐บ Video Stabilization endpoint
@app.post("/video/stabilize")
async def stabilize_video(request: VideoRequest):
"""๐บ Stabilize shaky video"""
try:
output_path = f"stabilized_{request.video_path.split('/')[-1]}"
result = video_analyzer.stabilize_video(request.video_path, output_path)
return {
"status": "โ
success",
"output_file": output_path,
"stabilization_level": result["stabilization_level"],
"emoji": "๐บ",
"message": "๐บ Video stabilized successfully!"
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Video stabilization failed: {str(e)}")
# ๐จ Object Removal endpoint
@app.post("/video/remove-objects")
async def remove_objects(request: VideoRequest):
"""๐จ Remove objects from video"""
try:
output_path = f"cleaned_{request.video_path.split('/')[-1]}"
result = video_analyzer.remove_objects(request.video_path, output_path)
return {
"status": "โ
success",
"output_file": output_path,
"frames_processed": result["frames_processed"],
"emoji": "๐จ",
"message": "๐จ Objects removed successfully!"
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Object removal failed: {str(e)}")
# ๐ Information Extraction endpoint
@app.post("/video/extract-info")
async def extract_video_info(request: VideoRequest):
"""๐ Extract information from video"""
try:
result = video_analyzer.extract_information(request.video_path)
return {
"status": "โ
success",
"extracted_info": result,
"emoji": "๐",
"message": "๐ Information extracted successfully!"
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Information extraction failed: {str(e)}")
# ๐บ Real-time Video Analysis endpoint
@app.post("/video/realtime")
async def real_time_video(request: RealTimeVideoRequest):
"""๐บ Real-time video analysis from camera or screen"""
try:
result = video_analyzer.real_time_analysis(source=request.source)
return {
"status": "โ
success",
"frames_analyzed": result["frames_analyzed"],
"analysis_results": result["analysis_results"][:10], # Last 10 results
"emoji": "๐บ",
"message": "๐บ Real-time analysis complete!"
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Real-time analysis failed: {str(e)}")
# ๐ท Camera Processing endpoint
@app.post("/camera/process")
async def process_camera():
"""๐ท Process camera frame for visual recognition"""
try:
# Capture frame from camera
import cv2
cap = cv2.VideoCapture(0)
ret, frame = cap.read()
cap.release()
if not ret:
raise HTTPException(status_code=500, detail="โ Cannot capture from camera")
# Process frame
result = camera_processor.process_camera_frame(frame)
recognition = camera_processor.recognize_visual_elements(frame)
return {
"status": "โ
success",
"frame_analysis": result,
"visual_recognition": recognition,
"emoji": "๐ท",
"message": "๐ท Camera frame processed successfully!"
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Camera processing failed: {str(e)}")
# ๐ฅ Video Features Summary endpoint
@app.get("/video/features")
async def get_video_features():
"""๐ฅ Get all video AI features"""
features = {
"video_analysis": {
"emoji": "๐ฌ",
"description": "Understand and analyze video content",
"capabilities": [
"Scene detection",
"Object recognition",
"Motion analysis",
"Quality assessment",
"Content classification"
]
},
"video_enhancement": {
"emoji": "๐ฏ",
"description": "Enhance video quality",
"capabilities": [
"Brightness adjustment",
"Color correction",
"Sharpening",
"Contrast enhancement",
"Noise reduction"
]
},
"video_stabilization": {
"emoji": "๐บ",
"description": "Stabilize shaky video",
"capabilities": [
"Motion compensation",
"Frame alignment",
"Smooth transitions",
"Jitter removal",
"Professional stabilization"
]
},
"object_removal": {
"emoji": "๐จ",
"description": "Remove unwanted objects",
"capabilities": [
"Object detection",
"Smart inpainting",
"Background reconstruction",
"Seamless removal",
"Batch processing"
]
},
"information_extraction": {
"emoji": "๐",
"description": "Extract information from video",
"capabilities": [
"Text recognition (OCR)",
"Data extraction",
"Pattern detection",
"Key moment identification",
"Metadata analysis"
]
},
"real_time_analysis": {
"emoji": "๐บ",
"description": "Real-time video processing",
"capabilities": [
"Live camera feed",
"Screen share analysis",
"Instant object detection",
"Real-time classification",
"Live streaming support"
]
},
"camera_processing": {
"emoji": "๐ท",
"description": "Camera and visual recognition",
"capabilities": [
"Face detection",
"Object recognition",
"Scene classification",
"Visual element detection",
"Real-time processing"
]
}
}
return {
"features": features,
"total_features": len(features),
"emoji": "๐ฅ",
"message": "๐ฅ Complete video AI suite available!"
}
# ๐ป CODING AI ENDPOINTS
from model.coding_ai import CodingAI
# ๐งโ๐ป Initialize coding AI
coding_ai = CodingAI()
class CodingRequest(BaseModel):
"""๐ป Coding request"""
code: str
language: str = "python"
operation: str = "debug" # debug, refactor, analyze, generate
task: str = ""
class CodebaseRequest(BaseModel):
"""๐๏ธ Codebase analysis request"""
project_path: str
class AutonomousRequest(BaseModel):
"""๐ค Autonomous task request"""
task: str
project_path: str = ""
# ๐ป Code analysis endpoint
@app.post("/coding/analyze")
async def analyze_code(request: CodingRequest):
"""๐ป Analyze and debug code"""
try:
if request.operation == "debug":
result = coding_ai.debug_code(request.code, request.task)
elif request.operation == "refactor":
result = coding_ai.refactor_code(request.code, request.language, request.task)
elif request.operation == "generate":
result = coding_ai.generate_code(request.task, request.language)
else:
result = {"status": "โ unknown operation"}
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Error: {str(e)}")
# ๐๏ธ Codebase analysis endpoint
@app.post("/coding/codebase")
async def analyze_codebase(request: CodebaseRequest):
"""๐๏ธ Analyze entire codebase"""
try:
result = coding_ai.analyze_codebase(request.project_path)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Error: {str(e)}")
# ๐ค Autonomous coding endpoint
@app.post("/coding/autonomous")
async def autonomous_coding(request: AutonomousRequest):
"""๐ค Execute autonomous coding task"""
try:
result = coding_ai.autonomous_task(request.task, request.project_path)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Error: {str(e)}")
# ๐ผ๏ธ VISION AI ENDPOINTS
from model.vision_ai import VisionAI
# ๐จ Initialize vision AI
vision_ai = VisionAI()
class ImageGenerationRequest(BaseModel):
"""๐จ Image generation request"""
prompt: str
style: str = "realistic"
size: List[int] = [512, 512]
class ImageAnalysisRequest(BaseModel):
"""๐๏ธ Image analysis request"""
image_data: str
class ImageProcessingRequest(BaseModel):
"""๐ง Image processing request"""
image_data: str
operation: str = "enhance" # enhance, resize, grayscale, blur, sharpen, edge_detect
# ๐จ Image generation endpoint
@app.post("/vision/generate")
async def generate_image(request: ImageGenerationRequest):
"""๐จ Generate image from text"""
try:
result = vision_ai.generate_image(
request.prompt,
request.style,
tuple(request.size)
)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Error: {str(e)}")
# ๐๏ธ Image analysis endpoint
@app.post("/vision/analyze")
async def analyze_image(request: ImageAnalysisRequest):
"""๐๏ธ Analyze image content"""
try:
result = vision_ai.analyze_image(request.image_data)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Error: {str(e)}")
# ๐ง Image processing endpoint
@app.post("/vision/process")
async def process_image(request: ImageProcessingRequest):
"""๐ง Process image"""
try:
result = vision_ai.process_image(request.image_data, request.operation)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"โ Error: {str(e)}")
# ๐ฏ ALL CAPABILITIES ENDPOINT
@app.get("/capabilities")
async def get_all_capabilities():
"""๐ฏ Get all EKALAVYA capabilities"""
return {
"name": "๐ฏ EKALAVYA",
"version": "๐ฆ 3.0",
"status": "โ
Most Powerful AI",
"capabilities": {
"๐ป Coding": {
"emoji": "๐ป",
"features": [
"๐๏ธ Large project understanding",
"๐ง Code refactoring",
"๐ Debugging & bug fixing",
"๐ป Code generation",
"๐ค Autonomous coding agent",
"๐งช Test generation",
"๐ Documentation"
],
"better_than": "Claude, ChatGPT"
},
"๐ผ๏ธ Image": {
"emoji": "๐ผ๏ธ",
"features": [
"๐จ Image generation",
"๐๏ธ Image analysis",
"๐ง Image processing",
"๐ Visual understanding",
"๐จ Style transfer"
],
"better_than": "Gemini, ChatGPT"
},
"๐ฅ Video": {
"emoji": "๐ฅ",
"features": [
"๐ฌ Video analysis",
"๐ฏ Video enhancement",
"๐บ Video stabilization",
"๐จ Object removal",
"๐ Information extraction",
"๐บ Real-time analysis",
"๐ท Camera processing"
],
"better_than": "Gemini, Samsung, iPhone"
},
"๐ Teaching": {
"emoji": "๐",
"features": [
"๐ Mistake detection",
"โ
Instant corrections",
"๐ Detailed explanations",
"๐ช Encouragement",
"๐ฏ Personalized learning"
],
"better_than": "All others"
},
"๐ง Memory": {
"emoji": "๐ง ",
"features": [
"๐ Progress tracking",
"๐ Improvement monitoring",
"๐ฏ Weak area identification",
"๐ฌ Conversation memory",
"๐ Pattern analysis"
],
"better_than": "Claude, ChatGPT, Gemini"
},
"๐ Styles": {
"emoji": "๐",
"features": [
"๐ซ Friend style",
"๐จโ๐ซ Teacher style",
"๐ Lover style",
"๐ Mentor style"
],
"better_than": "All others (unique)"
},
"๐ Languages": {
"emoji": "๐",
"features": [
"๐ฎ๐ณ 23 Indian languages",
"๐ฌ๐ง English",
"๐ค Multi-lingual support"
],
"better_than": "All others"
},
"๐ก๏ธ Safety": {
"emoji": "๐ก๏ธ",
"features": [
"๐ก๏ธ Scam detection",
"๐ป Hacking prevention",
"๐ Privacy protection",
"โ
Ethical guidelines",
"๐ Privacy policy"
],
"better_than": "All others"
},
"๐ง Reasoning": {
"emoji": "๐ง ",
"features": [
"๐ญ Deep reasoning",
"๐ Step-by-step analysis",
"๐ฏ Problem solving",
"๐ Complex tasks"
],
"better_than": "ChatGPT"
},
"โ๏ธ Writing": {
"emoji": "โ๏ธ",
"features": [
"๐ Creative writing",
"๐ Technical writing",
"๐ฏ Precise editing",
"๐ก Style adaptation"
],
"better_than": "Claude"
}
},
"emoji": "๐",
"message": "๐ EKALAVYA - The Most Powerful AI Assistant!"
}