app.py

import time
import gradio as gr
import networkx as nx
import matplotlib.pyplot as plt

# ---- Define 20 Steps ----
pipeline_steps = [
    "ML Model File Loaded",
    "ELT Data Set Loaded",
    "Code Pushed to Repo",
    "Pull Request Created",
    "Unit Tests Executed",
    "Integration Tests Executed",
    "Data Validation",
    "Security Scan",
    "Environment Build",
    "Artifact Stored",
    "Model Training Pipeline Run",
    "Model Validation (Accuracy, Recall)",
    "Performance Test (Latency)",
    "Bias & Ethics Check",
    "Test Report Generated",
    "Deploy to Staging",
    "Shadow Deployment",
    "Deploy to Production",
    "Monitoring & Drift Detection",
    "Feedback Loop & Retraining Trigger"
]

# ---- Status for Demo ----
demo_status = [
    "✅ ML Model File (demo.py) loaded",
    "✅ ELT Data Set (sample.csv) loaded",
    "✅ Code pushed to GitHub",
    "✅ Pull request created & approved",
    "✅ Unit tests passed (12/12)",
    "✅ Integration tests passed",
    "✅ Data validation successful",
    "✅ Security scan passed",
    "✅ Environment build successful",
    "✅ Artifact stored in registry",
    "✅ Model trained (epoch=5)",
    "✅ Validation passed (Accuracy=92%)",
    "✅ Latency test passed (50ms avg)",
    "⚠️ Bias check warning (dataset imbalance found)",
    "✅ Test report generated",
    "✅ Staging deployment successful",
    "✅ Shadow deployment running",
    "✅ Production deployment successful",
    "✅ Monitoring enabled (no drift detected)",
    "🚀 Feedback loop active – system ready"
]

# ---- Function to Simulate ----
def run_pipeline(code_input, dataset_input):
    logs = []
    
    # If no input, run demo
    if not code_input:
        code_input = "ML Model File: demo.py"
    if not dataset_input:
        dataset_input = "ELT Data Set: sample.csv"
    
    logs.append(f"Pipeline started with {code_input} and {dataset_input}.\n")
    
    # Simulate pipeline execution
    for i, step in enumerate(pipeline_steps):
        logs.append(f"Step {i+1}: {demo_status[i]}")
        time.sleep(0.3)  # simulate delay
    
    # ---- Create Graph ----
    G = nx.DiGraph()
    for i, step in enumerate(pipeline_steps):
        G.add_node(f"{i+1}")
        if i > 0:
            G.add_edge(f"{i}", f"{i+1}")

    plt.figure(figsize=(12, 6))
    pos = nx.spring_layout(G, seed=42)
    nx.draw(G, pos, with_labels=True, node_size=1200, node_color="lightgreen", font_size=8, font_weight="bold")
    nx.draw_networkx_labels(G, pos, labels={str(i+1): f"{i+1}" for i in range(len(pipeline_steps))})
    
    plt.title("CI/CD/CT 20-Step Pipeline (Simulator)")
    plt.tight_layout()
    plt.savefig("pipeline.png")
    plt.close()

    return "pipeline.png", "\n".join(logs)


# ---- Gradio UI ----
with gr.Blocks() as demo:
    gr.Markdown("# 🔄 CI/CD/CT Pipeline Simulator (Demo Version)")
    gr.Markdown("This app simulates a **20-step CI/CD/CT pipeline**. Upload your own files or leave blank to run the prebuilt demo.")
    
    with gr.Row():
        code_input = gr.Textbox(label="Upload/Enter ML Model File", placeholder="Leave empty to use demo.py")
        dataset_input = gr.Textbox(label="Upload/Enter ELT Data Set", placeholder="Leave empty to use sample.csv")
    
    run_button = gr.Button("▶️ Run 20-Step Pipeline")
    
    with gr.Row():
        graph_output = gr.Image(label="Pipeline Graph")
        log_output = gr.Textbox(label="Pipeline Logs", lines=25)
    
    run_button.click(run_pipeline, [code_input, dataset_input], [graph_output, log_output])

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