import streamlit as st import requests import os from gliner import GLiNER from streamlit_autorefresh import st_autorefresh import time import ast tok = os.getenv("TOK") #st_autorefresh(interval=10000, key="volter") st.write(tok) def Target_Identification(userinput): model = GLiNER.from_pretrained("Ihor/gliner-biomed-bi-small-v1.0") labels = ["Protein","Mutation"] entities = model.predict_entities(userinput, labels, threshold=0.5) for entity in entities: if entity["label"] == "Protein": return entity["text"] appuser=st.user def APP(): def sidebarview(): '''experimental sidebar''' # Get projects with error handling and caching @st.cache_data(ttl=20) # Cache for 5 minutes to reduce API calls def scan_for_project_availability_cached(user_id): try: request_url = f"https://thexforce-combat-backend.hf.space/{user_id}/projects" response = requests.get( request_url, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {tok}", }, timeout=10 # Add timeout to prevent hanging ) if response.status_code == 200: response_json = response.json() pros = response_json.get("projects", []) # Default to empty list if no projects key project_list = [] if pros: # Check if pros is not None or empty for pro in pros: if isinstance(pro, dict): project_name = pro.get("project") if project_name: # Only add if project name exists project_list.append(project_name) else: if pro: # Only add if pro is not None or empty project_list.append(str(pro)) return project_list else: st.error(f"Failed to fetch projects. Status code: {response.status_code}") return [] except requests.exceptions.RequestException as e: st.error(f"Error fetching projects: {str(e)}") return [] except Exception as e: st.error(f"Unexpected error: {str(e)}") return [] with st.sidebar: projects = scan_for_project_availability_cached(appuser.get("email")) projectname = None if len(projects) > 0: agree = st.checkbox("create a new project ?") if agree: #projectname = st.selectbox("Select Project", projects) projectname = st.text_input("Enter project name:") st.session_state.projectname=projectname else: projectname = st.selectbox("Select Project", projects) st.session_state.projectname=projectname else: projectname = st.text_input("Enter project name:") st.session_state.projectname = projectname '''experimental sidebar''' sidebarview() tab_map = { 0: "BIO ENGINEERING LAB @newMATTER", 1:"BIO ENGINEERING LAB 2 @newMATTER" } tab_selection = st.pills( "TABS", options=tab_map.keys(), format_func=lambda option: tab_map[option], selection_mode="single", ) def SHOWTABS(): if tab_selection == 0: # Two-column split #left_col, right_col = st.columns([0.3, 0.7],vertical_alignment="center",border=True) # CSS to make right column sticky st.markdown(""" """, unsafe_allow_html=True) ''' with left_col: option_map = { 0: "@OriginAI Nanobody Engineering:", } selection = st.pills( "BIOLOGICS", options=option_map.keys(), format_func=lambda option: option_map[option], selection_mode="single", ) if selection == 0: st.markdown( "

Nanobody [CANCER targeted]

", unsafe_allow_html=True, ) with right_col: ''' bio_input = st.chat_input(" Ready for Action ! ") # FIXED: Removed caching and added force_refresh parameter def fetch_ops(force_refresh=False): # Clear cache if force_refresh is True if force_refresh and 'ops_cache' in st.session_state: del st.session_state.ops_cache fetch_url=f"https://thexforce-combat-backend.hf.space/{appuser.get('email')}/{st.session_state.projectname}/individual/experiment" response = requests.get( fetch_url, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {tok}", }, ) return response.json() if "messages" not in st.session_state: st.session_state.messages = [] if len(st.session_state.messages) > 0: for msg in st.session_state.messages: with st.chat_message(msg["role"]): st.markdown(msg["content"]) if bio_input: st.session_state.messages.append({"role": "user", "content": bio_input}) with st.chat_message("user"): st.markdown(bio_input) if st.session_state.projectname in [None, ""]: st.markdown(":orange-badge[⚠️ Set Projectname]") else: identified_target = Target_Identification(bio_input) #st.warning(f"TARGET IDENTIFIED IS : {identified_target}") payload = { "uid": appuser.get("email"), "pid": st.session_state.projectname, "target": identified_target or None, "high_level_bio_query": bio_input, } response = requests.post( "https://thexforce-combat-backend.hf.space/application_layer_agent", json=payload, headers={ "Content-Type": "application/json", "Authorization":f"Bearer {tok}", }, ) plan_response = response.json() with st.chat_message("assistant"): st.markdown(plan_response) # FIXED: Use animations while waiting for backend processing with st.spinner("🧬 Processing biological data..."): # Try multiple times with short delays between attempts fetch_ops_response = None max_attempts = 6 for attempt in range(max_attempts): fetch_ops_response = fetch_ops(force_refresh=True) # Check if we got data if fetch_ops_response and fetch_ops_response.get("exp"): break # Short delay before next attempt (non-blocking) if attempt < max_attempts - 1: # Don't sleep on last attempt import time time.sleep(0.5) # Much shorter sleep between retries # Show completion animation #if fetch_ops_response and fetch_ops_response.get("exp"): #st.write("") #else: #st.info("⏳ Operations are still being processed...") # FIXED: Better error handling with animations if fetch_ops_response: #st.write("Debug - Full response:", fetch_ops_response) # Debug line - remove in production exp_data = fetch_ops_response.get("exp") if exp_data is not None and len(exp_data) > 0: try: exp_list = ast.literal_eval(exp_data) if isinstance(exp_data, str) else exp_data st.session_state.exp_list = exp_list except Exception as e: st.error(f"Failed to parse experimental data: {e}") st.session_state.exp_list = [] #----------------------------- #RENDER UI FOR PILLS # ----------------------------- #This part must be OUTSIDE `st.chat_message` if "exp_list" in st.session_state and st.session_state.exp_list: tabs = st.tabs([item["operation"] for item in st.session_state.exp_list]) for i, tab in enumerate(tabs): with tab: st.write(st.session_state.exp_list[i]["output"]) else: # Animated waiting message st.info("🔄 No experimental data found yet. Operations may still be processing...") with st.container(): st.markdown("""
""", unsafe_allow_html=True) else: st.error("❌ Failed to fetch operations data") st.markdown("🔄 **Tip:** Try refreshing or submitting your query again.") st.session_state.messages.append( {"role": "assistant", "content": str(plan_response)} ) elif tab_selection == 1: st.markdown("coming soon") if st.user.is_logged_in: if st.button("🚪 Logout"): st.logout() st.rerun() #st.markdown(f"## {st.user.email}") SHOWTABS() else: st.info("Please log in to access the Bio Lab") if st.button("Log in"): st.login("auth0") st.stop()