Update 2026-03-26 17:41:53

app.py

@@ -30,6 +30,17 @@ def _mlflow_client():
 
 # ── Seed demo data on first launch ────────────────────────────────────────────
 
+def _warm_imports():
+    """Pre-import heavy ML libraries so the first pipeline run is instant."""
+    try:
+        import sklearn, sklearn.ensemble, sklearn.preprocessing  # noqa: F401
+        import mlflow, mlflow.sklearn                            # noqa: F401
+        from mlops.datasets import load_dataset
+        load_dataset("Iris Flowers")   # primes sklearn's data cache
+    except Exception:
+        pass
+
+
 def _seed_demo():
     """Pre-populate a few MLflow runs so the dashboard looks great immediately."""
     client = _mlflow_client()
@@ -77,7 +88,8 @@ def _seed_demo():
             pass
 
 
-# Seed in background so startup isn't delayed
+# Warm imports and seed demo data in background so startup isn't delayed
+threading.Thread(target=_warm_imports, daemon=True).start()
 threading.Thread(target=_seed_demo, daemon=True).start()
 
 
@@ -224,16 +236,15 @@ def api_pipeline_execute(pipeline_id):
     except ValueError as e:
         return jsonify({"error": str(e)}), 400
 
-    # Use Apache Airflow. Falls back to the built-in engine only if Airflow
-    # is not importable (i.e. not installed at all — should not happen in prod).
-    try:
-        from mlops.airflow_runner import trigger_pipeline
-        exec_id = trigger_pipeline(pipeline_id, context=context, dag=dag)
-        return jsonify({"exec_id": exec_id, "status": "queued", "engine": "airflow"})
-    except ImportError:
-        app.logger.warning("Airflow not installed — using built-in DAG engine")
-    except Exception as af_err:
-        app.logger.warning(f"Airflow trigger failed, falling back to built-in engine: {af_err}")
+    # Built-in engine is the default — zero scheduler latency, runs immediately.
+    # Set USE_AIRFLOW=true in the environment to hand off to Apache Airflow instead.
+    if os.environ.get("USE_AIRFLOW", "").lower() in ("1", "true"):
+        try:
+            from mlops.airflow_runner import trigger_pipeline
+            exec_id = trigger_pipeline(pipeline_id, context=context, dag=dag)
+            return jsonify({"exec_id": exec_id, "status": "queued", "engine": "airflow"})
+        except Exception as af_err:
+            app.logger.warning(f"Airflow trigger failed, falling back to built-in engine: {af_err}")
 
     exec_id = execute_dag(dag, context)
     return jsonify({"exec_id": exec_id, "status": "queued", "engine": "builtin"})

pipelines/dag_engine.py

@@ -141,7 +141,7 @@ def _run_dag(exec_id: str, dag: DAG, context: dict):
             progress = int(100 * (step_idx + 1) / total)
             _upd(progress=progress)
 
-            time.sleep(0.4)   # small delay so the UI can animate
+            time.sleep(0.1)   # small delay so the UI can animate
 
         _upd(status="completed", progress=100,
              completed_at=datetime.utcnow().isoformat())

pipelines/pipeline_defs.py

@@ -1,5 +1,4 @@
 """Pre-built ML pipeline DAG definitions."""
-import time
 import numpy as np
 from pipelines.dag_engine import DAG, Task
 
@@ -18,21 +17,18 @@ def _load_data(ctx, _results):
 def _validate_data(ctx, results):
     log = ctx.get("_log")
     if log: log("Checking schema, nulls, and feature ranges…")
-    time.sleep(0.2)
     if log: log("No nulls found · All feature ranges valid")
     return "Schema OK · No nulls detected · Feature ranges valid"
 
 def _preprocess(ctx, results):
     log = ctx.get("_log")
     if log: log("Fitting StandardScaler on training split…")
-    time.sleep(0.3)
     if log: log("80/20 stratified train/test split applied")
     return "StandardScaler fitted · Train/test split 80/20"
 
 def _feature_engineering(ctx, results):
     log = ctx.get("_log")
     if log: log("Evaluating polynomial and interaction features…")
