Blame for app.py

#!/usr/bin/env python
"""
SKENAI-RENO Event Processing Demo
---------------------------------
This Gradio app demonstrates how the SKENAI-RENO event processing pipeline
enriches events with AI insights using Hugging Face models.
"""

import gradio as gr
import requests
import json
import os
import time
import numpy as np
import pandas as pd
import plotly.express as px
from datetime import datetime, timedelta

# Configure API access (in production, use Secrets)
API_KEY = os.environ.get("HF_API_KEY", "")
HF_ENDPOINT = "https://api-inference.huggingface.co/models/"

# Sample models to use for different event types
MODEL_MAPPING = {
    "user_feedback": "distilbert-base-uncased-finetuned-sst-2-english",
    "transaction": "bert-base-uncased",
    "system_event": "gpt2",
    "achievement": "roberta-base"
}

def process_event(event_type, event_content, user_id="user123", game_id="game456"):
    """Process an event through the SKENAI-RENO pipeline with Hugging Face model."""
    
    # Select model based on event type
    model_id = MODEL_MAPPING.get(event_type, "distilbert-base-uncased-finetuned-sst-2-english")
    
    # Prepare payload for Hugging Face model
    payload = {
        "inputs": event_content,
    }
    
    # For demo purposes, simulate API call if no key is provided
    if not API_KEY:
        # Simulate processing delay
        time.sleep(1)
        
        # Generate simulated AI insights based on event type
        if event_type == "user_feedback":
            ai_insights = [
                {"label": "POSITIVE", "score": 0.92},
                {"label": "NEGATIVE", "score": 0.08}
            ]
        elif event_type == "transaction":
            ai_insights = {
                "transaction_type": "purchase",
                "value_assessment": "high",
                "confidence": 0.87
            }
        elif event_type == "achievement":
            ai_insights = {
                "achievement_level": "significant",
                "percentile": 85,
                "engagement_impact": "positive"
            }
        else:
            ai_insights = {
                "classification": "routine",
                "priority": "low",
                "action_required": False
            }
    else:
        # Call Hugging Face Inference API with actual key
        try:
            headers = {"Authorization": f"Bearer {API_KEY}"}
            response = requests.post(
                f"{HF_ENDPOINT}{model_id}",
                headers=headers,
                json=payload,
                timeout=10
            )
            ai_insights = response.json()
        except Exception as e:
            ai_insights = {"error": str(e)}
    
    # Current timestamp for the demo
    current_time = datetime.now().isoformat()
    
    # Format response as it would appear in the SKENAI-RENO pipeline
    enriched_event = {
        "event_id": f"evt-{int(time.time())}-{hash(event_content) % 10000}",
        "type": event_type,
        "content": event_content,
        "user_id": user_id,
        "game_id": game_id,
        "timestamp": current_time,
        "ai_enrichment": {
            "model_id": model_id,
            "insights": ai_insights,
            "processed_timestamp": current_time
        },
        "processing": {
            "lambda_execution_time_ms": 120,
            "sqs_wait_time_ms": 15,
            "dynamodb_write_time_ms": 8
        }
    }
    
    # Add relevant additional fields based on event type
    if event_type == "transaction":
        enriched_event["transaction_details"] = {
            "amount": 4.99,
            "currency": "USD",
            "item_id": "coins_500",
            "payment_method": "app_store"
        }
    elif event_type == "achievement":
        enriched_event["achievement_details"] = {
            "achievement_id": "level_completion",
            "level": 10,
            "score": 95,
            "completion_time_sec": 320
        }
    
    return json.dumps(enriched_event, indent=2)

def generate_metrics_visualization():
    """Generate a sample visualization of event processing metrics."""
    # Create sample data for the past 7 days
    dates = [(datetime.now() - timedelta(days=i)).strftime('%Y-%m-%d') for i in range(7)]
    dates.reverse()
    
    # Sample event counts by type
    event_types = ["user_feedback", "transaction", "system_event", "achievement"]
    data = []
    
    for date in dates:
        for event_type in event_types:
            # Generate some random data with trends
            if event_type == "transaction":
                # Transactions increasing over time
                count = np.random.randint(800, 1200) + (dates.index(date) * 50)
            elif event_type == "user_feedback":
                # Feedback relatively stable
                count = np.random.randint(400, 600)
            elif event_type == "achievement":
                # Achievements with weekly pattern (higher on weekends)
                day_of_week = datetime.strptime(date, '%Y-%m-%d').weekday()
                weekend_boost = 200 if day_of_week >= 5 else 0
                count = np.random.randint(300, 500) + weekend_boost
            else:
                # System events decreasing (good sign, fewer errors)
                count = np.random.randint(100, 300) - (dates.index(date) * 10)
                count = max(count, 50)  # Ensure not negative
            
            data.append({
                "Date": date,
                "Event Type": event_type,
                "Count": count
            })
    
    # Create DataFrame
    df = pd.DataFrame(data)
    
    # Create plot
    fig = px.line(
        df, 
        x="Date", 
        y="Count", 
        color="Event Type",
        title="SKENAI-RENO: Event Processing Volume (Last 7 Days)",
        markers=True
    )
    
    fig.update_layout(
        xaxis_title="Date",
        yaxis_title="Number of Events",
        legend_title="Event Type",
        template="plotly_white"
    )
    
    return fig

