Spaces:
Sleeping
Sleeping
| import subprocess | |
| import streamlit as st | |
| import os | |
| import pandas as pd | |
| import lib.ui.apiClient.APIClient as APIClient | |
| # Function to process the training Excel file | |
| def fn_process_training_excel(lv_file_name, ui_domain_name, ui_status_message, sm): | |
| """ | |
| Process the training Excel file and display status messages based on the response. | |
| Parameters: | |
| - lv_file_name (str): The name of the training Excel file. | |
| - ui_domain_name (str): The domain name for the vector store. | |
| - ui_status_message (str): The UI element to display the status message. | |
| - sm (object): An instance of the StatusMessage class. | |
| Returns: | |
| None | |
| """ | |
| try: | |
| lv_response = APIClient.fn_create_vector_store(lv_file_name, ui_domain_name) | |
| if lv_response.status_code == 200: | |
| # Display a success message | |
| sm.fn_display_status_messages(lv_response.json()['status'], "Success", ui_status_message) | |
| else: | |
| raise Exception(f"Error: {lv_response.json()['error_message']}") | |
| except Exception as e: | |
| # Display an error message if there is an exception | |
| sm.fn_display_status_messages(f"Error: {e}", "Error", ui_status_message) | |
| # Function to process the mapping Excel file | |
| def fn_process_mapping_excel(lv_file_name, ui_source_domain_name, ui_status_message, sm): | |
| """ | |
| Process the mapping Excel file and display status messages based on the response. | |
| Parameters: | |
| - lv_file_name (str): The name of the mapping Excel file. | |
| - ui_source_domain_name (str): The source vector store domain name. | |
| - ui_status_message (str): The UI element to display the status message. | |
| - sm (object): An instance of the StatusMessage class. | |
| Returns: | |
| None | |
| """ | |
| try: | |
| lv_response = APIClient.fn_create_data_mapping(lv_file_name, ui_source_domain_name) | |
| if lv_response.status_code == 200: | |
| lv_response_json = lv_response.json() | |
| lv_response_pd = pd.DataFrame(lv_response_json) | |
| lv_mapping_file_name = 'db/'+ui_source_domain_name+"_"+lv_file_name.split('/')[-1] | |
| lv_response_pd.to_excel(lv_mapping_file_name, index=False) | |
| st.session_state.sv_mapping_file_name = lv_mapping_file_name | |
| else: | |
| raise Exception(f"Error: {lv_response.json()['error_message']}") | |
| except Exception as e: | |
| # Display an error message if there is an exception | |
| sm.fn_display_status_messages(f"Error: {e}", "Error", ui_status_message) | |
| # Function to start the Flask server | |
| def start_flask_server(): | |
| """ | |
| Function to start the Flask server. | |
| Returns: | |
| None | |
| """ | |
| lv_proc = subprocess.Popen(["python", "flask_app.py"]) | |
| st.session_state.sv_flask_server_proc = lv_proc | |
| # Function to stop the Flask server | |
| def stop_flask_server(): | |
| """ | |
| Function to stop the Flask server. | |
| Returns: | |
| None | |
| """ | |
| lv_proc = st.session_state.sv_flask_server_proc | |
| if lv_proc is not None: | |
| lv_proc.terminate() | |
| lv_proc.wait(timeout=5) | |
| if lv_proc.poll() is None: | |
| lv_proc.kill() | |
| st.session_state.sv_flask_server_proc = None | |
| # Function to configure the sidebar UI elements. | |
| def fn_sidebar_configuration(ui_status_message,sm): | |
| """ | |
| Function to configure the sidebar UI elements. | |
| Parameters: | |
| - ui_status_message (str): The status message to be display in UI. | |
| Returns: | |
| None | |
| """ | |
| try: | |
| if not(st.session_state.sv_flask_server_proc): | |
| # Start Flask Server | |
| start_flask_server() | |
| # Toggle button to enable/disable training mode | |
| ui_training_indicator = st.toggle("Train Domain Data", value=False) | |
| # Container to hold the UI elements | |
| ui_container = st.container(border=True) | |
| with ui_container: | |
| # Training new domain | |
| if ui_training_indicator: | |
| # File uploader for training file | |
| ui_training_file = st.file_uploader("Training File", type=["xlsx"], accept_multiple_files=False, key="training_file") | |
| # Text input for domain name | |
| ui_domain_name = st.text_input("Domain Name") | |
| # Button to submit the form | |
| if st.button("Submit"): | |
| # Check if the training file is uploaded | |
| if ui_training_file is not None and ui_domain_name != "": | |
| lv_file_name = "storage/"+ui_domain_name+".xlsx" | |
| # Saving File | |
| with open(lv_file_name, "wb") as f: | |
| f.write(ui_training_file.getvalue()) | |
| # Process the Excel file | |
| fn_process_training_excel(lv_file_name,ui_domain_name,ui_status_message,sm) | |
| else: | |
| # Display an error message if the training file is not uploaded | |
| sm.fn_display_status_messages("Please upload the training file.", "Error", ui_status_message) | |
| # Mapping data to trained domain | |
| else: | |
| # Get all file names that end with ".pak" from the "db" folder | |
| lv_trained_domain_name = [filename[:-len('_index.faiss')] for filename in os.listdir('db') if filename.endswith('_index.faiss')] | |
| # Selectbox for domain selection | |
| ui_training_vector_db = st.selectbox("Domain", lv_trained_domain_name) | |
| # Destination Mapping File | |
| ui_destination_file = st.file_uploader("Destination File", type=["xlsx"], accept_multiple_files=False, key="training_file") | |
| # Button to submit the form | |
| if st.button("Submit"): | |
| if ui_destination_file is not None: | |
| lv_file_name = "storage/"+ui_destination_file.name | |
| # Saving File | |
| with open(lv_file_name, "wb") as f: | |
| f.write(ui_destination_file.getvalue()) | |
| # Process the Excel file | |
| fn_process_mapping_excel(lv_file_name,ui_training_vector_db,ui_status_message,sm) | |
| except Exception as e: | |
| # Display an error message if there is an exception | |
| sm.fn_display_status_messages(f"Error: {e}", "Error", ui_status_message) | |