import os import json import logging import tempfile import re from dotenv import load_dotenv from flask import Flask, request, jsonify, render_template, send_file from groq import Groq import google.generativeai as genai import pandas as pd from datetime import datetime, date, time from apscheduler.schedulers.background import BackgroundScheduler from twilio.rest import Client import io # Load environment variables load_dotenv() app = Flask(__name__) # Configure API keys (keep as is) GROQ_API_KEY = os.getenv('GROQ_API_KEY', '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') GEMINI_API_KEY = os.getenv('GEMINI_API_KEY', 'AIzaSyCizcswP6vlKDMdB3HRAtVi2JbifOpbPvA') # Initialize clients groq_client = Groq(api_key=GROQ_API_KEY) genai.configure(api_key=GEMINI_API_KEY) MODEL_NAME = "gemini-1.5-flash" # Initialize scheduler with local timezone to ensure correct timing scheduler = BackgroundScheduler(timezone='Asia/Kolkata') scheduler.start() # Logger setup logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Twilio configuration (keep as is) ACCOUNT_SID = 'AC490e071f8d01bf0df2f03d086c788d87' AUTH_TOKEN = '224b23b950ad5a4052aba15893fdf083' TWILIO_FROM = 'whatsapp:+14155238886' PATIENT_PHONE = 'whatsapp:+917559355282' # Initialize Twilio client twilio_client = Client(ACCOUNT_SID, AUTH_TOKEN) def send_whatsapp_message(message): """Send WhatsApp notification""" # Format the message with WhatsApp-compatible formatting # Use emojis and formatting to make the feedback link stand out feedback_message = message + "\n\nšŸ”„ Need help or plan not working?\nšŸ‘‰ *FEEDBACK*: https://rajkhanke-patient-feedback-system.hf.space/" try: msg = twilio_client.messages.create( from_=TWILIO_FROM, body=feedback_message, to=PATIENT_PHONE ) logger.info(f"WhatsApp notification sent: {msg.sid}") return True except Exception as e: logger.error(f"Failed to send WhatsApp notification: {str(e)}") return False # Predefined descriptive slots with time objects DESCRIPTIVE_SLOTS = { 'Morning': time(hour=8, minute=0), 'Afternoon': time(hour=13, minute=0), 'Evening': time(hour=18, minute=0), 'Night': time(hour=21, minute=0), } def parse_time_slot(time_str): """ Parse time strings like "hh:mm AM/PM" or descriptive slots. Returns a time object or None. """ # Try "hh:mm AM/PM" try: dt = datetime.strptime(time_str, '%I:%M %p') return dt.time() except ValueError: # Fallback to descriptor map slot = DESCRIPTIVE_SLOTS.get(time_str.title()) if slot: return slot logger.warning(f"Unrecognized time format: '{time_str}'") return None def schedule_notifications(schedule_data): """Schedule WhatsApp notifications.""" try: # Clear existing jobs scheduler.remove_all_jobs() # Sections to schedule sections = [ ('medication_schedule', 'med', {'time': 'Scheduled Time', 'fields': ['Meal', 'Medication Name', 'Dosage']}), ('meal_schedule', 'meal', {'time': 'Time', 'fields': ['Meal', 'Details']}), ('activity_schedule', 'activity', {'time': 'Time', 'fields': ['Activity', 'Duration', 'Notes']}), ] for section, prefix, key_map in sections: for item in schedule_data.get(section, []): time_str = item.get(key_map['time'], '') t = parse_time_slot(time_str) if not t: continue # Build message if section == 'medication_schedule': message = ( f"šŸ”” Medication Reminder\n" f"Time: {time_str}\n" f"Meal: {item.get('Meal', '')}\n" f"Medication: {item.get('Medication Name', '')}\n" f"Dosage: {item.get('Dosage', 'As prescribed')}" ) elif section == 'meal_schedule': message = ( f"šŸ½ Meal Reminder\n" f"Time: {time_str}\n" f"Meal: {item.get('Meal', '')}\n" f"Details: {item.get('Details', '')}" ) else: message = ( f"šŸƒ Activity Reminder\n" f"Time: {time_str}\n" f"Activity: {item.get('Activity', '')}\n" f"Duration: {item.get('Duration', '')}\n" f"Notes: {item.get('Notes', '')}" ) # Generate unique job ID safe_name = re.sub(r"[^A-Za-z0-9_]+", "_", item.get(key_map['fields'][0], '')) safe_time = time_str.replace(':', '').replace(' ', '') job_id = f"{prefix}_{safe_name}_{safe_time}" # Schedule the job scheduler.add_job( send_whatsapp_message, trigger='cron', args=[message], hour=t.hour, minute=t.minute, id=job_id, replace_existing=True ) return True except Exception as e: logger.error(f"Failed to schedule notifications: {e}") return False # āœ… Function to Extract JSON from Response def extract_json(text): json_match = re.search(r'\{.*\}', text, re.DOTALL) if json_match: return json_match.group(0) return None # āœ… Function to Transcribe Audio Using Groq def transcribe_audio(audio_file): try: temp_dir = tempfile.mkdtemp() temp_path = os.path.join(temp_dir, "temp_audio.mp3") audio_file.save(temp_path) # Transcribe audio with open(temp_path, "rb") as file: transcription = groq_client.audio.transcriptions.create( file=(temp_path, file.read()), model="whisper-large-v3-turbo", response_format="json", language="en", temperature=0.0 ) # Cleanup temporary files os.remove(temp_path) os.rmdir(temp_dir) logger.info(f"Transcription successful: {transcription.text[:100]}...") return transcription.text except Exception as e: logger.error(f"Transcription error: {str(e)}") return None # āœ… Function to Generate Care Plan Using Gemini def generate_care_plan(conversation_text): try: model = genai.GenerativeModel(MODEL_NAME) prompt = f""" Based on the following conversation transcript between a doctor and a patient, generate a structured care plan in *valid JSON format*. *JSON Format:* {{ "medication_schedule": [{{"Medication Name": ""}}], "meal_schedule": [{{"Time": "", "Meal": "", "Details": ""}}], "activity_schedule": [{{"Time": "", "Activity": "", "Duration": "", "Notes": ""}}] }} *Fixed Meal Times:* - Breakfast: 7:00 AM - Snack: 10:00 AM - Lunch: 1:00 PM - Snack: 5:00 PM - Dinner: 8:00 PM Ensure medications align with meals. Adjust exercise based on health conditions. *Conversation Transcript:* {conversation_text} *Output Strictly JSON. Do NOT add explanations or extra text.* """ response = model.generate_content(prompt) raw_response = response.text logger.info(f"Raw Gemini Response: {raw_response[:100]}...") json_text = extract_json(raw_response) if json_text: return json.loads(json_text) logger.error("No valid JSON found in response.") return create_default_care_plan() except Exception as e: logger.error(f"Care plan generation error: {str(e)}") return create_default_care_plan() # āœ… Function to Create a Default Care Plan (Fallback) def create_default_care_plan(): return { "medication_schedule": [], "meal_schedule": [ {"Time": "7:00 AM", "Meal": "Breakfast", "Details": ""}, {"Time": "1:00 PM", "Meal": "Lunch", "Details": ""}, {"Time": "8:00 PM", "Meal": "Dinner", "Details": ""} ], "activity_schedule": [] } # āœ… Route: Homepage @app.route('/') def index(): return render_template('index.html') # āœ… Route: Generate Care Plan @app.route("/generate_care_plan", methods=["POST"]) def generate_care_plan_route(): try: if 'audio' not in request.files: return jsonify({"error": "No audio file provided"}), 400 audio_file = request.files['audio'] if not audio_file: return jsonify({"error": "Empty audio file"}), 400 transcript = transcribe_audio(audio_file) if not transcript: return jsonify({"error": "Transcription failed"}), 500 care_plan = generate_care_plan(transcript) return jsonify(care_plan) except Exception as e: logger.error(f"Error in generate_care_plan_route: {str(e)}") return jsonify({"error": str(e)}), 500 # āœ… Route: Generate Medication Schedule @app.route('/generate_schedule', methods=['POST']) def generate_schedule(): try: data = request.get_json() medicines = data.get("medicines", []) schedule_list = [] meal_times = { 3: [("Breakfast", "7:00 AM"), ("Lunch", "1:00 PM"), ("Dinner", "8:00 PM")], 2: [("Breakfast", "7:00 AM"), ("Dinner", "8:00 PM")], 1: [("Dinner", "8:00 PM")] } for med in medicines: medication_name = med.get("medication_name", "") frequency = int(med.get("frequency", 1)) dosage = med.get("dosage", "") for meal, time in meal_times.get(frequency, []): schedule_list.append({ "Medication Name": medication_name, "Dosage": dosage, "Meal": meal, "Scheduled Time": time }) # Schedule notifications schedule_data = { 'medication_schedule': schedule_list, 'meal_schedule': data.get('meal_schedule', []), 'activity_schedule': data.get('activity_schedule', []) } notifications_scheduled = schedule_notifications(schedule_data) return jsonify({ "schedule_list": schedule_list, "notifications_scheduled": notifications_scheduled }) except Exception as e: logger.error(f"Schedule generation error: {str(e)}") return jsonify({"error": str(e)}), 500 @app.route('/download_schedule', methods=['POST']) def download_schedule(): try: data = request.json # Create Excel writer object output = io.BytesIO() with pd.ExcelWriter(output, engine='xlsxwriter') as writer: workbook = writer.book # Convert medication schedule to DataFrame and write to Excel if data.get('medication_schedule'): med_df = pd.DataFrame(data['medication_schedule']) med_df = med_df.sort_values('Time') if 'Time' in med_df.columns else med_df med_df.to_excel(writer, sheet_name='Medication Schedule', index=False) worksheet = writer.sheets['Medication Schedule'] header_format = workbook.add_format({ 'bold': True, 'bg_color': '#3498db', 'font_color': 'white' }) for col_num, value in enumerate(med_df.columns.values): worksheet.write(0, col_num, value, header_format) worksheet.set_column(col_num, col_num, 15) # Set column width # Write meal schedule if data.get('meal_schedule'): meal_df = pd.DataFrame(data['meal_schedule']) meal_df = meal_df.sort_values('Time') meal_df.to_excel(writer, sheet_name='Meal Schedule', index=False) worksheet = writer.sheets['Meal Schedule'] header_format = workbook.add_format({ 'bold': True, 'bg_color': '#2ecc71', 'font_color': 'white' }) for col_num, value in enumerate(meal_df.columns.values): worksheet.write(0, col_num, value, header_format) worksheet.set_column(col_num, col_num, 15) # Write activity schedule if data.get('activity_schedule'): activity_df = pd.DataFrame(data['activity_schedule']) activity_df = activity_df.sort_values('Time') activity_df.to_excel(writer, sheet_name='Activity Schedule', index=False) worksheet = writer.sheets['Activity Schedule'] header_format = workbook.add_format({ 'bold': True, 'bg_color': '#e74c3c', 'font_color': 'white' }) for col_num, value in enumerate(activity_df.columns.values): worksheet.write(0, col_num, value, header_format) worksheet.set_column(col_num, col_num, 15) output.seek(0) return send_file( output, mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', as_attachment=True, download_name='daily_schedule.xlsx' ) except Exception as e: logger.error(f"Error generating Excel file: {str(e)}") return jsonify({"error": str(e)}), 500 # āœ… Run Flask App if __name__ == '__main__': app.run(debug=True, threaded=True)