app.py

import numpy as np
import tensorflow as tf
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.layers import Layer, Dense, Dropout, LayerNormalization, MultiHeadAttention
from tensorflow.keras.models import Sequential
from sklearn.preprocessing import LabelEncoder
import gradio as gr
import os

# Define the EnhancedTransformerBlock class
class EnhancedTransformerBlock(Layer):
    def __init__(self, embed_dim, num_heads, ff_dim, rate=0.1, **kwargs):
        super(EnhancedTransformerBlock, self).__init__(**kwargs)
        self.embed_dim = embed_dim
        self.num_heads = num_heads
        self.ff_dim = ff_dim
        self.rate = rate
        
        self.att = MultiHeadAttention(num_heads=num_heads, key_dim=embed_dim)
        self.ffn = Sequential([
            Dense(ff_dim, activation="relu"),
            Dense(embed_dim),
        ])
        self.layernorm1 = LayerNormalization(epsilon=1e-6)
        self.layernorm2 = LayerNormalization(epsilon=1e-6)
        self.dropout1 = Dropout(rate)
        self.dropout2 = Dropout(rate)
        self.self_attention = MultiHeadAttention(num_heads=1, key_dim=embed_dim)

    def call(self, inputs, training=False):
        attn_output = self.att(inputs, inputs)
        attn_output = self.dropout1(attn_output, training=training)
        out1 = self.layernorm1(inputs + attn_output)
        ffn_output = self.ffn(out1)
        ffn_output = self.dropout2(ffn_output, training=training)
        out2 = self.layernorm2(out1 + ffn_output)
        self_attn_output = self.self_attention(out2, out2)
        return self.layernorm2(out2 + self_attn_output)

    def get_config(self):
        config = super().get_config()
        config.update({
            "embed_dim": self.embed_dim,
            "num_heads": self.num_heads,
            "ff_dim": self.ff_dim,
            "rate": self.rate,
        })
        return config

# Initialize global variables
sequence_length = 10
data = []  # This will store the recent outcomes
encoder = LabelEncoder()

# Try to load the saved model and encoder classes
try:
    model = tf.keras.models.load_model("enhanced_adaptive_model.keras", custom_objects={'EnhancedTransformerBlock': EnhancedTransformerBlock})
    encoder.classes_ = np.load('label_encoder_classes.npy', allow_pickle=True)
except FileNotFoundError:
    print("Model or encoder classes file not found. Using a dummy model and encoder for demonstration.")
    # Create a dummy model and encoder for demonstration
    model = tf.keras.Sequential([tf.keras.layers.Dense(10, input_shape=(sequence_length,), activation='softmax')])
    encoder.classes_ = np.array(['Single small', 'Single big', 'Double small', 'Double big', 'Triple'])

def update_data(data, new_outcome):
    data.append(new_outcome)
    if len(data) > sequence_length:
        data.pop(0)
    return data

def enhanced_predict_next(model, data, sequence_length, encoder):
    last_sequence = data[-sequence_length:]
    last_sequence = np.array(encoder.transform(last_sequence)).reshape((1, sequence_length))
    
    # Monte Carlo Dropout for uncertainty estimation
    predictions = []
    for _ in range(100):
        prediction = model(last_sequence, training=True)
        predictions.append(prediction)
    
    mean_prediction = np.mean(predictions, axis=0)
    std_prediction = np.std(predictions, axis=0)
    
    predicted_label = encoder.inverse_transform([np.argmax(mean_prediction)])
    uncertainty = np.mean(std_prediction)
    
    return predicted_label[0], uncertainty

def gradio_predict(outcome):
    global data

    if outcome not in encoder.classes_:
        return "Invalid outcome. Please try again."

    data = update_data(data, outcome)

    if len(data) < sequence_length:
        return f"Not enough data to make a prediction. Please enter {sequence_length - len(data)} more outcomes."

    predicted_next, uncertainty = enhanced_predict_next(model, data, sequence_length, encoder)
    return f'Predicted next outcome: {predicted_next} (Uncertainty: {uncertainty:.4f})'

def gradio_update(actual_next):
    global data, model

    if actual_next not in encoder.classes_:
        return "Invalid outcome. Please try again."

    data = update_data(data, actual_next)

    if len(data) < sequence_length:
        return f"Not enough data to update the model. Please enter {sequence_length - len(data)} more outcomes."

    encoded_actual_next = encoder.transform([actual_next])[0]
    new_X = np.array(encoder.transform(data[-sequence_length:])).reshape((1, sequence_length))
    new_y = to_categorical(encoded_actual_next, num_classes=len(encoder.classes_))

    model.fit(new_X, new_y, epochs=1, verbose=0)
    return "Model updated with new data."

# Gradio interface
with gr.Blocks() as demo:
    gr.Markdown("## Enhanced Outcome Prediction Model")
    gr.Markdown(f"Enter a sequence of {sequence_length} outcomes to get started.")
    gr.Markdown(f"Valid outcomes: {', '.join(encoder.classes_)}")
    
    with gr.Row():
        outcome_input = gr.Textbox(label="Enter an outcome")
        predict_button = gr.Button("Predict Next")
        predicted_output = gr.Textbox(label="Prediction")
    
    with gr.Row():
        actual_input = gr.Textbox(label="Enter actual next outcome")
        update_button = gr.Button("Update Model")
        update_output = gr.Textbox(label="Update Status")

    predict_button.click(gradio_predict, inputs=outcome_input, outputs=predicted_output)
    update_button.click(gradio_update, inputs=actual_input, outputs=update_output)

demo.launch()