app.py

import gradio as gr
import pandas as pd
import numpy as np
import plotly.express as px
from datetime import datetime, timedelta
from scipy import stats

def generate_synthetic_data(n=800):
    np.random.seed(42)
    start = datetime(2025, 1, 1, 0, 0)
    ts = [start + timedelta(minutes=5 * i) for i in range(n)]
    
    signal_dbm = np.random.normal(-82, 4, n)
    snr_db = np.random.normal(24, 3, n)
    ber = np.random.lognormal(-6.5, 0.6, n)  # ~ 10^-6 to 10^-4
    packet_loss = np.random.uniform(0, 8, n)
    
    # Inject realistic interference (2 events)
    signal_dbm[180:230] -= 22          # deep fade / jamming
    snr_db[180:230] -= 18
    packet_loss[180:230] = np.random.uniform(25, 60, 50)
    
    signal_dbm[520:560] -= 12          # mild interference
    snr_db[520:560] -= 9
    packet_loss[520:560] = np.random.uniform(12, 30, 40)
    
    df = pd.DataFrame({
        "Timestamp": ts,
        "Signal_Strength_dBm": signal_dbm,
        "SNR_dB": snr_db,
        "BER": ber,
        "Packet_Loss_%": packet_loss
    })
    return df

def detect_anomalies_and_alerts(df):
    df = df.copy()
    # Z-score on SNR and Signal
    df["SNR_z"] = np.abs(stats.zscore(df["SNR_dB"], nan_policy="omit"))
    df["Signal_z"] = np.abs(stats.zscore(df["Signal_Strength_dBm"], nan_policy="omit"))
    df["is_anomaly"] = (df["SNR_z"] > 3) | (df["Signal_z"] > 3) | (df["Packet_Loss_%"] > 20)
    
    # Sudden drops (interference)
    df["Signal_Diff"] = df["Signal_Strength_dBm"].diff()
    sudden_drops = df[df["Signal_Diff"] < -12]
    
    alerts = []
    if len(sudden_drops) > 0:
        alerts.append(f"🚨 {len(sudden_drops)} sudden signal drops (>12 dB) detected – possible interference/jamming.")
    if (df["SNR_dB"] < 12).mean() > 0.1:
        alerts.append("⚠️ More than 10% of time SNR < 12 dB – link degradation likely.")
    if df["Packet_Loss_%"].mean() > 15:
        alerts.append("⚠️ High average packet loss – interference or rain fade suspected.")
    if len(alerts) == 0:
        alerts.append("✅ No major interference or anomalies detected.")
    
    return df, "\n".join(alerts)

def calculate_risk_score(df):
    score = 0
    # SNR penalties
    avg_snr = df["SNR_dB"].mean()
    low_snr_pct = (df["SNR_dB"] < 15).mean() * 100
    score += max(0, 40 - avg_snr) * 1.5
    score += low_snr_pct * 0.8
    
    # Signal & loss
    if df["Signal_Strength_dBm"].min() < -105:
        score += 25
    score += df["Packet_Loss_%"].mean() * 1.2
    
    # Anomaly count
    anomaly_pct = df["is_anomaly"].mean() * 100 if "is_anomaly" in df.columns else 0
    score += anomaly_pct * 0.7
    
    risk = min(100, max(0, int(score)))
    level = "Low 🟢" if risk < 35 else "Medium 🟡" if risk < 70 else "High 🔴"
    return f"**Risk Score: {risk}/100 – {level}**", risk

def create_plots(df):
    fig_snr = px.line(df, x="Timestamp", y="SNR_dB", title="SNR Trend (dB)", markers=False)
    fig_snr.update_traces(line_color="#00ff88")
    
    fig_signal = px.line(df, x="Timestamp", y="Signal_Strength_dBm", title="Signal Strength (dBm)", markers=False)
    fig_signal.update_traces(line_color="#ffaa00")
    
    # Highlight anomalies
    if "is_anomaly" in df.columns:
        anomalies = df[df["is_anomaly"]]
        if not anomalies.empty:
            fig_snr.add_scatter(x=anomalies["Timestamp"], y=anomalies["SNR_dB"],
                                mode="markers", marker=dict(color="red", size=8, symbol="x"),
                                name="Anomaly / Interference")
            fig_signal.add_scatter(x=anomalies["Timestamp"], y=anomalies["Signal_Strength_dBm"],
                                   mode="markers", marker=dict(color="red", size=8, symbol="x"),
                                   name="Anomaly / Interference")
    
    return fig_snr, fig_signal

