Spaces:
Paused
Paused
File size: 6,326 Bytes
bc31a94 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | # main.py updates
import streamlit as st
from yoga_analysis import YogaPoseAnalysis
from chat_handler import YogaChatHandler
from image_processor import ImageProcessor
class YogaPoseAnalysisApp:
# ... (previous init and other methods remain the same) ...
def __init__(self):
self.yoga_analysis = YogaPoseAnalysis()
self.chat_handler = YogaChatHandler()
self.image_processor = ImageProcessor()
# Initialize session state for chat history and context
if 'chat_history' not in st.session_state:
st.session_state.chat_history = []
if 'yoga_context' not in st.session_state:
st.session_state.yoga_context = ""
if 'current_image_id' not in st.session_state:
st.session_state.current_image_id = None
def reset_chat_history(self):
"""Reset chat history when new image is processed"""
st.session_state.chat_history = []
st.session_state.yoga_context = ""
def setup_streamlit(self):
st.title("🧘♂️ Yoga Pose Analysis System")
# Sidebar for input methods
with st.sidebar:
analysis_option = st.radio(
"Choose your input method",
("Upload Image", "Provide Image URL")
)
if analysis_option == "Upload Image":
uploaded_file = st.file_uploader(
"Upload Yoga Pose Image",
type=['png', 'jpg', 'jpeg'],
help="Upload a clear photo of your yoga pose for analysis"
)
if uploaded_file:
# Check if this is a new image
current_file_id = hash(uploaded_file.getvalue())
if current_file_id != st.session_state.current_image_id:
self.reset_chat_history()
st.session_state.current_image_id = current_file_id
self.process_uploaded_file(uploaded_file)
elif analysis_option == "Provide Image URL":
image_url = st.text_input("Enter the Image URL")
if image_url:
# Check if this is a new URL
current_url_id = hash(image_url)
if current_url_id != st.session_state.current_image_id:
self.reset_chat_history()
st.session_state.current_image_id = current_url_id
self.process_image_url(image_url)
# Main content area
if 'current_image' in st.session_state:
st.image(st.session_state.current_image, caption="Yoga Pose", use_container_width=True)
if 'analysis_result' in st.session_state:
self.display_analysis_results()
# Chat interface at the bottom
if 'analysis_result' in st.session_state:
self.display_chat_interface()
def process_uploaded_file(self, uploaded_file):
with st.spinner('Uploading and analyzing image...'):
image_path = self.image_processor.save_uploaded_file(uploaded_file)
st.session_state.current_image = uploaded_file
self.perform_analysis(image_path)
def process_image_url(self, image_url):
with st.spinner('Downloading and analyzing image...'):
image_path = self.image_processor.download_image(image_url)
if image_path:
st.session_state.current_image = image_url
self.perform_analysis(image_path)
def perform_analysis(self, image_path):
analysis_result = self.yoga_analysis.analyze_image(image_path)
st.session_state.analysis_result = analysis_result
# Update yoga context with the full analysis
st.session_state.yoga_context = analysis_result['full_analysis']
def display_analysis_results(self):
result = st.session_state.analysis_result
st.subheader("📊 Pose Analysis")
st.metric(label="Identified Pose", value=result['pose_name'])
with st.expander("📝 Detailed Pose Analysis"):
st.markdown(result['full_analysis'])
def display_chat_interface(self):
st.subheader("Chat Assistant")
# Display chat history
for message in st.session_state.chat_history:
with st.chat_message(message["role"]):
st.markdown(message["content"])
# Chat input at the bottom
with st.container():
st.markdown('<div style="margin-bottom: 50px;"></div>', unsafe_allow_html=True)
user_query = st.chat_input("Ask a question about your pose or yoga practice")
if user_query:
# Add user message to chat history
st.session_state.chat_history.append({"role": "user", "content": user_query})
# Create a placeholder for streaming response
with st.chat_message("assistant"):
with st.spinner('Typing...'):
message_placeholder = st.empty()
full_response = ""
# Get streaming response
for response_chunk in self.chat_handler.get_streaming_response(
user_query,
st.session_state.yoga_context
):
full_response += response_chunk
# Add a blinking cursor to simulate typing
message_placeholder.markdown(full_response + "▌")
# Remove cursor and show final response
message_placeholder.markdown(full_response)
# Add assistant response to chat history
st.session_state.chat_history.append({"role": "assistant", "content": full_response})
# Rerun to update chat display
st.rerun()
def main():
app = YogaPoseAnalysisApp()
app.setup_streamlit()
if __name__ == "__main__":
main()
|