#!/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()