def analyze(file):
    if file is None:
        df = generate_synthetic_data()
        demo_note = "🔬 Using built-in synthetic dataset with injected interference events (for demo)."
    else:
        df = pd.read_csv(file)
        demo_note = "📁 Uploaded file loaded successfully."
    
    df["Timestamp"] = pd.to_datetime(df["Timestamp"], errors="coerce")
    df = df.dropna(subset=["Timestamp"]).sort_values("Timestamp").reset_index(drop=True)
    
    df, alerts = detect_anomalies_and_alerts(df)
    risk_text, risk_val = calculate_risk_score(df)
    fig_snr, fig_signal = create_plots(df)
    
    stats = df.describe().round(3).to_html()
    
    preview = df.head(15).to_html(index=False)
    
    return (
        preview,
        fig_snr,
        fig_signal,
        risk_text,
        alerts,
        stats,
        demo_note
    )

# ==================== Gradio UI ====================
with gr.Blocks(title="Satellite Signal Log Analyzer", theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 🛰️ Satellite Signal Log Analyzer\n**Risk scoring • Interference alerts • Trend visualization**\nUpload your CSV or try the demo.")
    
    with gr.Row():
        with gr.Column(scale=1):
            file_input = gr.File(label="Upload CSV Log (or leave empty for synthetic demo)", file_types=[".csv"])
            analyze_btn = gr.Button("🚀 Analyze", variant="primary", size="large")
        
        with gr.Column(scale=1):
            sample_btn = gr.Button("📥 Load Synthetic Demo Data", variant="secondary")
    
    gr.Markdown("**Expected columns**: `Timestamp`, `Signal_Strength_dBm`, `SNR_dB` (optional: `BER`, `Packet_Loss_%`).")
    
    with gr.Tabs():
        with gr.Tab("📊 Overview"):
            gr.Markdown("### Data Preview")
            preview_html = gr.HTML()
            gr.Markdown("### Summary Statistics")
            stats_html = gr.HTML()
        
        with gr.Tab("📈 Visualizations"):
            with gr.Row():
                snr_plot = gr.Plot(label="SNR Trend")
                signal_plot = gr.Plot(label="Signal Strength Trend")
        
        with gr.Tab("⚠️ Alerts & Risk"):
            risk_output = gr.Markdown()
            alerts_output = gr.Markdown()
        
        with gr.Tab("ℹ️ Info"):
            gr.Markdown("""
            **How the risk score works**  
            • Weighted penalties for low SNR, deep fades, high packet loss, and statistical outliers (Z-score > 3).  
            • Interference alerts use sudden drops (>12 dB) and threshold crossings.  
            **Public datasets** to test with:  
            • OPSSAT-AD (Kaggle) – real CubeSat telemetry with anomalies  
            • LENS (Zenodo) – Starlink/LEO network measurements  
            • GOCE Telemetry (Kaggle)
            """)
    
    gr.Markdown("### Demo Note")
    demo_note_out = gr.Markdown()

    # Button actions
    analyze_btn.click(
        fn=analyze,
        inputs=file_input,
        outputs=[preview_html, snr_plot, signal_plot, risk_output, alerts_output, stats_html, demo_note_out]
    )
    
    sample_btn.click(
        fn=lambda: (None, None, None, None, None, None, None),  # clear file
        outputs=[file_input, preview_html, snr_plot, signal_plot, risk_output, alerts_output, stats_html]
    ).then(
        fn=analyze,
        inputs=[file_input],  # still None → uses synthetic
        outputs=[preview_html, snr_plot, signal_plot, risk_output, alerts_output, stats_html, demo_note_out]
    )

if __name__ == "__main__":
    demo.launch()