Create app.py

app.py

@@ -0,0 +1,190 @@
+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()