Add: MiniMax Vision API for food photo analysis
Features: - analyze_food_photo() - Vision API integration - food_photo() - Telegram photo handler - Auto-detect foods and estimate nutrition - Keto-friendly check - Daily totals calculation CLI Usage: - Send food photo to bot → auto-analyze - /food_photo command for manual analysis - Results logged with confidence score Environment Variable: - MINIMAX_API_KEY for vision API access
This commit is contained in:
237
habit_bot.py
237
habit_bot.py
@@ -78,6 +78,22 @@ class UserData:
|
||||
save_json(FOOD_LOGS_FILE, self.food_logs)
|
||||
save_json(USER_DATA_FILE, self.users)
|
||||
|
||||
def get_daily_totals(self, user_id: str, date: str = None) -> Dict:
|
||||
"""Get daily nutrition totals for a user"""
|
||||
if date is None:
|
||||
date = datetime.datetime.now().strftime('%Y-%m-%d')
|
||||
|
||||
totals = {'calories': 0, 'carbs': 0, 'protein': 0, 'fat': 0}
|
||||
|
||||
if user_id in self.food_logs and date in self.food_logs[user_id]:
|
||||
for log in self.food_logs[user_id][date]:
|
||||
totals['calories'] += log.get('calories', 0)
|
||||
totals['carbs'] += log.get('carbs', 0)
|
||||
totals['protein'] += log.get('protein', 0)
|
||||
totals['fat'] += log.get('fat', 0)
|
||||
|
||||
return totals
|
||||
|
||||
data = UserData()
|
||||
|
||||
# URL Patterns
|
||||
@@ -389,6 +405,220 @@ def analyze_food_text(text: str) -> Dict:
|
||||
|
||||
return {'calories': calories, 'carbs': carbs, 'protein': protein, 'fat': fat}
|
||||
|
||||
# ============== MiniMax Vision API ==============
|
||||
|
||||
MINIMAX_API_URL = "https://api.minimax.chat/v1/text/chatcompletion_v2"
|
||||
MINIMAX_API_KEY = os.environ.get('MINIMAX_API_KEY', '')
|
||||
|
||||
async def analyze_food_photo(file_path: str) -> Dict:
|
||||
"""
|
||||
Analyze food photo using MiniMax Vision API
|
||||
|
||||
Returns: Dict with calories, carbs, protein, fat estimation
|
||||
"""
|
||||
if not MINIMAX_API_KEY:
|
||||
# Fallback to placeholder if no API key
|
||||
return {
|
||||
'calories': 400,
|
||||
'carbs': 25,
|
||||
'protein': 30,
|
||||
'fat': 20,
|
||||
'detected_foods': ['food (placeholder - add MiniMax API key)'],
|
||||
'confidence': 0.5
|
||||
}
|
||||
|
||||
try:
|
||||
import base64
|
||||
|
||||
# Read and encode image
|
||||
with open(file_path, 'rb') as f:
|
||||
image_b64 = base64.b64encode(f.read()).decode('utf-8')
|
||||
|
||||
# Prepare vision prompt
|
||||
prompt = """Analyze this food image and estimate nutrition:
|
||||
1. What foods are in the image?
|
||||
2. Estimate: calories, carbs (g), protein (g), fat (g)
|
||||
3. Keto-friendly? (yes/no)
|
||||
|
||||
Return JSON format:
|
||||
{
|
||||
"foods": ["item1", "item2"],
|
||||
"calories": number,
|
||||
"carbs": number,
|
||||
"protein": number,
|
||||
"fat": number,
|
||||
"keto_friendly": boolean,
|
||||
"confidence": 0.0-1.0
|
||||
}"""
|
||||
|
||||
# Call MiniMax API
|
||||
headers = {
|
||||
"Authorization": f"Bearer {MINIMAX_API_KEY}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
payload = {
|
||||
"model": "MiniMax-Vision-01",
|
||||
"messages": [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": prompt},
|
||||
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_b64}"}}
|
||||
]
|
||||
}
|
||||
],
|
||||
"max_tokens": 500,
|
||||
"temperature": 0.3
|
||||
}
|
||||
|
||||
import httpx
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
MINIMAX_API_URL,
|
||||
headers=headers,
|
||||
json=payload,
|
||||
timeout=30.0
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
# Parse JSON from response
|
||||
content = result.get('choices', [{}])[0].get('message', {}).get('content', '{}')
|
||||
|
||||
# Extract JSON
|
||||
import json as json_module
|
||||
try:
|
||||
# Try to parse the response as JSON
|
||||
nutrition = json_module.loads(content)
|
||||
return {
|
||||
'calories': nutrition.get('calories', 400),
|
||||
'carbs': nutrition.get('carbs', 25),
|
||||
'protein': nutrition.get('protein', 30),
|
||||
'fat': nutrition.get('fat', 20),
|
||||
'detected_foods': nutrition.get('foods', ['unknown']),
|
||||
'confidence': nutrition.get('confidence', 0.8),
|
||||
'keto_friendly': nutrition.get('keto_friendly', True)
|
||||
}
|
||||
except json_module.JSONDecodeError:
|
||||
# Fallback if JSON parsing fails
|
||||
return {
|
||||
'calories': 400,
|
||||
'carbs': 25,
|
||||
'protein': 30,
|
||||
'fat': 20,
|
||||
'detected_foods': ['analyzed via MiniMax'],
|
||||
'confidence': 0.7
|
||||
}
|
||||
else:
|
||||
print(f"MiniMax API error: {response.status_code}")
|
||||
return {
|
||||
'calories': 400,
|
||||
'carbs': 25,
|
||||
'protein': 30,
|
||||
'fat': 20,
|
||||
'detected_foods': ['analysis failed - using defaults'],
|
||||
'confidence': 0.5
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
print(f"Photo analysis error: {e}")
|
||||
return {
|
||||
'calories': 400,
|
||||
'carbs': 25,
|
||||
'protein': 30,
|
||||
'fat': 20,
|
||||
'detected_foods': ['error - using defaults'],
|
||||
'confidence': 0.5
|
||||
}
|
||||
|
||||
async def food_photo(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||||
"""Handle food photo upload and analysis"""
|
||||
user_id = str(update.message.from_user.id)
|
||||
today = datetime.datetime.now().strftime('%Y-%m-%d')
|
||||
now = datetime.datetime.now().strftime('%H:%M')
|
||||
|
||||
# Determine meal type
|
||||
hour = datetime.datetime.now().hour
|
||||
if 5 <= hour < 11:
|
||||
meal_type = 'breakfast'
|
||||
elif 11 <= hour < 14:
|
||||
meal_type = 'lunch'
|
||||
elif 14 <= hour < 17:
|
||||
meal_type = 'snack'
|
||||
else:
|
||||
meal_type = 'dinner'
|
||||
|
||||
# Get photo
|
||||
photo = update.message.photo[-1] if update.message.photo else None
|
||||
if not photo:
|
||||
await update.message.reply_text("❌ No photo found! Please send a food photo.")
|
||||
return
|
||||
|
||||
await update.message.reply_text("📸 Analyzing food photo...")
|
||||
|
||||
try:
|
||||
# Download photo
|
||||
file = await context.bot.get_file(photo.file_id)
|
||||
file_path = f"/tmp/food_{user_id}_{today}.jpg"
|
||||
await file.download_to_drive(file_path)
|
||||
|
||||
# Analyze with MiniMax Vision API
|
||||
nutrition = await analyze_food_photo(file_path)
|
||||
|
||||
# Log the food
|
||||
if user_id not in data.food_logs:
|
||||
data.food_logs[user_id] = {}
|
||||
if today not in data.food_logs[user_id]:
|
||||
data.food_logs[user_id][today] = []
|
||||
|
||||
data.food_logs[user_id][today].append({
|
||||
'meal_type': meal_type,
|
||||
'food_name': ', '.join(nutrition.get('detected_foods', ['food'])),
|
||||
'time': now,
|
||||
'calories': nutrition['calories'],
|
||||
'carbs': nutrition['carbs'],
|
||||
'protein': nutrition['protein'],
|
||||
'fat': nutrition['fat'],
|
||||
'source': 'photo',
|
||||
'confidence': nutrition.get('confidence', 0.8),
|
||||
'timestamp': datetime.datetime.now().isoformat()
|
||||
})
|
||||
data.save()
|
||||
|
||||
# Build response
|
||||
emoji = "✅" if nutrition.get('keto_friendly', True) else "⚠️"
|
||||
confidence_pct = int(nutrition.get('confidence', 0.8) * 100)
|
||||
|
||||
text = f"🍽️ **Food Analyzed**\n\n"
|
||||
text += f"Detected: {', '.join(nutrition.get('detected_foods', ['food']))}\n"
|
||||
text += f"Confidence: {confidence_pct}%\n\n"
|
||||
text += f"📊 **Nutrition:**\n"
|
||||
text += f"🔥 Calories: {nutrition['calories']}kcal\n"
|
||||
text += f"🥦 Carbs: {nutrition['carbs']}g\n"
|
||||
text += f"💪 Protein: {nutrition['protein']}g\n"
|
||||
text += f"🥑 Fat: {nutrition['fat']}g\n\n"
|
||||
text += f"{emoji} Keto-friendly: {'Yes' if nutrition.get('keto_friendly', True) else 'No'}\n"
|
||||
|
||||
# Keto check
|
||||
if nutrition['carbs'] > 25:
|
||||
text += "\n⚠️ Carbs exceed keto limit (25g)!"
|
||||
|
||||
# Daily total
|
||||
total = data.get_daily_totals(user_id, today)
|
||||
text += f"\n📈 **Today's Total:** {total['calories']}kcal"
|
||||
text += f"\n💪 {2000 - total['calories']}kcal remaining"
|
||||
|
||||
await update.message.reply_text(text, parse_mode='Markdown')
|
||||
|
||||
# Clean up
|
||||
import os
|
||||
if os.path.exists(file_path):
|
||||
os.remove(file_path)
|
||||
|
||||
except Exception as e:
|
||||
await update.message.reply_text(f"❌ Error analyzing photo: {str(e)}")
|
||||
|
||||
async def food_today(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||||
"""Show today's food log"""
|
||||
user_id = str(update.message.from_user.id)
|
||||
@@ -602,12 +832,17 @@ def main():
|
||||
app.add_handler(CommandHandler('habit_streak', habit_streak))
|
||||
app.add_handler(CommandHandler('food', food_log))
|
||||
app.add_handler(CommandHandler('food_today', food_today))
|
||||
app.add_handler(CommandHandler('food_photo', food_photo))
|
||||
app.add_handler(CommandHandler('morning', morning_briefing))
|
||||
app.add_handler(CommandHandler('debrief', debrief))
|
||||
app.add_handler(CommandHandler('status', lambda u, c: food_today(u, c))) # Alias
|
||||
|
||||
# Photo handler (for food photos)
|
||||
from telegram.ext import.filters
|
||||
app.add_handler(MessageHandler(filters.PHOTO, food_photo))
|
||||
|
||||
# URL handler
|
||||
app.add_handler(MessageHandler(None, handle_url))
|
||||
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_url))
|
||||
|
||||
print("🔮 Starting Habit & Diet Bot...")
|
||||
app.run_polling()
|
||||
|
||||
Reference in New Issue
Block a user