import streamlit as st import pymongo from datetime import date import os import random from models import ( MemoryRecallQuestionandAnswer, MemoryRecallResponseValidation, invoke_model, load_model, ) from dotenv import load_dotenv load_dotenv() import schedule import threading from utils import ( encode_uploaded_image, get_upcoming_events, get_people, get_datetime, get_random_memory, get_random_people, fetch_primary_emails, get_people_name, schedule_checker, check_reminders, encode_image_from_bytes, get_user_details ) from langchain_google_genai import GoogleGenerativeAIEmbeddings from langchain_community.vectorstores import FAISS from langchain.chains.summarize import load_summarize_chain from langchain.prompts import PromptTemplate os.environ["GOOGLE_API_KEY"] = os.getenv("GEMINI_API_KEY") st.set_page_config(page_title="AI Alzheimer Companion", page_icon="🧠", layout="wide") mongodb_uri = os.getenv("MONGODB_URI") mongodb_database = os.getenv("MONGODB_DATABASE") collection_name = "companion" client = pymongo.MongoClient(mongodb_uri) db = client[mongodb_database] collection = db[collection_name] schedule.every().minute.do(check_reminders) threading.Thread(target=schedule_checker, daemon=True).start() if "signed_in" not in st.session_state: st.session_state.signed_in = False if "username" not in st.session_state: st.session_state.username = None def signin(): st.markdown(""" """, unsafe_allow_html=True) st.markdown("

Sign In

Access your account

", unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) with st.form("signin_form"): username = st.text_input("Username", placeholder="Enter your username") password = st.text_input("Password", type="password", placeholder="Enter your password") submitted = st.form_submit_button("Sign In") if submitted: user_document = collection.find_one({ "table_name": "about_user", "about_user.Username": username }) if user_document: stored_password = user_document["about_user"]["Password"] if stored_password == password: st.session_state.signed_in = True st.session_state.username = username st.success("Login successful!") st.rerun() else: st.error("Username not found. Please register.") st.markdown("
", unsafe_allow_html=True) def main(): name = st.session_state.username about_user = get_user_details(name) profile_image_base64 = "" photo_bytes = about_user.get("image") if photo_bytes: if isinstance(photo_bytes, str): profile_image_base64 = photo_bytes print(profile_image_base64[:30]) else: profile_image_base64 = encode_image_from_bytes(photo_bytes) email = about_user.get("Email Retrieval", {}) mail_id = email.get("Gmail Address", "").strip() app_password = email.get("App Password", "").strip() no_of_mails = email.get("Emails to Retrieve", "") st.sidebar.markdown( f"""

Hello {name} πŸ‘‹


""", unsafe_allow_html=True, ) st.markdown( """ """, unsafe_allow_html=True, ) upcoming_events = get_upcoming_events() # ========== Add Person Section ========== with st.sidebar.expander("Add Person", expanded=False): uploaded_file = st.file_uploader( "Upload a person's image", type=["jpg", "png", "jpeg"] ) name = st.text_input("Name") age = st.number_input("Age", min_value=0, max_value=120, step=1) gender = st.selectbox("Gender", ["Male", "Female"]) relation = st.selectbox( "Relation", ["Father", "Mother", "Sister", "Brother", "Friend", "Other"] ) description = st.text_area("Description") mobile_number = st.text_input("Mobile Number") home_town= st.text_input("Home Town") skin_tone = st.selectbox("Skin Tone", ["Not Sure","Light", "Medium", "Dark"]) hair_style = st.selectbox("Hair Style", ["Not Sure","Straight", "Wavy", "Curly", "Bald"]) hair_color = st.selectbox("Hair Color", [ "Not Sure","Black", "Brown", "Blonde", "Grey", "Dyed", "Other"]) glasses = st.selectbox("Wears Glasses?", ["Not Sure", "Yes", "No"]) moles_or_marks = st.text_input("Moles or Distinct Marks (if any)") beard = st.selectbox("Beard", ["Not Sure","Yes", "No"]) mustache = st.selectbox("Mustache", ["Not Sure","Yes", "No"]) if st.button("Add Person"): if uploaded_file and name and relation and description: image_data = encode_uploaded_image(uploaded_file) person_entry = { "age": age, "gender": gender, "image": image_data, "relation": relation, "mobile_number": mobile_number, "home_town": home_town, "description": description, "conversations": {}, "appearance": { "skin_tone": skin_tone, "hair_style": hair_style, "hair_color": hair_color, "glasses": glasses, "moles_or_marks": moles_or_marks, "beard": beard, "mustache": mustache } } existing_people = collection.find_one({"table_name": "people"}) if existing_people: collection.update_one( {"table_name": "people"}, {"$set": {f"people.{name}": person_entry}} ) else: collection.insert_one( {"table_name": "people", "people": {name: person_entry}} ) st.success( "Added to your memory successfully! Every moment matters and is now part of your cherished memories 🧠." ) else: st.error("Please fill in all details and upload an image.") with st.sidebar.expander("Add Conversation", expanded=False): existing_person = get_people_name() person_name = st.selectbox("Name", existing_person) conversation_date = st.date_input("Conversation Date", date.today()) conversation_text = st.text_area("Conversation") if st.button("Add Conversation"): if person_name and conversation_text: conversation_entry = {"conversation": conversation_text} date_str = conversation_date.strftime("%Y-%m-%d") update_result = collection.update_one( {"table_name": "people", f"people.{person_name}": {"$exists": True}}, { "$set": { f"people.{person_name}.conversations.{date_str}": conversation_entry } }, ) if update_result.modified_count > 0: st.success(f"Conversation added for {person_name} on {date_str}.") else: st.error(f"Person named {person_name} not found.") else: st.error("Please provide the person's name and the conversation text.") # ========== Add Memory Section ========== with st.sidebar.expander("Add Memory", expanded=False): memory_title = st.text_input("Memory Title") add_date = st.checkbox("Include memory date?") memory_date = st.date_input("Memory Date") if add_date else None memory_description = st.text_area("Memory Description") if st.button("Add Memory"): if memory_title and memory_description: memory_entry = {"description": memory_description} if add_date and memory_date: memory_entry["date"] = str(memory_date) existing_memories = collection.find_one({"table_name": "memories"}) if existing_memories: collection.update_one( {"table_name": "memories"}, {"$set": {f"memories.{memory_title}": memory_entry}}, ) else: collection.insert_one( {"table_name": "memories", "memories": {memory_title: memory_entry}} ) st.success(f"Memory '{memory_title}' added successfully! πŸ’™") else: st.error("Please fill in all details to add a memory.") # ========== Add Event Section ========== with st.sidebar.expander("Add Upcoming Event", expanded=False): event_title = st.text_input("Event Title") event_date = st.date_input("Event Date") event_description = st.text_area("Event Description") if st.button("Add Event"): if event_title and event_date and event_description: event_entry = {"description": event_description, "date": str(event_date)} existing_events = collection.find_one({"table_name": "events"}) if existing_events: collection.update_one( {"table_name": "events"}, {"$set": {f"events.{event_title}": event_entry}}, ) else: collection.insert_one( {"table_name": "events", "events": {event_title: event_entry}} ) st.success(f"Event '{event_title}' added successfully! πŸ“…") else: st.error("Please fill in all details to add an event.") # ========== Header Section ========== marquee_messages = [ f"Hello {st.session_state.username}! Hope you're having a wonderful day! 😊", f"πŸ“… Today is {get_datetime()}", ] marquee_content = " " * 50 + " | " + " " * 50 marquee_content = marquee_content.join(marquee_messages) marquee_style = """ """ st.markdown(marquee_style, unsafe_allow_html=True) st.markdown( f"""
{marquee_content}
""", unsafe_allow_html=True, ) if "activity" not in st.session_state: st.session_state.activity = random.choice([1, 2, 3]) if "question" not in st.session_state or "answer" not in st.session_state: st.session_state.question = None st.session_state.answer = None left_col, middle_col, right_col = st.columns([1, 1, 1]) # ========== 🧠 Memory Recall Section ========== with left_col: st.markdown( """

Memory Recall

""", unsafe_allow_html=True, ) if st.session_state.activity == 1: # ====== πŸ”Ή Activity 1: Recognizing a Person ====== person_name, person_info = get_random_people() if person_name and person_info: image_data = person_info["image"] st.markdown( f"""

Do you remember this person?

""", unsafe_allow_html=True, ) with st.form("person_form", clear_on_submit=True): user_input = st.text_input( "Enter their name:", placeholder="Type the name here..." ) submitted = st.form_submit_button("Check Answer ➑️") if submitted: if user_input.strip().lower() == person_name.strip().lower(): st.success( f"βœ… Correct! This is {person_name}, your {person_info['relation']}" ) else: st.error( f"This is {person_name}, your ({person_info['relation']})" ) elif st.session_state.activity == 2: # ====== πŸ“– Activity 2: Recalling a Memory ====== memory_title, memory_info = get_random_memory() if memory_title and memory_info: st.markdown( f"""

Do you remember: {memory_title}?

""", unsafe_allow_html=True, ) with st.form("memory_form", clear_on_submit=True): user_input = st.text_area( "Share your thoughts:", placeholder="Write your memories here..." ) submitted = st.form_submit_button("Submit Memory πŸ“") if submitted: validation_prompt = f""" You are a compassionate cognitive assistant helping an Alzheimer's patient recall a past event. Your goal is to validate whether the user’s memory aligns with the expected description of the event. **Guidelines for Validation:** - Accept responses that capture the core meaning, even if phrased differently. - Allow partial recall if key elements are correct. - Do not penalize minor spelling or grammatical errors. - ONLY If the response is completely unrelated, gently indicate that it does not match. **Event Title:** {memory_title} **Expected Memory Description:** {memory_info['description']} **User's Recollection:** {user_input} Determine if the user's response is correct, considering Alzheimer's-related recall difficulties. """ response = invoke_model( validation_prompt, MemoryRecallResponseValidation ) if response.is_correct: st.success("βœ… Great job! Keep rocking. πŸŽ‰") else: st.warning( f"That's an interesting perspective! Here's a hint to help you recall: \n\n **{memory_info['description']}** 😊" ) elif st.session_state.activity == 3: # ====== Activity 3: Answering a General Question ====== prompt = "" if not st.session_state.question: qna_prompt = "​You are a compassionate cognitive assistant dedicated to aiding individuals with Alzheimer's in memory recall. Your task is to generate simple, clear questions that gently stimulate memory without causing frustration. Each question must be designed to elicit a precise, one-word answer. Avoid questions that require memorization of names, dates, places, specific past events, general knowledge, or historical facts. Ensure that each question is easy to understand and answer, aligning with everyday life scenarios." response = invoke_model(qna_prompt, MemoryRecallQuestionandAnswer) st.session_state.question = response.question st.session_state.answer = response.answer question = st.session_state.question answer = st.session_state.answer st.markdown( f"""

{question}

""", unsafe_allow_html=True, ) with st.form("general_form", clear_on_submit=True): user_input = st.text_input( "Your answer:", placeholder="Type your answer here..." ) submitted = st.form_submit_button("Submit") if submitted: validation_prompt = f""" You are a compassionate cognitive assistant designed to validate memory recall responses for Alzheimer's patients. Your goal is to determine whether the user's response conveys the same meaning as the expected answer, even if phrased differently. Validation Criteria: - Accept responses with minor variations, synonyms, or paraphrased meanings that retain the core intent of the expected answer. - Allow sensory descriptions, emotions, or everyday expressions if they align with the expected response. - Do not penalize minor spelling mistakes or slight grammatical differences. - If the response is partially correct, it should still be considered valid. - Mark the response as incorrect only if it is entirely unrelated or contextually incorrect. Validation Task: Question: {question} Expected Answer: {answer} User's Answer: {user_input} Assess whether the user's response aligns with the expected answer while maintaining a gentle and supportive approach. """ response = invoke_model( validation_prompt, MemoryRecallResponseValidation ) if response.is_correct: st.balloons() st.success("Great Job! Keep going") else: st.error(f"The correct answer is **{answer}**.") if submitted: st.session_state.question = None st.session_state.answer = None st.session_state.activity = random.choice([1, 2, 3]) # =========== Mail Summary Section ============ with middle_col: with st.container(): st.markdown( """

Email Highlights

""", unsafe_allow_html=True, ) emails = fetch_primary_emails(mail_id, app_password, no_of_mails) if not emails: st.info("No primary emails found.") else: documents = [email_data['body'] for email_data in emails if email_data['body']] if documents: embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001") vectorstore = FAISS.from_texts(texts=documents, embedding=embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 3}) llm = load_model() if llm is None: st.error("Failed to load the language model.") st.stop() prompt_template = """ You are MemoMate, a gentle memory assistant helping users summarize and remember key information from emails. ### Instructions: - AVOID if any promotional messages, spam, or irrelevant content. - Use the context to generate a friendly and helpful summary only highlighting main elements AVOID mentioning all things. - Focus on important parts like names, dates, events, tasks, or anything that might be a helpful memory cue. - Keep it short and conversational. - If the body is too long or unclear, highlight what seems most relevant. ### Context: {text} ### Response: """ PROMPT = PromptTemplate(template=prompt_template, input_variables=["text"]) summarize_chain = load_summarize_chain( llm=llm, chain_type="map_reduce", map_prompt=PROMPT, combine_prompt=PROMPT ) from langchain.schema import Document email_docs = [Document(page_content=body) for body in documents] with st.spinner("Generating consolidated email summary..."): summary_output = summarize_chain.run(email_docs) consolidated_summary = summary_output.strip() if summary_output else "No summary available." st.markdown(f"""
MemoMate Consolidated Summary: {consolidated_summary}
""", unsafe_allow_html=True) else: st.info("No email bodies available for processing.") # ========== Upcoming Events Section ========== with right_col: st.markdown( """

Upcoming Events

""", unsafe_allow_html=True, ) upcoming_events = get_upcoming_events() if upcoming_events: for title, event_date in upcoming_events: st.markdown( f"""

{title}

{event_date}

""", unsafe_allow_html=True, ) else: st.info("No upcoming events.") # ========== People & Friends Section ========== st.markdown( "

People & Friends

", unsafe_allow_html=True, ) random_people = get_people() if random_people: col_count = 4 row_count = 2 total_slots = col_count * row_count random_people = random_people[:total_slots] cols = st.columns(col_count) for idx, (name, person) in enumerate(random_people): with cols[idx % col_count]: st.markdown( f"""

{name}

{person['relation']}

""", unsafe_allow_html=True, ) else: st.info("No people found. Start adding memories!") # ========== Medication reminder Section ========== with st.expander("πŸ’Š Medication Reminder System", expanded=False): st.markdown("#### Set a Medication Reminder") medication = st.text_input("Medication Name") hour = st.selectbox("Hour", options=[f"{i:02d}" for i in range(1, 13)]) minute = st.selectbox("Minute", options=[f"{i:02d}" for i in range(0, 60)]) period = st.selectbox("AM/PM", options=["AM", "PM"]) phone_number = st.text_input("Phone Number", max_chars=15) if st.button("Set Reminder"): if medication and phone_number: formatted_time = f"{hour}:{minute} {period}" existing_entry = collection.find_one( {"table_name": "medications", "phone_number": phone_number} ) if existing_entry: if medication in existing_entry["medications"]: existing_entry["medications"][medication].append(formatted_time) else: existing_entry["medications"][medication] = [formatted_time] collection.update_one( {"_id": existing_entry["_id"]}, {"$set": {"medications": existing_entry["medications"]}}, ) else: collection.insert_one( { "table_name": "medications", "phone_number": phone_number, "medications": {medication: [formatted_time]}, } ) st.success("Reminder set successfully!") else: st.error("Please fill all fields.") if st.session_state.signed_in: main() else: signin()