def create_processing_pipeline_visualization():
    """Create a visual representation of the SKENAI-RENO event processing pipeline."""
    # Sample data for pipeline stages and their performance metrics
    stages = [
        "Redpanda", 
        "Lambda Connector", 
        "SQS Queue", 
        "AI Processor Lambda", 
        "DynamoDB"
    ]
    
    avg_latency = [5, 25, 15, 120, 8]
    error_rate = [0.01, 0.05, 0.001, 0.2, 0.02]
    
    # Create DataFrame
    df = pd.DataFrame({
        "Stage": stages,
        "Average Latency (ms)": avg_latency,
        "Error Rate (%)": [e * 100 for e in error_rate]
    })
    
    # Create bar chart for latency
    fig_latency = px.bar(
        df,
        x="Stage",
        y="Average Latency (ms)",
        title="Pipeline Stage Latency",
        color="Average Latency (ms)",
        color_continuous_scale="Viridis"
    )
    
    # Create bar chart for error rates
    fig_errors = px.bar(
        df,
        x="Stage",
        y="Error Rate (%)",
        title="Pipeline Stage Error Rates",
        color="Error Rate (%)",
        color_continuous_scale="Reds"
    )
    
    return fig_latency, fig_errors

# Define the Gradio interface
with gr.Blocks(theme=gr.themes.Soft(primary_hue="indigo")) as demo:
    gr.Markdown("# SKENAI-RENO Event Processing Demo")
    gr.Markdown("""
    Experience the power of AI-enriched event processing with SKENAI-RENO's serverless pipeline.
    This demo simulates how events from gaming and enterprise applications are processed 
    and enriched with AI insights in real-time.
    """)
    
    with gr.Tabs():
        with gr.TabItem("Event Processing Simulator"):
            with gr.Row():
                with gr.Column(scale=2):
                    gr.Markdown("### Input Event Details")
                    with gr.Row():
                        event_type = gr.Dropdown(
                            choices=["user_feedback", "transaction", "system_event", "achievement"],
                            value="user_feedback",
                            label="Event Type"
                        )
                    with gr.Row():
                        user_id = gr.Textbox(value="user123", label="User ID")
                        game_id = gr.Textbox(value="game456", label="Game/App ID")
                    with gr.Row():
                        event_content = gr.Textbox(
                            lines=5,
                            placeholder="Enter event content here...",
                            label="Event Content"
                        )
                    with gr.Row():
                        submit_btn = gr.Button("Process Event", variant="primary")
                
                with gr.Column(scale=3):
                    gr.Markdown("### Enriched Event Output")
                    output = gr.JSON()
            
            with gr.Row():
                gr.Markdown("### Example Events")
            
            with gr.Row():
                example1 = gr.Button("User Feedback Example")
                example2 = gr.Button("Transaction Example")
                example3 = gr.Button("Achievement Example")
            
            # Set up example event handlers
            example1.click(
                lambda: ["user_feedback", "user123", "game456", "I love this game! The new level design is amazing."],
                outputs=[event_type, user_id, game_id, event_content]
            )
            
            example2.click(
                lambda: ["transaction", "user789", "game456", "User purchased 500 coins for $4.99"],
                outputs=[event_type, user_id, game_id, event_content]
            )
            
            example3.click(
                lambda: ["achievement", "user456", "game789", "User completed level 10 with score 95/100"],
                outputs=[event_type, user_id, game_id, event_content]
            )
        
        with gr.TabItem("System Analytics"):
            gr.Markdown("### SKENAI-RENO Event Processing Analytics")
            gr.Markdown("""
            This dashboard shows sample analytics of the SKENAI-RENO event processing pipeline.
            In a production environment, these metrics would be generated from real-time CloudWatch data.
            """)
            
            with gr.Row():
                metrics_plot = gr.Plot()
            
            with gr.Row():
                with gr.Column():
                    latency_plot = gr.Plot()
                with gr.Column():
                    error_plot = gr.Plot()
            
            refresh_btn = gr.Button("Refresh Analytics")
        
        with gr.TabItem("About SKENAI-RENO"):
            gr.Markdown("""
            # About SKENAI-RENO
            
            SKENAI-RENO is a fully serverless event processing pipeline designed for high-reliability, 
            AI-enhanced insights, and seamless integration with existing systems.
            
            ## Key Features
            
            * **Serverless Architecture**: AWS Lambda, SQS, DynamoDB
            * **AI Enrichment**: Hugging Face models integrated directly into the event pipeline
            * **High Reliability**: Idempotent processing, dead-letter queues, comprehensive monitoring
            * **Enterprise Ready**: Secure, scalable, and compliant
            
            ## Technical Architecture
            
            SKENAI-RENO uses a multi-stage serverless architecture:
            
            1. **Event Sources**: Capture events from various sources (gaming, web, mobile, IoT)
            2. **Redpanda**: High-performance event streaming platform
            3. **Lambda Connector**: Processes events from Redpanda and sends to SQS
            4. **SQS Queue**: Decouples components and provides buffering
            5. **AI Processor Lambda**: Enriches events with AI insights using Hugging Face models
            6. **DynamoDB**: Stores processed events for analysis and retrieval
            
            ## Integration Options
            
            * REST API for direct event submission
            * SDK libraries for major platforms
            * Webhook integrations
            * Custom connectors for enterprise systems
            
            ## Learn More
            
            For enterprise solutions and partnership opportunities, visit [skenai-reno.com](https://skenai-reno.com)
            """)
    
    # Set up event handlers
    submit_btn.click(
        process_event,
        inputs=[event_type, event_content, user_id, game_id],
        outputs=output
    )
    
    # Initialize and refresh the analytics
    refresh_btn.click(
        lambda: [generate_metrics_visualization(), *create_processing_pipeline_visualization()],
        outputs=[metrics_plot, latency_plot, error_plot]
    )
    
    # Initialize the analytics on page load
    demo.load(
        lambda: [generate_metrics_visualization(), *create_processing_pipeline_visualization()],
        outputs=[metrics_plot, latency_plot, error_plot]
    )

# Launch the app
if __name__ == "__main__":
    demo.launch()