Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import numpy as np | |
| import tensorflow as tf | |
| from tensorflow.keras.models import load_model | |
| from tensorflow.keras.layers import Input # Explicitly import Input | |
| # Assuming TKAN and TKAT are available after installing the respective packages | |
| from tkan import TKAN | |
| # If TKAT is from a different library, import it similarly | |
| try: | |
| from tkat import TKAT | |
| except ImportError: | |
| print("TKAT library not found. If your model uses TKAT, make sure the library is installed.") | |
| TKAT = None # Set to None if TKAT is not available | |
| from tensorflow.keras.utils import custom_object_scope | |
| import pickle # Used for saving/loading the scaler | |
| import os # For checking file existence | |
| # --- Configuration --- | |
| MODEL_PATH = "best_model_TKAN_nahead_1 (2).keras" # Your saved model file | |
| INPUT_SCALER_PATH = "input_scaler.pkl" # Your saved input scaler file | |
| SEQUENCE_LENGTH = 24 # Matches the notebook | |
| NUM_INPUT_FEATURES = 5 # ['calculated_aqi', 'temp', 'pm25', 'pm10', 'co'] | |
| N_AHEAD = 1 # Matches the notebook | |
| # --- Ensure Required Files Exist --- | |
| if not os.path.exists(MODEL_PATH): | |
| print(f"Error: Model file not found at {MODEL_PATH}") | |
| import sys | |
| sys.exit("Model file missing. Exiting.") | |
| if not os.path.exists(INPUT_SCALER_PATH): | |
| print(f"Error: Input scaler file not found at {INPUT_SCALER_PATH}") | |
| import sys | |
| sys.exit("Input scaler file missing. Exiting.") | |
| # --- Load Model and Scalers --- | |
| # Define custom objects dictionary | |
| custom_objects = {"TKAN": TKAN} | |
| if TKAT is not None: | |
| custom_objects["TKAT"] = TKAT | |
| # Add your custom MinMaxScaler to custom_objects if you are using one that you defined | |
| # in your own code (not from a library). If your scaler is from scikit-learn, you | |
| # generally don't need to include it in custom_objects for pickle loading, but if it's | |
| # a custom implementation, you do. Based on your notebook, you have a custom MinMaxScaler. | |
| # Include the custom MinMaxScaler class definition here as well. | |
| # --- Your MinMaxScaler Class (Copied from Notebook) --- | |
| class MinMaxScaler: | |
| def __init__(self, feature_axis=None, minmax_range=(0, 1)): | |
| """ | |
| Initialize the MinMaxScaler. | |
| Args: | |
| feature_axis (int, optional): The axis that represents the feature dimension if applicable. | |
| Use only for 3D data to specify which axis is the feature axis. | |
| Default is None, automatically managed based on data dimensions. | |
| """ | |
| self.feature_axis = feature_axis | |
| self.min_ = None | |
| self.max_ = None | |
| self.scale_ = None | |
| self.minmax_range = minmax_range # Default range for scaling (min, max) | |
| def fit(self, X): | |
| """ | |
| Fit the scaler to the data based on its dimensionality. | |
| Args: | |
| X (np.array): The data to fit the scaler on. | |
| """ | |
| if X.ndim == 3 and self.feature_axis is not None: # 3D data | |
| axis = tuple(i for i in range(X.ndim) if i != self.feature_axis) | |
| self.min_ = np.min(X, axis=axis) | |
| self.max_ = np.max(X, axis=axis) | |
| elif X.ndim == 2: # 2D data | |
| self.min_ = np.min(X, axis=0) | |
| self.max_ = np.max(X, axis=0) | |
| elif X.ndim == 1: # 1D data | |
| self.min_ = np.min(X) | |
| self.max_ = np.max(X) | |
| else: | |
| raise ValueError("Data must be 1D, 2D, or 3D.") | |
| self.scale_ = self.max_ - self.min_ | |
| return self | |
| def transform(self, X): | |
| """ | |
| Transform the data using the fitted scaler. | |
| Args: | |
| X (np.array): The data to transform. | |
| Returns: | |
| np.array: The scaled data. | |
| """ | |
| X_scaled = (X - self.min_) / self.scale_ | |
| X_scaled = X_scaled * (self.minmax_range[1] - self.minmax_range[0]) + self.minmax_range[0] | |
| return X_scaled | |
| def fit_transform(self, X): | |
| """ | |
| Fit to data, then transform it. | |
| Args: | |
| X (np.array): The data to fit and transform. | |
| Returns: | |
| np.array: The scaled data. | |
| """ | |
| return self.fit(X).transform(X) | |
| def inverse_transform(self, X_scaled): | |
| """ | |
| Inverse transform the scaled data to original data. | |
| Args: | |
| X_scaled (np.array): The scaled data to inverse transform. | |
| Returns: | |
| np.array: The original data scale. | |
| """ | |
| X = (X_scaled - self.minmax_range[0]) / (self.minmax_range[1] - self.minmax_range[0]) | |
| X = X * self.scale_ + self.min_ | |
| return X | |
| # --- End of MinMaxScaler Class --- | |
| # Add your custom MinMaxScaler to custom_objects | |
| custom_objects["MinMaxScaler"] = MinMaxScaler | |
| model = None | |
| input_scaler = None | |
| # target_scaler = None # Load if needed | |
| try: | |
| # Use custom_object_scope for both model and scaler loading | |
| with custom_object_scope(custom_objects): | |
| model = load_model(MODEL_PATH) | |
| print("Model loaded successfully!") | |
| model.summary() # Verify the model structure after loading | |
| with open(INPUT_SCALER_PATH, 'rb') as f: | |
| input_scaler = pickle.load(f) | |
| print(f"Input scaler loaded successfully from {INPUT_SCALER_PATH}") | |
| # If you also scaled your target variable and need to inverse transform the prediction, | |
| # load the target scaler here as well. | |
| # with custom_object_scope(custom_objects): # Need custom_object_scope if target scaler is custom | |
| # with open(TARGET_SCALER_PATH, 'rb') as f: | |
| # target_scaler = pickle.load(f) | |
| # print(f"Target scaler loaded successfully from {TARGET_SCALER_PATH}") | |
| except Exception as e: | |
| print(f"Error during loading: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| import sys | |
| sys.exit("Failed to load model or scaler. Exiting.") | |
| # --- Data Preparation (get_latest_data_sequence needs implementation) --- | |
| def get_latest_data_sequence(sequence_length, num_features): | |
| """ | |
| Retrieves the latest sequence of data for the required features. | |
| This function needs to be implemented based on your data source in the | |
| deployment environment. It should return a numpy array with shape | |
| (sequence_length, num_features). | |
| Args: | |
| sequence_length (int): The length of the historical sequence required. | |
| num_features (int): The number of features in each time step. | |
| Returns: | |
| np.ndarray: A numpy array containing the historical data sequence. | |
| Shape: (sequence_length, num_features) | |
| """ | |
| print("WARNING: Using dummy data sequence. Implement get_latest_data_sequence.") | |
| # --- REPLACE THIS WITH YOUR ACTUAL DATA RETRIEVAL LOGIC --- | |
| # The data should be in the correct order (oldest to newest time step). | |
| # The columns should be in the order ['calculated_aqi', 'temp', 'pm25', 'pm10', 'co']. | |
| # For now, returning a placeholder with the correct shape. | |
| dummy_data = np.zeros((sequence_length, num_features)) | |
| # Populate dummy_data with some values for testing if you can load historical data | |
| # Example: If you saved a sample of X_test_unscaled, load it here temporarily. | |
| # You need to ensure this dummy data has the correct structure and feature order. | |
| return dummy_data | |
| # --- END OF PLACEHOLDER --- | |
| # --- Define Predict Function --- | |
| def predict(): # Modify inputs as needed based on how you get data | |
| """ | |
| Retrieves the latest data sequence, preprocesses it, and makes a prediction. | |
| The Gradio interface will need to trigger this function. | |
| """ | |
| if model is None or input_scaler is None: | |
| return "Model or scaler not loaded. Check logs." | |
| # 1. Get the latest historical data sequence | |
| latest_data_sequence = get_latest_data_sequence(SEQUENCE_LENGTH, NUM_INPUT_FEATURES) | |
| # Ensure the retrieved data has the correct shape | |
| if latest_data_sequence.shape != (SEQUENCE_LENGTH, NUM_INPUT_FEATURES): | |
| return f"Error: Retrieved data has incorrect shape {latest_data_sequence.shape}. Expected ({SEQUENCE_LENGTH}, {NUM_INPUT_FEATURES})." | |
| # 2. Scale the data sequence using the loaded input scaler | |
| # Your MinMaxScaler from the notebook had feature_axis=2 for 3D data (samples, sequence, features). | |
| # So, for a single sequence (2D array), you need to add a batch dimension (1) before scaling. | |
| latest_data_sequence_with_batch = latest_data_sequence[np.newaxis, :, :] | |
| scaled_input_data = input_scaler.transform(latest_data_sequence_with_batch) | |
| # 3. Perform prediction | |
| # The model expects input shape (batch_size, sequence_length, num_features) | |
| output = model.predict(scaled_input_data) | |
| # 4. Process the output | |
| # The output shape is (batch_size, n_ahead). Since n_ahead=1, shape is (batch_size, 1). | |
| predicted_scaled_value = output[0][0] # Get the first prediction for the first sample | |
| # 5. Inverse transform the prediction if the target was scaled | |
| # If you scaled the target variable (calculated_aqi) before training, | |
| # you need to inverse transform the prediction back to the original scale. | |
| # This requires saving and loading the target_scaler as well and using it here. | |
| # Example if you need to inverse transform the target: | |
| if target_scaler is not None: | |
| # # Need to put the single predicted value into an array with the shape | |
| # # that the target_scaler's inverse_transform expects. | |
| # # Assuming y_scaler was fitted on a shape like (samples, n_ahead, 1) or (samples, 1) | |
| # # and inverse_transform works on a similar shape. | |
| # # If y_train shape was (samples, n_ahead): | |
| predicted_original_scale = target_scaler.inverse_transform(np.array([[predicted_scaled_value]]))[0][0] | |
| # # If y_train shape was (samples, n_ahead, 1): | |
| # # predicted_original_scale = target_scaler.inverse_transform(np.array([[[predicted_scaled_value]]]))[0][0][0] | |
| # pass # Implement the correct inverse transform based on how y_scaler was used | |
| else: | |
| predicted_original_scale = predicted_scaled_value | |
| predicted_value = predicted_original_scale | |
| # For now, assuming the model outputs directly in the desired scale or | |
| # you handle inverse transformation elsewhere if needed. | |
| # predicted_value = predicted_scaled_value # Adjust this if inverse transformation is needed | |
| return float(predicted_value) | |
| # --- Gradio Interface --- | |
| # Keep inputs=None as the predict function gets data internally. | |
| interface = gr.Interface( | |
| fn=predict, | |
| inputs=None, # `predict` function doesn't take direct inputs from Gradio | |
| outputs=gr.Number(label=f"Predicted AQI (Next {N_AHEAD} Hour(s))") | |
| ) | |
| # --- Launch Gradio Interface --- | |
| if __name__ == "__main__": | |
| interface.launch() |