Create app.py

app.py

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