import streamlit as st import anthropic, openai, base64, cv2, glob, json, math, os, pytz, random, re, requests, textract, time, zipfile import plotly.graph_objects as go import streamlit.components.v1 as components from datetime import datetime from audio_recorder_streamlit import audio_recorder from bs4 import BeautifulSoup from collections import deque from dotenv import load_dotenv from gradio_client import Client from huggingface_hub import InferenceClient from io import BytesIO from PIL import Image from PyPDF2 import PdfReader from urllib.parse import quote from xml.etree import ElementTree as ET from openai import OpenAI import extra_streamlit_components as stx from streamlit.runtime.scriptrunner import get_script_run_ctx import asyncio import edge_tts # ensure this is installed (pip install edge-tts) # 🔧 Config & Setup st.set_page_config( page_title="🚲BikeAI🏆 Claude/GPT Research", page_icon="🚲🏆", layout="wide", initial_sidebar_state="auto", menu_items={ 'Get Help': 'https://huggingface.co/awacke1', 'Report a bug': 'https://huggingface.co/spaces/awacke1', 'About': "🚲BikeAI🏆 Claude/GPT Research AI" } ) load_dotenv() openai.api_key = os.getenv('OPENAI_API_KEY') or st.secrets['OPENAI_API_KEY'] anthropic_key = os.getenv("ANTHROPIC_API_KEY_3") or st.secrets["ANTHROPIC_API_KEY"] claude_client = anthropic.Anthropic(api_key=anthropic_key) openai_client = OpenAI(api_key=openai.api_key, organization=os.getenv('OPENAI_ORG_ID')) HF_KEY = os.getenv('HF_KEY') API_URL = os.getenv('API_URL') st.session_state.setdefault('transcript_history', []) st.session_state.setdefault('chat_history', []) st.session_state.setdefault('openai_model', "gpt-4o-2024-05-13") st.session_state.setdefault('messages', []) st.session_state.setdefault('last_voice_input', "") # 🎨 Minimal Custom CSS st.markdown(""" """, unsafe_allow_html=True) # 🔑 Common Utilities def generate_filename(prompt, file_type="md"): ctz = pytz.timezone('US/Central') date_str = datetime.now(ctz).strftime("%m%d_%H%M") safe = re.sub(r'[<>:"/\\\\|?*\n]', ' ', prompt) safe = re.sub(r'\s+', ' ', safe).strip()[:90] return f"{date_str}_{safe}.{file_type}" def create_file(filename, prompt, response): with open(filename, 'w', encoding='utf-8') as f: f.write(prompt + "\n\n" + response) def get_download_link(file): with open(file, "rb") as f: b64 = base64.b64encode(f.read()).decode() return f'📂 Download {os.path.basename(file)}' @st.cache_resource def speech_synthesis_html(result): html_code = f""" """ components.html(html_code, height=0) #------------add EdgeTTS # --- NEW FUNCTIONS FOR EDGE TTS --- async def edge_tts_generate_audio(text, voice="en-US-AriaNeural", rate=0, pitch=0): if not text.strip(): return None rate_str = f"{rate:+d}%" pitch_str = f"{pitch:+d}Hz" communicate = edge_tts.Communicate(text, voice, rate=rate_str, pitch=pitch_str) out_fn = generate_filename(text,"mp3") await communicate.save(out_fn) return out_fn def speak_with_edge_tts(text, voice="en-US-AriaNeural", rate=0, pitch=0): return asyncio.run(edge_tts_generate_audio(text, voice, rate, pitch)) def play_and_download_audio(file_path): if file_path and os.path.exists(file_path): st.audio(file_path) st.markdown(get_download_link(file_path), unsafe_allow_html=True) #--------------------------- def process_image(image_path, user_prompt): with open(image_path, "rb") as imgf: image_data = imgf.read() b64img = base64.b64encode(image_data).decode("utf-8") resp = openai_client.chat.completions.create( model=st.session_state["openai_model"], messages=[ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": [ {"type": "text", "text": user_prompt}, {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64img}"}} ]} ], temperature=0.0, ) return resp.choices[0].message.content def process_audio(audio_path): with open(audio_path, "rb") as f: transcription = openai_client.audio.transcriptions.create(model="whisper-1", file=f) st.session_state.messages.append({"role": "user", "content": transcription.text}) return transcription.text def process_video(video_path, seconds_per_frame=1): vid = cv2.VideoCapture(video_path) total = int(vid.get(cv2.CAP_PROP_FRAME_COUNT)) fps = vid.get(cv2.CAP_PROP_FPS) skip = int(fps*seconds_per_frame) frames_b64 = [] for i in range(0, total, skip): vid.set(cv2.CAP_PROP_POS_FRAMES, i) ret, frame = vid.read() if not ret: break _, buf = cv2.imencode(".jpg", frame) frames_b64.append(base64.b64encode(buf).decode("utf-8")) vid.release() return frames_b64 def process_video_with_gpt(video_path, prompt): frames = process_video(video_path) resp = openai_client.chat.completions.create( model=st.session_state["openai_model"], messages=[ {"role":"system","content":"Analyze video frames."}, {"role":"user","content":[ {"type":"text","text":prompt}, *[{"type":"image_url","image_url":{"url":f"data:image/jpeg;base64,{fr}"}} for fr in frames] ]} ] ) return resp.choices[0].message.content def search_arxiv(query): st.write("🔍 Searching ArXiv...") client = Client("awacke1/Arxiv-Paper-Search-And-QA-RAG-Pattern") r1 = client.predict(prompt=query, llm_model_picked="mistralai/Mixtral-8x7B-Instruct-v0.1", stream_outputs=True, api_name="/ask_llm") st.markdown("### Mistral-8x7B-Instruct-v0.1 Result") st.markdown(r1) r2 = client.predict(prompt=query, llm_model_picked="mistralai/Mistral-7B-Instruct-v0.2", stream_outputs=True, api_name="/ask_llm") st.markdown("### Mistral-7B-Instruct-v0.2 Result") st.markdown(r2) return f"{r1}\n\n{r2}" def perform_ai_lookup(q, vocal_summary=True, extended_refs=False, titles_summary=True): start = time.time() client = Client("awacke1/Arxiv-Paper-Search-And-QA-RAG-Pattern") r = client.predict(q,20,"Semantic Search","mistralai/Mixtral-8x7B-Instruct-v0.1",api_name="/update_with_rag_md") refs = r[0] r2 = client.predict(q,"mistralai/Mixtral-8x7B-Instruct-v0.1",True,api_name="/ask_llm") result = f"### 🔎 {q}\n\n{r2}\n\n{refs}" st.markdown(result) # Main Vocal Summary (Short Answer) if vocal_summary: start_main_part = time.time() audio_file_main = speak_with_edge_tts(r2, voice="en-US-AriaNeural", rate=0, pitch=0) st.write("### 🎙️ Vocal Summary (Short Answer)") play_and_download_audio(audio_file_main) st.write(f"**Elapsed (Short Answer):** {time.time() - start_main_part:.2f} s") # Extended References & Summaries (optional) if extended_refs: start_refs_part = time.time() summaries_text = "Here are the summaries from the references: " + refs.replace('"','') audio_file_refs = speak_with_edge_tts(summaries_text, voice="en-US-AriaNeural", rate=0, pitch=0) st.write("### 📜 Extended References & Summaries") play_and_download_audio(audio_file_refs) st.write(f"**Elapsed (Extended References):** {time.time() - start_refs_part:.2f} s") # Paper Titles Only (short) if titles_summary: start_titles_part = time.time() titles = [] for line in refs.split('\n'): m = re.search(r"\[([^\]]+)\]", line) if m: titles.append(m.group(1)) if titles: titles_text = "Here are the titles of the papers: " + ", ".join(titles) audio_file_titles = speak_with_edge_tts(titles_text, voice="en-US-AriaNeural", rate=0, pitch=0) st.write("### 🔖 Paper Titles") play_and_download_audio(audio_file_titles) st.write(f"**Elapsed (Titles):** {time.time() - start_titles_part:.2f} s") elapsed = time.time()-start st.write(f"**Total Elapsed:** {elapsed:.2f} s") fn = generate_filename(q,"md") create_file(fn,q,result) return result def process_with_gpt(text): if not text: return st.session_state.messages.append({"role":"user","content":text}) with st.chat_message("user"): st.markdown(text) with st.chat_message("assistant"): c = openai_client.chat.completions.create( model=st.session_state["openai_model"], messages=st.session_state.messages, stream=False ) ans = c.choices[0].message.content st.write("GPT-4o: " + ans) create_file(generate_filename(text,"md"),text,ans) st.session_state.messages.append({"role":"assistant","content":ans}) return ans def process_with_claude(text): if not text: return with st.chat_message("user"): st.markdown(text) with st.chat_message("assistant"): r = claude_client.messages.create( model="claude-3-sonnet-20240229", max_tokens=1000, messages=[{"role":"user","content":text}] ) ans = r.content[0].text st.write("Claude: " + ans) create_file(generate_filename(text,"md"),text,ans) st.session_state.chat_history.append({"user":text,"claude":ans}) return ans def create_zip_of_files(): md_files = glob.glob("*.md") mp3_files = glob.glob("*.mp3") all_files = md_files + mp3_files zip_name = "all_files.zip" with zipfile.ZipFile(zip_name,'w') as z: for f in all_files: z.write(f) return zip_name def get_media_html(p,typ="video",w="100%"): d = base64.b64encode(open(p,'rb').read()).decode() if typ=="video": return f'' else: return f'' # CHANGES START HERE: # Define file emojis and the directory to scan FILE_EMOJIS = { "cards": "🗃️", "csv": "📊", "heic": "🖼️", "ico": "🪧", "jpeg": "🖼️", "json": "🔧", "md": "📝", "mid": "🎼", "mov": "🎥", "mp3": "🎵", "mp4": "🎞️", "png": "🖼️", "svg": "🖌️", "txt": "📄", "wav": "🎶", "webm": "📽️", "webp": "🏞️", "zip": "📦", } MEDIA_DIR = "Media" def get_file_ext(filename): return os.path.splitext(filename)[1].lower().strip('.') def load_files(): all_files = [] for root, dirs, files in os.walk(MEDIA_DIR): for f in files: fp = os.path.join(root, f) if os.path.isfile(fp): ext = get_file_ext(fp) mod_time = os.path.getmtime(fp) all_files.append((fp, ext, mod_time)) return all_files def display_files_sidebar(): st.sidebar.title("📂 Media Files") all_files = load_files() from collections import defaultdict ext_map = defaultdict(list) for fp, ext, mod_time in all_files: ext_map[ext].append((fp, mod_time)) # Sort files in each extension group by modification time descending for ext in ext_map: ext_map[ext].sort(key=lambda x: x[1], reverse=True) # Sort extensions by number of files descending sorted_ext = sorted(ext_map.keys(), key=lambda x: len(ext_map[x]), reverse=True) for ext in sorted_ext: emoji = FILE_EMOJIS.get(ext, "📁") count = len(ext_map[ext]) with st.sidebar.expander(f"{emoji} {ext.upper()} ({count})"): for fp, mod_time in ext_map[ext]: basename = os.path.basename(fp) last_mod = datetime.fromtimestamp(mod_time).strftime("%Y-%m-%d %H:%M:%S") col1, col2 = st.columns([3,1]) with col1: st.write(f"**{basename}** - Modified: {last_mod}") with col2: if ext == "mp3": # For MP3, load download link only after user clicks an expander mp3_exp = st.expander("Load MP3 Download Link") with mp3_exp: st.markdown(get_download_link(fp), unsafe_allow_html=True) # If desired, add an on-demand audio player here. else: # Direct download link for other files st.markdown(get_download_link(fp), unsafe_allow_html=True) # CHANGES END HERE def main(): st.sidebar.markdown("### 🚲BikeAI🏆 Multi-Agent Research AI") tab_main = st.radio("Action:",["🎤 Voice Input","📸 Media Gallery","🔍 Search ArXiv","📝 File Editor"],horizontal=True) model_choice = st.sidebar.radio("AI Model:", ["Arxiv","GPT-4o","Claude-3","GPT+Claude+Arxiv"], index=0) # Declare the component mycomponent = components.declare_component("mycomponent", path="mycomponent") val = mycomponent(my_input_value="Hello") if val: user_input = val.strip() if user_input: if model_choice == "GPT-4o": process_with_gpt(user_input) elif model_choice == "Claude-3": process_with_claude(user_input) elif model_choice == "Arxiv": st.subheader("Arxiv Only Results:") perform_ai_lookup(user_input, vocal_summary=True, extended_refs=False, titles_summary=True) else: col1,col2,col3=st.columns(3) with col1: st.subheader("GPT-4o Omni:") try: process_with_gpt(user_input) except: st.write('GPT 4o error') with col2: st.subheader("Claude-3 Sonnet:") try: process_with_claude(user_input) except: st.write('Claude error') with col3: st.subheader("Arxiv + Mistral:") try: perform_ai_lookup(user_input, vocal_summary=True, extended_refs=False, titles_summary=True) except: st.write("Arxiv error") if tab_main == "🔍 Search ArXiv": st.subheader("🔍 Search ArXiv") q=st.text_input("Research query:") # 🎛️ Audio Generation Options st.markdown("### 🎛️ Audio Generation Options") vocal_summary = st.checkbox("🎙️ Vocal Summary (Short Answer)", value=True) extended_refs = st.checkbox("📜 Extended References & Summaries (Long)", value=False) titles_summary = st.checkbox("🔖 Paper Titles Only", value=True) if q: q = q.strip() if q and st.button("Run ArXiv Query"): r = perform_ai_lookup(q, vocal_summary=vocal_summary, extended_refs=extended_refs, titles_summary=titles_summary) st.markdown(r) elif tab_main == "🎤 Voice Input": st.subheader("🎤 Voice Recognition") user_text = st.text_area("Message:", height=100) user_text = user_text.strip() if st.button("Send 📨"): if user_text: if model_choice == "GPT-4o": process_with_gpt(user_text) elif model_choice == "Claude-3": process_with_claude(user_text) elif model_choice == "Arxiv": st.subheader("Arxiv Only Results:") perform_ai_lookup(user_text, vocal_summary=True, extended_refs=False, titles_summary=True) else: col1,col2,col3=st.columns(3) with col1: st.subheader("GPT-4o Omni:") process_with_gpt(user_text) with col2: st.subheader("Claude-3 Sonnet:") process_with_claude(user_text) with col3: st.subheader("Arxiv & Mistral:") res = perform_ai_lookup(user_text, vocal_summary=True, extended_refs=False, titles_summary=True) st.markdown(res) st.subheader("📜 Chat History") t1,t2=st.tabs(["Claude History","GPT-4o History"]) with t1: for c in st.session_state.chat_history: st.write("**You:**", c["user"]) st.write("**Claude:**", c["claude"]) with t2: for m in st.session_state.messages: with st.chat_message(m["role"]): st.markdown(m["content"]) elif tab_main == "📸 Media Gallery": st.header("🎬 Media Gallery - Images and Videos") tabs = st.tabs(["🖼️ Images", "🎥 Video"]) with tabs[0]: imgs = glob.glob("*.png")+glob.glob("*.jpg") if imgs: c = st.slider("Cols",1,5,3) cols = st.columns(c) for i,f in enumerate(imgs): with cols[i%c]: st.image(Image.open(f),use_container_width=True) if st.button(f"👀 Analyze {os.path.basename(f)}"): a = process_image(f,"Describe this image.") st.markdown(a) else: st.write("No images found.") with tabs[1]: vids = glob.glob("*.mp4") if vids: for v in vids: with st.expander(f"🎥 {os.path.basename(v)}"): st.markdown(get_media_html(v,"video"),unsafe_allow_html=True) if st.button(f"Analyze {os.path.basename(v)}"): a = process_video_with_gpt(v,"Describe video.") st.markdown(a) else: st.write("No videos found.") elif tab_main == "📝 File Editor": if getattr(st.session_state,'current_file',None): st.subheader(f"Editing: {st.session_state.current_file}") new_text = st.text_area("Content:", st.session_state.file_content, height=300) if st.button("Save"): with open(st.session_state.current_file,'w',encoding='utf-8') as f: f.write(new_text) st.success("Updated!") else: st.write("Select a file from the sidebar to edit.") # Use the new sidebar display function display_files_sidebar() if __name__=="__main__": main()