import streamlit as st import tensorflow as tf import numpy as np import requests import zipfile import os import pickle import tensorflow_addons as tfa # GitHub repository details GITHUB_REPO_OWNER = "HeshamSaadi" GITHUB_REPO_NAME = "sentiment-app" GITHUB_RELEASE_TAG = "Model-Release" # Assuming this is the tag for your release GITHUB_MODEL_ZIP_NAME = "models.zip" # Paths for model and tokenizer within the extracted zip MODEL_FILE_NAME = "simplified_lstm_20250612-172438_best.h5" TOKENIZER_FILE_NAME = "simplified_lstm_20250612-172438_tokenizer.pickle" LABEL_MAPPING_FILE_NAME = "simplified_lstm_20250612-172438_label_mapping.pickle" # Custom FocalLoss class from the notebook class FocalLoss(tf.keras.losses.Loss): def __init__(self, gamma=2.0, alpha=0.25, name="focal_loss", **kwargs): super(FocalLoss, self).__init__(name=name, **kwargs) self.gamma = gamma self.alpha = alpha def call(self, y_true, y_pred): y_pred = tf.clip_by_value(y_pred, tf.keras.backend.epsilon(), 1. - tf.keras.backend.epsilon()) pt = tf.where(tf.equal(y_true, 1), y_pred, 1 - y_pred) loss = -tf.keras.backend.mean(self.alpha * tf.keras.backend.pow(1. - pt, self.gamma) * tf.keras.backend.log(pt), axis=-1) return loss def get_config(self): config = super(FocalLoss, self).get_config() config.update({ "gamma": self.gamma, "alpha": self.alpha, }) return config @st.cache_resource def load_model_and_tokenizer(): model = None tokenizer = None label_mapping = None st.write("Downloading model from GitHub...") try: # Construct the URL for the release asset release_url = f"https://api.github.com/repos/{GITHUB_REPO_OWNER}/{GITHUB_REPO_NAME}/releases/tags/{GITHUB_RELEASE_TAG}" headers = {"Accept": "application/vnd.github.v3+json"} response = requests.get(release_url, headers=headers) response.raise_for_status() # Raise an exception for HTTP errors release_info = response.json() asset_url = None for asset in release_info["assets"]: if asset["name"] == GITHUB_MODEL_ZIP_NAME: asset_url = asset["url"] break if not asset_url: st.error(f"Could not find asset \'{GITHUB_MODEL_ZIP_NAME}\' in release \'{GITHUB_RELEASE_TAG}\\\'. Please check the release name and asset name.") return None, None, None # Download the zip file st.write(f"Found asset: {GITHUB_MODEL_ZIP_NAME}. Downloading...") asset_response = requests.get(asset_url, headers={ "Accept": "application/octet-stream", "Authorization": f"token {os.environ.get("GITHUB_TOKEN")}" # Use a token if repo is private or for higher rate limits }, stream=True) asset_response.raise_for_status() zip_path = GITHUB_MODEL_ZIP_NAME with open(zip_path, "wb") as f: for chunk in asset_response.iter_content(chunk_size=8192): f.write(chunk) st.write("Download complete. Extracting files...") # Extract the zip file with zipfile.ZipFile(zip_path, "r") as zip_ref: # Extract all contents directly into the current directory # This assumes the zip contains a \'models\' directory at its root zip_ref.extractall(".") st.write("Extraction complete.") # Define paths to the extracted files # Assuming the zip extracts to a \'models\' directory at the root model_path = os.path.join("models", MODEL_FILE_NAME) tokenizer_path = os.path.join("models", TOKENIZER_FILE_NAME) label_mapping_path = os.path.join("models", LABEL_MAPPING_FILE_NAME) # Custom objects dictionary for model loading custom_objects = { "FocalLoss": FocalLoss, # Registering the custom FocalLoss class # Add any other custom objects from tensorflow_addons if they are explicitly used # For example, if a custom loss function from TFA was used: # \'Addons>SigmoidFocalCrossEntropy\': tfa.losses.SigmoidFocalCrossEntropy # You might need to inspect your model\'s config if this doesn\'t work. } with tf.keras.utils.custom_object_scope(custom_objects): model = tf.keras.models.load_model(model_path) with open(tokenizer_path, "rb") as handle: tokenizer = pickle.load(handle) with open(label_mapping_path, "rb") as handle: label_mapping = pickle.load(handle) st.success("Model and tokenizer loaded successfully!") except requests.exceptions.RequestException as e: st.error(f"Network error during model download: {e}") except zipfile.BadZipFile: st.error("Downloaded file is not a valid zip file.") except FileNotFoundError as e: st.error(f"File not found after extraction: {e}. Please check the paths within your zip file.") except Exception as e: st.error(f"An unexpected error occurred: {e}") st.error("Could not load the model. Please check the GitHub repository and asset name.") return model, tokenizer, label_mapping model, tokenizer, label_mapping = load_model_and_tokenizer() if model and tokenizer and label_mapping: st.title("Sentiment Analysis App") user_input = st.text_area("Enter text for sentiment analysis:", "") if st.button("Analyze Sentiment"): if user_input: # Preprocess the input text (you might need to adapt this based on your model\'s preprocessing) # For an LSTM, typically tokenization and padding are needed # Ensure the tokenizer is fitted on the same vocabulary as during training # and the max_len matches your model\'s input shape. # Example preprocessing (adjust as per your model\'s requirements): # Assuming your tokenizer expects text and outputs sequences # And your model expects padded sequences sequence = tokenizer.texts_to_sequences([user_input]) padded_sequence = tf.keras.preprocessing.sequence.pad_sequences(sequence, maxlen=model.input_shape[1]) # Make prediction prediction = model.predict(padded_sequence) predicted_class = np.argmax(prediction, axis=1)[0] # Map prediction to sentiment label sentiment_labels = {v: k for k, v in label_mapping.items()} predicted_sentiment = sentiment_labels.get(predicted_class, "Unknown") st.write(f"Sentiment: **{predicted_sentiment}**") else: st.warning("Please enter some text to analyze.") else: st.warning("Model could not be loaded. Please check the logs above for details.")