app.py

import gradio as gr
import re

# ============================================
# KPI Extraction
# ============================================

def extract_value(text, key):
    pattern = rf"{key}\s*=\s*(-?\d+\.?\d*)"
    match = re.search(pattern, text, re.IGNORECASE)

    if match:
        return float(match.group(1))

    return None

# ============================================
# Main Analysis Logic
# ============================================

def analyze_ul_interference(vendor, data):

    findings = []
    causes = []
    actions = []

    severity = "LOW"
    condition = "NORMAL"

    ul_sinr = extract_value(data, "UL SINR")
    rtwp = extract_value(data, "RTWP")
    ul_prb = extract_value(data, "UL PRB")
    dl_prb = extract_value(data, "DL PRB")
    bler = extract_value(data, "PUSCH BLER")
    cqi = extract_value(data, "CQI")

    # ============================================
    # UL SINR
    # ============================================

    if ul_sinr is not None:

        if ul_sinr < -5:
            findings.append(f"Critical UL SINR degradation ({ul_sinr} dB)")
            severity = "HIGH"

        elif ul_sinr < 0:
            findings.append(f"Poor UL SINR detected ({ul_sinr} dB)")
            severity = "MEDIUM"

    # ============================================
    # RTWP
    # ============================================

    if rtwp is not None:

        if rtwp > -95:
            findings.append(f"Severe RTWP elevation ({rtwp} dBm)")
            severity = "HIGH"

        elif rtwp > -100:
            findings.append(f"Elevated RTWP/noise floor ({rtwp} dBm)")
            severity = "MEDIUM"

    # ============================================
    # BLER
    # ============================================

    if bler is not None:

        if bler > 30:
            findings.append(f"Critical PUSCH BLER ({bler}%)")
            severity = "HIGH"

        elif bler > 15:
            findings.append(f"Elevated PUSCH BLER ({bler}%)")

    # ============================================
    # CQI
    # ============================================

    if cqi is not None:

        if cqi < 5:
            findings.append(f"Low CQI detected ({cqi})")

    # ============================================
    # PRB
    # ============================================

    congestion = False
    interference = False

    if ul_prb is not None:

        if ul_prb > 85:
            findings.append(f"High UL PRB utilization ({ul_prb}%)")
            congestion = True

    # ============================================
    # Classification Logic
    # ============================================

    if (
        ul_sinr is not None and ul_sinr < 0 and
        rtwp is not None and rtwp > -100 and
        bler is not None and bler > 15
    ):
        interference = True

    # ============================================
    # Interference Condition
    # ============================================

    if interference:

        condition = "LIKELY UL INTERFERENCE"

        causes = [
            "External uplink interference",
            "Passive Intermodulation (PIM)",
            "Overshooting sector",
            "Feeder or jumper degradation",
            "Radio hardware impairment"
        ]

        actions = [
            "Review RTWP trend history",
            "Compare neighboring sector RTWP",
            "Inspect feeder and jumper path",
            "Validate antenna alignment",
            "Perform spectrum sweep",
            "Review recent maintenance activity"
        ]

    # ============================================
    # Congestion Condition
    # ============================================

    elif congestion:

        condition = "LIKELY CAPACITY CONGESTION"

        causes = [
            "High uplink traffic demand",
            "Sector loading imbalance",
            "Subscriber concentration",
            "Limited uplink scheduling capacity"
        ]

        actions = [
            "Review busy-hour utilization",
            "Evaluate load balancing",
            "Assess scheduler efficiency",
            "Review carrier utilization"
        ]

    # ============================================
    # Normal Condition
    # ============================================

    else:

        condition = "NO MAJOR UL IMPAIRMENT DETECTED"

        actions = [
            "Continue KPI monitoring"
        ]

    # ============================================
    # Markdown Formatting
    # ============================================

    findings_md = "### Findings\n\n" + "\n".join(
        [f"- {x}" for x in findings]
    )

    causes_md = "### Likely Causes\n\n" + "\n".join(
        [f"- {x}" for x in causes]
    )

    actions_md = "### Recommended Actions\n\n" + "\n".join(
        [f"- {x}" for x in actions]
    )

    assessment_md = """
### Engineering Assessment

Analysis completed using LTE/5G uplink KPI and PM counter thresholds.
"""

    return (
        condition,
        severity,
        vendor,
        findings_md,
        causes_md,
        actions_md,
        assessment_md
    )

# ============================================
# Example Inputs
# ============================================

examples = [
    [
        "Ericsson",
        """UL SINR = -8
PUSCH BLER = 41
RTWP = -94
UL PRB = 72
DL PRB = 35
CQI = 3"""
    ],
    [
        "Nokia",
        """UL SINR = 4
PUSCH BLER = 5
RTWP = -108
UL PRB = 96
DL PRB = 92
CQI = 11"""
    ],
    [
        "Samsung",
        """UL SINR = -2
PUSCH BLER = 18
RTWP = -99
UL PRB = 88
DL PRB = 81
CQI = 6"""
    ]
]

# ============================================
# UI
# ============================================

with gr.Blocks(theme=gr.themes.Default()) as demo:

    gr.Markdown("""
# UL Interference Analyzer

AI-assisted LTE and 5G uplink interference analysis using KPI and PM counter indicators.
""")

    vendor = gr.Dropdown(
        choices=["Ericsson", "Nokia", "Samsung"],
        value="Ericsson",
        label="RAN Vendor",
        container=False
    )

    data_input = gr.Textbox(
        lines=10,
        label="Paste KPI / PM Counter Data"
    )

    analyze_btn = gr.Button(
        "Analyze",
        variant="primary"
    )

    # ============================================
    # Top Status Row
    # ============================================

    with gr.Row():

        condition_box = gr.Textbox(
            label="Condition",
            interactive=False
        )

        severity_box = gr.Textbox(
            label="Severity",
            interactive=False
        )

        vendor_box = gr.Textbox(
            label="Vendor",
            interactive=False
        )

    # ============================================
    # Main Analysis Panels
    # ============================================

    with gr.Row():

        findings_box = gr.Markdown()

        causes_box = gr.Markdown()

        actions_box = gr.Markdown()

    assessment_box = gr.Markdown()

    # ============================================
    # Examples
    # ============================================

    gr.Examples(
        examples=examples,
        inputs=[vendor, data_input]
    )

    # ============================================
    # Button Action
    # ============================================

    analyze_btn.click(
        fn=analyze_ul_interference,
        inputs=[vendor, data_input],
        outputs=[
            condition_box,
            severity_box,
            vendor_box,
            findings_box,
            causes_box,
            actions_box,
            assessment_box
        ]
    )

demo.launch()