-    time.sleep(0.2)
     if log: log("No additional features needed · all originals retained")
     return "Polynomial features skipped · All features retained"
 
@@ -65,79 +61,64 @@ def _train_model(ctx, results):
 def _evaluate_model(ctx, results):
     log = ctx.get("_log")
     if log: log("Computing accuracy / R² on hold-out set…")
-    time.sleep(0.2)
     if log: log("5-fold cross-validation passed")
     return "Accuracy / R² computed · Cross-val 5-fold done"
 
 def _generate_report(ctx, results):
     log = ctx.get("_log")
     if log: log("Writing evaluation artefacts to MLflow…")
-    time.sleep(0.15)
     return "HTML report generated · Metrics written to mlflow"
 
 def _register_model(ctx, _results):
     log = ctx.get("_log")
     if log: log("Pushing model artifact to MLflow Model Registry…")
-    time.sleep(0.1)
     return "Model artifact registered in MLflow Model Registry"
 
 def _deploy_staging(ctx, _results):
     log = ctx.get("_log")
     if log: log("Transitioning model version to Staging…")
-    time.sleep(0.2)
     if log: log("REST endpoint ready")
     return "Model transitioned to Staging · REST endpoint ready"
 
 # ── Retraining pipeline tasks ──────────────────────────────────────────────────
 
 def _check_drift(ctx, _):
-    time.sleep(0.2)
     drift = round(np.random.uniform(0.01, 0.08), 4)
     return f"PSI={drift} · {'Drift detected — retraining triggered' if drift > 0.05 else 'No drift · pipeline skipped'}"
 
 def _fetch_new_data(ctx, _):
-    time.sleep(0.3)
     n = np.random.randint(200, 800)
     return f"{n} new labelled samples fetched from data store"
 
 def _merge_datasets(ctx, _):
-    time.sleep(0.2)
     return "New data merged with historical · duplicates removed"
 
 def _retrain_champion(ctx, _):
-    time.sleep(0.4)
     acc = round(np.random.uniform(0.88, 0.97), 4)
     return f"Champion model retrained · new accuracy={acc}"
 
 def _ab_test(ctx, _):
-    time.sleep(0.2)
     return "A/B test scheduled · 10% traffic split for 24 h"
 
 def _promote_production(ctx, _):
-    time.sleep(0.15)
     return "Champion model promoted to Production · old version archived"
 
 # ── Data pipeline tasks ────────────────────────────────────────────────────────
 
 def _ingest_raw(ctx, _):
-    time.sleep(0.2)
     return "Raw data ingested from source"
 
 def _clean_data(ctx, _):
-    time.sleep(0.3)
     removed = np.random.randint(5, 40)
     return f"{removed} anomalous rows removed · missing values imputed"
 
 def _encode_features(ctx, _):
-    time.sleep(0.2)
     return "Categorical features one-hot encoded · ordinals label-encoded"
 
 def _scale_features(ctx, _):
-    time.sleep(0.2)
     return "Numeric features scaled with StandardScaler"
 
 def _save_processed(ctx, _):
-    time.sleep(0.1)
     return "Processed dataset saved to feature store"
 
 

templates/pipeline.html

@@ -175,42 +175,6 @@
 .cfg-row-k { color: var(--text-muted); white-space: nowrap; padding-right: 8px; }
 .cfg-row-v { color: var(--text-primary); font-weight: 500; text-align: right; word-break: break-word; max-width: 62%; font-size: .77rem; }
 
