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("
", 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"""
""",
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"""
""",
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"""
""",
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()