Spaces:
Sleeping
Sleeping
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 | |
def index(): | |
return render_template('index.html') | |
# β Route: Generate Care Plan | |
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 | |
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 | |
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) |