-/* ── Steps progress bar ───────────────────────────────────────────────────── */
-.ps-steps {
-  flex-shrink: 0; display: none;
-  padding: 5px 16px; min-height: 38px;
-  background: var(--bg-secondary);
-  border-bottom: 1px solid var(--border-color);
-  overflow-x: auto; align-items: center; gap: 4px;
-  scrollbar-width: none;
-}
-.ps-steps::-webkit-scrollbar { display: none; }
-.ps-steps.visible { display: flex; }
-
-.step-pill {
-  display: inline-flex; align-items: center; gap: 5px;
-  padding: 3px 9px; border-radius: 20px; flex-shrink: 0;
-  font-size: .7rem; font-weight: 500; white-space: nowrap;
-  border: 1px solid var(--border-color);
-  background: var(--bg-tertiary); color: var(--text-secondary);
-  transition: background .2s, border-color .2s, color .2s;
-}
-.step-pill.s-running {
-  border-color: var(--warning); background: rgba(245,158,11,.12); color: var(--warning);
-  animation: pill-pulse 1.4s ease-in-out infinite;
-}
-.step-pill.s-success {
-  border-color: rgba(34,197,94,.35); background: rgba(34,197,94,.08); color: var(--success);
-}
-.step-pill.s-failed {
-  border-color: rgba(239,68,68,.35); background: rgba(239,68,68,.08); color: var(--danger);
-}
-.step-sep { color: var(--border-color); font-size: .65rem; flex-shrink: 0; user-select: none; }
-@keyframes pill-pulse {
-  0%,100% { box-shadow: none; }
-  50%      { box-shadow: 0 0 0 3px rgba(245,158,11,.15); }
-}
-
 /* ── Terminal ─────────────────────────────────────────────────────────────── */
 .ps-term {
   flex-shrink: 0; height: 34px; overflow: hidden;
@@ -275,9 +239,6 @@
     </button>
   </div>
 
-  <!-- ── Steps progress bar ────────────────────────────────────────────── -->
-  <div class="ps-steps" id="ps-steps"></div>
-
   <!-- ── Main area ──────────────────────────────────────────────────────── -->
   <div class="ps-main">
 
@@ -364,7 +325,6 @@ document.addEventListener('DOMContentLoaded', () =>
 function switchPipeline(id, btn) {
   if (pollIv) { clearInterval(pollIv); pollIv = null; }
   cur = id; execId = null; tstates = {}; selNode = null;
-  _hideSteps();
   closeConfig(false);
   document.querySelectorAll('.ps-tab').forEach(b => b.classList.remove('active'));
   btn.classList.add('active');
@@ -618,38 +578,6 @@ async function onTtChange(tt) {
 }
 function onCatChange(cat) { pCtx.category=cat; _fillAlgos(cat); }
 
-// ── Steps bar ─────────────────────────────────────────────────────────────────
-function _stepsOrder(dag) {
-  return Object.values(dag.tasks)
-    .sort((a, b) => a.layer !== b.layer ? a.layer - b.layer : a.task_id.localeCompare(b.task_id));
-}
-
-function _renderSteps(dag, tstates) {
-  const el = document.getElementById('ps-steps');
-  el.innerHTML = '';
-  _stepsOrder(dag).forEach((t, i) => {
-    if (i > 0) {
-      const sep = document.createElement('span');
-      sep.className = 'step-sep'; sep.textContent = '›';
-      el.appendChild(sep);
-    }
-    const st = (tstates[t.task_id] || {}).status || 'pending';
-    const pill = document.createElement('div');
-    pill.className = `step-pill s-${st}`;
-    pill.textContent = `${t.icon} ${t.name}`;
-    el.appendChild(pill);
-  });
-}
-
-function _showSteps(dag) {
-  _renderSteps(dag, {});
-  document.getElementById('ps-steps').classList.add('visible');
-}
-
-function _hideSteps() {
-  document.getElementById('ps-steps').classList.remove('visible');
-}
-
 // ── Execute pipeline ──────────────────────────────────────────────────────────
 async function runPipeline() {
   const runBtn = document.getElementById('ps-run-btn');
@@ -657,7 +585,6 @@ async function runPipeline() {
   document.getElementById('ps-btn-icon').className = 'spinner';
   document.getElementById('ps-btn-txt').textContent = 'Running…';
   _setBadge('running');
-  _showSteps(DAGS[cur]);
   _openTerm();
 
   const ctx = {};
@@ -703,7 +630,6 @@ function _poll() {
 
       document.getElementById('term-pct').textContent = exec.progress!=null?exec.progress+'%':'';
       renderDAG(DAGS[cur], tstates);
-      _renderSteps(DAGS[cur], tstates);
       _updateCfgStatus();
 
       if (exec.status==='completed') {