Spaces:
Runtime error
Runtime error
| import os | |
| from pathlib import Path | |
| from openai import OpenAI | |
| from github import Github | |
| import streamlit as st | |
| def upload_to_github(file_path: str, repo_name: str, github_token: str, branch: str = "main") -> str: | |
| #Uploads the image inputs to github. If the file exists, updates it, otherwise, creates the file | |
| github_token = github_token.strip() | |
| g = Github(github_token) | |
| user = g.get_user() | |
| repo = user.get_repo(repo_name) | |
| with open(file_path, "rb") as f: | |
| content = f.read() | |
| file_name = os.path.basename(file_path) | |
| try: | |
| existing_file = repo.get_contents(file_name, ref=branch) | |
| repo.update_file( | |
| existing_file.path, | |
| f"Update {file_name}", | |
| content, | |
| existing_file.sha, | |
| branch=branch | |
| ) | |
| print(f"Updated {file_name} in repo {repo_name}") | |
| except Exception: | |
| repo.create_file( | |
| file_name, | |
| f"Add {file_name}", | |
| content, | |
| branch=branch | |
| ) | |
| return f"https://raw.githubusercontent.com/{repo.owner.login}/{repo_name}/{branch}/{file_name}" | |
| def text_to_speech(text: str, file_path: str): | |
| #Converts text to speech | |
| client_tts = OpenAI(api_key=os.getenv('OPEN_AI_API_KEY')) | |
| with client_tts.audio.speech.with_streaming_response.create( | |
| model="gpt-4o-mini-tts", | |
| voice="alloy", | |
| input= text | |
| ) as response: | |
| response.stream_to_file(file_path) | |
| def speech_to_text(file_path: str): | |
| #Converts speech to text | |
| client_stt = OpenAI(api_key=os.getenv('OPEN_AI_API_KEY')) | |
| with open(file_path, "rb") as audio_file: | |
| response = client_stt.audio.transcriptions.create( | |
| model="whisper-1", | |
| file=audio_file, | |
| response_format="text" | |
| ) | |
| return response | |
| system_prompt = """"You are a virtual assistant for blind shoppers. Your role is to guide them safely and clearly through a supermarket to find specific items. | |
| Core Principles: Empathy, be kind and supportive; Clarity, Give precise instructions; Safety, Always avoid potential hazards. Rules: | |
| Never assume the user can see. Do not refer to visual cues (e.g., “look for,” “you should see,” etc.); Only describe what is visible in the picture provided. | |
| Shopping List Handling: Greet the user and ask for their shopping list; Once given, treat the list as final; If multiple items are listed, start with the first one (or whichever appears nearest) without asking the user to choose; After locating one item, move directly to the next. | |
| Guidance Workflow | |
| Location Request: Ask the user to take a picture of their current location. | |
| If the picture is unclear, request another from a different angle. | |
| Image Analysis: Describe what’s in front of the user—aisles, shelves, products, and any text. | |
| Directional Guidance: Give step-by-step directions using only steps and turns. | |
| Example: “Take 15 steps forward, then turn left.” | |
| Never tell the user to turn around or walk backward. | |
| Exclusive Support | |
| You are the user’s only source of assistance. | |
| Do not suggest asking store staff for help. | |
| If you find an item, provide information on the items in display. Any additional information on how to pick an item is not necessary. | |
| Safety Checks | |
| Warn the user if any obstacle or hazard is visible in the path.""" | |
| class LLMChat: | |
| def __init__(self,system_prompt=system_prompt,model="gpt-5"): | |
| self.model = model | |
| self.system_prompt = system_prompt | |
| self.history = [{"role":"system","content":system_prompt}] | |
| def llm_call(self, user_prompt, temperature=1): | |
| client = OpenAI( | |
| api_key=os.getenv('OPEN_AI_API_KEY'), | |
| ) | |
| self.history.append({"role": "user", "content": user_prompt}) | |
| response = client.chat.completions.create( | |
| messages=self.history, | |
| model=self.model, | |
| temperature=temperature, | |
| ) | |
| llm_response = str(response.choices[0].message.content).replace("```","").replace("```json","") | |
| self.history.append({"role": "assistant", "content": llm_response}) | |
| return llm_response | |
| if 'chat' not in st.session_state: | |
| chat = LLMChat() | |
| st.session_state.chat = chat | |
| else: | |
| chat = st.session_state.chat | |
| speech_file_path = "audio_support_file.mp3" | |
| st.set_page_config(page_title='Shopping Helper', page_icon='🛒') | |
| st.title('Shopping Helper') | |
| #Ensure intro message only autoplays the first time | |
| if 'intro_autoplay' not in st.session_state: | |
| text_to_speech('Hello! What is on your shopping list today?', speech_file_path) | |
| audio_bytes = open(speech_file_path, 'rb').read() | |
| st.audio(audio_bytes, format='audio/mp3', autoplay=True) | |
| st.session_state.intro_autoplay = True | |
| else: | |
| text_to_speech('Hello! What is on your shopping list today?', speech_file_path) | |
| audio_bytes = open(speech_file_path, 'rb').read() | |
| st.audio(audio_bytes, format='audio/mp3', autoplay=False) | |
| audio_shopping_list = st.audio_input(label='Shopping List') | |
| if audio_shopping_list is not None: #Receive shopping list input | |
| audio_bytes = audio_shopping_list.read() | |
| with open(speech_file_path, "wb") as f: | |
| f.write(audio_bytes) | |
| shopping_list = speech_to_text(speech_file_path) | |
| if audio_shopping_list is not None and shopping_list != '': | |
| if 'shopping_list_response_audio' not in st.session_state: | |
| content = [{"type": "text","text": f"Hello! I need help finding {shopping_list}."}] | |
| response = chat.llm_call(content) | |
| text_to_speech(response, speech_file_path) | |
| st.markdown(response) | |
| shopping_list_response = response | |
| audio_bytes = open(speech_file_path, 'rb').read() | |
| st.audio(audio_bytes, format='audio/mp3', autoplay=True) | |
| st.session_state.shopping_list_response = shopping_list_response | |
| st.session_state.shopping_list_response_audio = audio_bytes | |
| else: | |
| st.markdown(st.session_state.shopping_list_response) | |
| st.audio(st.session_state.shopping_list_response_audio, format='audio/mp3', autoplay=False) | |
| image_file = st.file_uploader("Upload a picture of your current position", type=['png','jpg','jpeg']) | |
| if image_file is not None: | |
| st.image(image_file) | |
| b = image_file.getvalue() | |
| with open(image_file.name, "wb") as f: | |
| f.write(b) | |
| image_data_url = upload_to_github(image_file.name, "data-analysis", os.getenv('GITHUB_ACCESS_TOKEN'), branch = 'llm-helper-pictures') | |
| content = [{"type": "text","text": f"Here is my current position."},{"type": "image_url","image_url": {"url": f"{image_data_url}"}}] | |
| response = chat.llm_call(content) | |
| text_to_speech(response, speech_file_path) | |
| st.markdown(response) | |
| audio_bytes = open(speech_file_path, 'rb').read() | |
| st.audio(audio_bytes, format='audio/mp3', autoplay=True) |