Spaces:
Running
Running
| import streamlit as st | |
| st.set_page_config(page_title="AI Pathology Assistant", layout="wide", initial_sidebar_state="collapsed") | |
| import os | |
| import time | |
| import re | |
| import requests | |
| from PIL import Image | |
| from io import BytesIO | |
| from openai import OpenAI | |
| # ------------------ Authentication ------------------ | |
| VALID_USERS = { | |
| "[email protected]": "Pass.123", | |
| "[email protected]": "Pass.123", | |
| "[email protected]": "Pass.123", | |
| "[email protected]": "Pass.123", | |
| "[email protected]": "Pass.123", | |
| "[email protected]": "Pass.123", | |
| } | |
| def login(): | |
| st.title("🔐 Login Required") | |
| email = st.text_input("Email") | |
| password = st.text_input("Password", type="password") | |
| if st.button("Login"): | |
| if VALID_USERS.get(email) == password: | |
| st.session_state["authenticated"] = True | |
| st.experimental_set_query_params(logged_in="1") | |
| st.rerun() | |
| else: | |
| st.error("❌ Incorrect email or password.") | |
| if not st.session_state.get("authenticated", False): | |
| login() | |
| st.stop() | |
| # ------------------ App Title ------------------ | |
| st.title("🧬 AI Pathology Assistant") | |
| # ------------------ Load OpenAI ------------------ | |
| OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY") | |
| if not OPENAI_API_KEY: | |
| st.error("❌ Missing OPENAI_API_KEY environment variable.") | |
| st.stop() | |
| client = OpenAI(api_key=OPENAI_API_KEY) | |
| # ------------------ Assistant Setup ------------------ | |
| ASSISTANT_ID = "asst_jXDSjCG8LI4HEaFEcjFVq8KB" | |
| # ------------------ Session State Initialization ------------------ | |
| for key in ["messages", "thread_id", "image_urls", "pending_prompt", "image_url", "image_updated"]: | |
| if key not in st.session_state: | |
| st.session_state[key] = [] if key.endswith("s") else None if "url" in key else False | |
| # ------------------ Tabs ------------------ | |
| show_tab2 = True # Set to True to activate visual tab | |
| if show_tab2: | |
| tab1, tab2 = st.tabs(["💬 Chat Assistant", "📷 Visual Reference Search"]) | |
| else: | |
| tab1 = st.tabs(["💬 Chat Assistant"])[0] | |
| tab2 = None | |
| # ------------------ Tab 1: Chat Assistant ------------------ | |
| with tab1: | |
| with st.sidebar: | |
| st.header("🧪 Pathology Tools") | |
| if st.button("🧹 Clear Chat"): | |
| for k in ["messages", "thread_id", "image_urls", "pending_prompt"]: | |
| st.session_state[k] = [] if k.endswith("s") else None | |
| st.rerun() | |
| show_image = st.toggle("📸 Show Images", value=True) | |
| keyword = st.text_input("Keyword Search", placeholder="e.g. mitosis, carcinoma") | |
| if st.button("🔎 Search") and keyword: | |
| st.session_state.pending_prompt = f"Find clauses or references related to: {keyword}" | |
| section = st.text_input("Section Lookup", placeholder="e.g. Connective Tissue") | |
| if section: | |
| st.session_state.pending_prompt = f"Summarize or list key points from section: {section}" | |
| action = st.selectbox("Common Pathology Queries", [ | |
| "Select an action...", | |
| "List histological features of inflammation", | |
| "Summarize features of carcinoma", | |
| "List muscle types and features", | |
| "Extract diagnostic markers", | |
| "Summarize embryology stages" | |
| ]) | |
| if action != "Select an action...": | |
| st.session_state.pending_prompt = action | |
| chat_col, image_col = st.columns([2, 1]) | |
| with chat_col: | |
| st.markdown("### 💬 Ask a Pathology-Specific Question") | |
| user_input = st.chat_input("Example: What are features of squamous cell carcinoma?") | |
| if user_input: | |
| st.session_state.messages.append({"role": "user", "content": user_input}) | |
| elif st.session_state.pending_prompt: | |
| st.session_state.messages.append({"role": "user", "content": st.session_state.pending_prompt}) | |
| st.session_state.pending_prompt = None | |
| # Only trigger assistant if last message is from user | |
| if st.session_state.messages and st.session_state.messages[-1]["role"] == "user": | |
| try: | |
| with st.spinner("🔬 Analyzing..."): | |
| if not st.session_state.thread_id: | |
| st.session_state.thread_id = client.beta.threads.create().id | |
| client.beta.threads.messages.create( | |
| thread_id=st.session_state.thread_id, | |
| role="user", | |
| content=st.session_state.messages[-1]["content"] | |
| ) | |
| run = client.beta.threads.runs.create( | |
| thread_id=st.session_state.thread_id, | |
| assistant_id=ASSISTANT_ID | |
| ) | |
| # Wait for run to complete | |
| while True: | |
| status = client.beta.threads.runs.retrieve( | |
| thread_id=st.session_state.thread_id, | |
| run_id=run.id | |
| ) | |
| if status.status in ("completed", "failed", "cancelled"): | |
| break | |
| time.sleep(1) | |
| if status.status == "completed": | |
| responses = client.beta.threads.messages.list( | |
| thread_id=st.session_state.thread_id | |
| ).data | |
| # Only take the last assistant message | |
| for m in responses: | |
| if m.role == "assistant": | |
| reply = m.content[0].text.value.strip() | |
| if not any( | |
| reply in msg["content"] or msg["content"] in reply | |
| for msg in st.session_state.messages if msg["role"] == "assistant" | |
| ): | |
| st.session_state.messages.append({"role": "assistant", "content": reply}) | |
| # Extract image URLs | |
| images = re.findall( | |
| r'https://raw\.githubusercontent\.com/AndrewLORTech/witspathologai/main/[^\s\n"]+\.png', | |
| reply | |
| ) | |
| st.session_state.image_urls = images | |
| break | |
| else: | |
| st.error("❌ Assistant failed to complete.") | |
| st.rerun() | |
| except Exception as e: | |
| st.error(f"❌ Error: {e}") | |
| # Display all messages | |
| for msg in st.session_state.messages: | |
| with st.chat_message(msg["role"]): | |
| st.markdown(msg["content"], unsafe_allow_html=True) | |
| # Follow-up Questions | |
| if st.session_state.messages and st.session_state.messages[-1]["role"] == "assistant": | |
| last = st.session_state.messages[-1]["content"] | |
| if "Some Possible Questions:" in last: | |
| suggestions = re.findall(r"[-•]\s*(.*)", last) | |
| questions = [q.strip() for q in suggestions if q.strip().endswith("?")] | |
| if questions: | |
| st.markdown("#### 💡 Follow-Up Suggestions") | |
| for q in questions: | |
| if st.button(f"📌 {q}"): | |
| st.session_state.pending_prompt = q | |
| st.rerun() | |
| else: | |
| st.markdown("#### 💡 No follow-up questions detected in the assistant's response.") | |
| with image_col: | |
| if show_image and st.session_state.image_urls: | |
| st.markdown("### 🖼️ Images") | |
| for url in st.session_state.image_urls: | |
| try: | |
| img = Image.open(BytesIO(requests.get(url, timeout=5).content)) | |
| st.image(img, caption=url.split("/")[-1], use_container_width=True) | |
| except Exception: | |
| st.warning(f"⚠️ Failed to load image: {url}") | |
| # ------------------ Tab 2: Visual Reference Search ------------------ | |
| import urllib.parse | |
| import requests | |
| from PIL import Image | |
| from io import BytesIO | |
| with tab2: | |
| ASSISTANT_ID = "asst_9v09zgizdcuuhNdcFQpRo9RO" | |
| if "image_thread_id" not in st.session_state: | |
| st.session_state.image_thread_id = None | |
| if "image_response" not in st.session_state: | |
| st.session_state.image_response = None | |
| if "image_results" not in st.session_state: | |
| st.session_state.image_results = [] | |
| if "image_lightbox" not in st.session_state: | |
| st.session_state.image_lightbox = None | |
| image_input = st.chat_input("Ask for histology visual references (e.g. ovary histology, mitosis)") | |
| if image_input: | |
| st.session_state.image_response = None | |
| st.session_state.image_results = [] | |
| st.session_state.image_lightbox = None | |
| try: | |
| if st.session_state.image_thread_id is None: | |
| thread = client.beta.threads.create() | |
| st.session_state.image_thread_id = thread.id | |
| client.beta.threads.messages.create( | |
| thread_id=st.session_state.image_thread_id, | |
| role="user", | |
| content=image_input | |
| ) | |
| run = client.beta.threads.runs.create( | |
| thread_id=st.session_state.image_thread_id, | |
| assistant_id=ASSISTANT_ID | |
| ) | |
| with st.spinner("🔬 Searching for histology references..."): | |
| while True: | |
| run_status = client.beta.threads.runs.retrieve( | |
| thread_id=st.session_state.image_thread_id, | |
| run_id=run.id | |
| ) | |
| if run_status.status in ("completed", "failed", "cancelled"): | |
| break | |
| time.sleep(1) | |
| if run_status.status == "completed": | |
| messages = client.beta.threads.messages.list(thread_id=st.session_state.image_thread_id) | |
| for msg in reversed(messages.data): | |
| if msg.role == "assistant": | |
| response_text = msg.content[0].text.value | |
| st.session_state.image_response = response_text | |
| # Extract and decode image URLs | |
| lines = response_text.splitlines() | |
| image_urls = [] | |
| expecting_url = False | |
| for line in lines: | |
| line_clean = line.strip().replace("**", "") | |
| if "Image URL:" in line_clean: | |
| parts = line_clean.split("Image URL:") | |
| if len(parts) > 1 and parts[1].strip().startswith("http"): | |
| image_urls.append(urllib.parse.unquote(parts[1].strip())) | |
| else: | |
| expecting_url = True | |
| elif expecting_url: | |
| if line_clean.startswith("http"): | |
| image_urls.append(urllib.parse.unquote(line_clean)) | |
| expecting_url = False | |
| st.session_state.image_results = [{"image": url} for url in image_urls] | |
| if image_urls and not st.session_state.image_lightbox: | |
| st.session_state.image_lightbox = image_urls[0] | |
| break | |
| except Exception as e: | |
| st.error(f"❌ Visual Assistant Error: {e}") | |
| if st.session_state.image_results and st.session_state.image_response: | |
| st.subheader("🖼️ Image Preview(s)") | |
| # Split the assistant response into metadata cards | |
| cards = [] | |
| blocks = st.session_state.image_response.split("### 🖼️ ") | |
| for block in blocks: | |
| if not block.strip(): | |
| continue | |
| lines = block.strip().splitlines() | |
| title = lines[0].strip() | |
| meta = "\n".join(lines[1:]) | |
| cards.append((title, meta)) | |
| cols = st.columns(4) | |
| for i, ((title, meta), item) in enumerate(zip(cards, st.session_state.image_results)): | |
| image_url = item.get("image") | |
| with cols[i % 4]: | |
| with st.container(): | |
| # Remove Image Filename and Image URL lines from display | |
| meta_clean = "\n".join( | |
| line for line in meta.splitlines() | |
| if all(x not in line.lower() for x in ["image filename", "image url"]) | |
| ) | |
| st.markdown(f"**🔬 {title}**") | |
| st.caption(meta_clean) | |
| try: | |
| r = requests.get(image_url, timeout=10) | |
| r.raise_for_status() | |
| img = Image.open(BytesIO(r.content)) | |
| st.image(img, caption=image_url.split("/")[-1], use_container_width=True) | |
| if st.button("🔍 Zoom", key=f"zoom_{i}"): | |
| st.session_state.image_lightbox = image_url | |
| except Exception as e: | |
| st.warning("⚠️ Could not load image.") | |
| st.error(str(e)) | |
| else: | |
| st.info("ℹ️ No image references found yet.") | |
| if st.session_state.image_lightbox: | |
| st.markdown("### 🔬 Full Image View") | |
| try: | |
| img_url = urllib.parse.unquote(st.session_state.image_lightbox) | |
| r = requests.get(img_url, timeout=10) | |
| r.raise_for_status() | |
| full_img = Image.open(BytesIO(r.content)) | |
| st.image(full_img, caption=img_url.split("/")[-1], use_container_width=True) | |
| except Exception as e: | |
| st.warning("⚠️ Could not load full image.") | |
| st.error(str(e)) | |
| if st.button("❌ Close Preview"): | |
| st.session_state.image_lightbox = None | |
| st.rerun() | |