app.py

import requests
import gradio as gr
import os
import random
from datetime import datetime, timedelta
import plotly.express as px
import re

api_key = os.getenv("CEREBRAS_API_KEY")
url = "https://api.cerebras.ai/v1/chat/completions"
headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}

SAMPLE_LOG = """
2025-04-12 10:01 UTC | 40.7N, 74.0W | Frequency: 14.5 GHz | Signal Strength: 85% | Message: Routine check, systems OK.
2025-04-12 10:02 UTC | 40.8N, 74.1W | Frequency: 14.7 GHz | Signal Strength: 60% | Message: Noise detected, possible interference.
2025-04-12 10:03 UTC | 40.9N, 74.2W | Frequency: 14.6 GHz | Signal Strength: 90% | Message: Emergency: Low battery alert.
"""

LOG_HISTORY = []

def generate_random_log():
    base_time = datetime(2025, 4, 12, 10, 0)
    entries = []
    for i in range(3):
        time = (base_time + timedelta(minutes=i)).strftime("%Y-%m-%d %H:%M UTC")
        lat = round(random.uniform(40.0, 41.0), 1)
        lon = round(random.uniform(73.0, 74.0), 1)
        freq = round(random.uniform(14.0, 15.0), 1)
        signal = random.randint(50, 100)
        messages = [
            "Routine check, systems OK.",
            "Noise detected, possible interference.",
            "Emergency: Low battery alert.",
            "Signal stable, no issues.",
            "Warning: Solar flare detected."
        ]
        message = random.choice(messages)
        entries.append(f"{time} | {lat}N, {lon}W | Frequency: {freq} GHz | Signal Strength: {signal}% | Message: {message}")
    return "\n".join(entries)

def analyze_log(log_text):
    if not log_text.strip():
        return "Error: Please enter a log.", None, None
    LOG_HISTORY.append(log_text)
    data = {
        "model": "llama-4-scout-17b-16e-instruct",
        "messages": [
            {
                "role": "user",
                "content": "Analyze this satellite radio log and summarize in bullet points. Include interference risk scores (0-100, low signal <70% = high risk >80, emergency = 90). Ensure frequencies in issues and details:\n- Issues (e.g., low signal, noise, interference with frequency, risk score)\n- High-priority messages (e.g., emergencies, warnings)\n- Key details (coordinates, times, frequencies, signal strengths)\nLog:\n" + log_text
            }
        ],
        "max_completion_tokens": 500,
        "temperature": 0.5
    }
    try:
        response = requests.post(url, json=data, headers=headers)
        summary = response.json()["choices"][0]["message"]["content"]
        html = "<div style='font-family:Roboto; color:black; background:white; padding:10px; border-radius:5px; border:1px solid #ff6200;'><h3>Analysis</h3><ul>"
        for line in summary.split("\n"):
            if "Issues:" in line:
                html += "<li><b style='color:#ff6200;'>Issues:</b><ul>"
            elif "High-priority messages:" in line:
                html += "</ul><li><b style='color:#ff6200;'>Priority Alerts:</b><ul>"
            elif "Key details:" in line:
                html += "</ul><li><b style='color:#ff6200;'>Details:</b><ul>"
            elif line.strip():
                html += f"<li>{line.strip()}</li>"
        html += "</ul></div>"
        signals = []
        times = []
        for line in log_text.split("\n"):
            if "Signal Strength" in line:
                match = re.search(r"Signal Strength: (\d+)%", line)
                if match:
                    signals.append(int(match.group(1)))
                time_match = re.search(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}) UTC", line)
                if time_match:
                    times.append(time_match.group(1)[-5:])
        fig = px.line(x=times, y=signals, labels={"x": "Time", "y": "Signal (%)"}, title="Signal Trend") if signals and len(signals) == len(times) else None
        return summary, html, fig
    except Exception as e:
        return f"Error: API call failed - {str(e)}", None, None

def generate_alert(log_text):
    if not log_text.strip():
        return "Error: Please enter a log."
    data = {
        "model": "llama-4-scout-17b-16e-instruct",
        "messages": [
            {
                "role": "user",
                "content": "Generate an urgent satellite alert based on this log’s conditions (e.g., interference for noise/low signal, escalation for emergencies). Include frequency, time, coordinates:\nLog:\n" + log_text
            }
        ],
        "max_completion_tokens": 100,
        "temperature": 0.7
    }
    try:
        response = requests.post(url, json=data, headers=headers)
        alert = response.json()["choices"][0]["message"]["content"]
        return f"<div style='font-family:Roboto; background:#ff6200; color:white; padding:10px; border-radius:5px;'>{alert}</div>"
    except Exception as e:
        return f"Error: Alert failed - {str(e)}"

def compare_logs():
    if len(LOG_HISTORY) < 2:
        return "Error: Need at least two logs.", None
    compare_text = "Previous log:\n" + LOG_HISTORY[-2] + "\nCurrent log:\n" + LOG_HISTORY[-1]
    data = {
        "model": "llama-4-scout-17b-16e-instruct",
        "messages": [
            {
                "role": "user",
                "content": "Compare these two satellite radio logs and summarize trends in bullet points (e.g., signal strength changes, frequency issues, new emergencies):\n" + compare_text
            }
        ],
        "max_completion_tokens": 400,
        "temperature": 0.5
    }
    try:
        response = requests.post(url, json=data, headers=headers)
        comparison = response.json()["choices"][0]["message"]["content"]
        html = "<div style='font-family:Roboto; color:black; background:white; padding:10px; border-radius:5px; border:1px solid #ff6200;'><h3>Comparison</h3><ul>"
        for line in comparison.split("\n"):
            if line.strip():
                html += f"<li>{line.strip()}</li>"
        html += "</ul></div>"
        return comparison, html
    except Exception as e:
        return f"Error: Comparison failed - {str(e)}", None

def load_sample_log():
    return SAMPLE_LOG

def clear_log():
    return ""

css = """
body, .gradio-container { background: white; color: black; font-family: Roboto, sans-serif; }
button { background: #ff6200; color: white; border: none; padding: 8px 16px; border-radius: 5px; }
button:hover { background: #e55a00; }
.input-text, .output-text { background: white; color: black; border: 1px solid #ff6200; border-radius: 5px; }
h3 { color: #ff6200; }
.header, .subheader { color: #ff6200; text-align: center; }
"""
with gr.Blocks(css=css) as interface:
    gr.Markdown("# Satellite Signal Log Analyzer", elem_classes="header")
    gr.Markdown("Analyze logs for issues, alerts, and trends.", elem_classes="subheader")
    log_input = gr.Textbox(lines=5, show_label=False, placeholder="Enter or generate a log...")
    with gr.Row():
        sample_button = gr.Button("Sample Log")
        random_button = gr.Button("Random Log")
        clear_button = gr.Button("Clear")
    with gr.Row():
        analyze_button = gr.Button("Analyze")
        alert_button = gr.Button("Alert")
        compare_button = gr.Button("Compare Logs")
    output = gr.HTML(show_label=False)
    plot_output = gr.Plot(show_label=False)
    alert_output = gr.HTML(show_label=False)
    compare_output = gr.HTML(show_label=False)
    sample_button.click(fn=load_sample_log, outputs=log_input)
    random_button.click(fn=generate_random_log, outputs=log_input)
    clear_button.click(fn=clear_log, outputs=log_input)
    analyze_button.click(fn=analyze_log, inputs=log_input, outputs=[output, output, plot_output])
    alert_button.click(fn=generate_alert, inputs=log_input, outputs=alert_output)
    compare_button.click(fn=compare_logs, outputs=[compare_output, compare_output])

interface.launch()