CI/CD for ML models

import time
from dataclasses import dataclass, field
from typing import List
from enum import Enum

class Status(Enum):
    PENDING = "⏳"
    RUNNING = "🔄"
    PASSED  = "✅"
    FAILED  = "❌"
    SKIPPED = "⏭️ "

@dataclass
class Step:
    name:     str
    command:  str
    status:   Status = Status.PENDING
    duration: float  = 0.0
    output:   str    = ""

class MLPipeline:
    """CI/CD Pipeline لنماذج ML"""

    def __init__(self, name: str):
        self.name  = name
        self.steps: List[Step] = []

    def add(self, name: str, command: str) -> "MLPipeline":
        self.steps.append(Step(name, command))
        return self

    def run(self, fail_at: str = "") -> bool:
        print(f"\n🚀 Pipeline: {self.name}")
        print("=" * 58)
        ok = True
        for step in self.steps:
            if not ok:
                step.status = Status.SKIPPED
                print(f"  {step.status.value} SKIP    {step.name}")
                continue

            print(f"  🔄 Running  {step.name}...", end="", flush=True)
            time.sleep(0.01)

            if step.name == fail_at:
                step.status = Status.FAILED
                step.output = f"بوابة الجودة: دقة النموذج أقل من 90%"
                ok = False
                print(f"\r  {step.status.value} FAILED  {step.name}")
                print(f"     └─ {step.output}")
            else:
                step.status   = Status.PASSED
                step.duration = round(0.5 + len(step.command) * 0.003, 1)
                dur_str = f"{step.duration}s"
                print(f"\r  {step.status.value} PASSED  {step.name} ({dur_str})")

        return ok

    def summary(self):
        passed   = sum(1 for s in self.steps if s.status == Status.PASSED)
        failed   = sum(1 for s in self.steps if s.status == Status.FAILED)
        skipped  = sum(1 for s in self.steps if s.status == Status.SKIPPED)
        total_t  = sum(s.duration for s in self.steps)
        dur_str  = f"{total_t:.1f}s"
        print(f"\n  {'─'*40}")
        print(f"  ✅ {passed} passed  ❌ {failed} failed  ⏭️  {skipped} skipped  ⏱️  {dur_str}")

# ─── إنشاء Pipeline ML ─────────────────────────────────────
def build_ml_pipeline(name: str) -> MLPipeline:
    return (
        MLPipeline(name)
        .add("تثبيت المتطلبات",     "pip install -r requirements.txt")
        .add("فحص جودة الكود",      "flake8 src/ && black --check src/")
        .add("اختبارات الوحدة",     "pytest tests/unit/ -v --cov=src")
        .add("التحقق من البيانات",  "python validate_data.py --schema schema.yaml")
        .add("تدريب النموذج",       "python train.py --config config.yaml --mlflow")
        .add("بوابة جودة النموذج",  "python gate.py --min-accuracy 0.90 --max-latency 100")
        .add("اختبارات التكامل",    "pytest tests/integration/ -v")
        .add("نشر على Staging",     "python deploy.py --env staging --blue-green")
        .add("اختبارات Smoke",      "python smoke_tests.py --env staging --timeout 30")
        .add("نشر على Production",  "python deploy.py --env production --canary 10%")
    )

# ─── سيناريو 1: نجاح كامل ─────────────────────────────────
print("🟢 سيناريو 1: Push على main — كل شيء يمر")
p1 = build_ml_pipeline("ML CI/CD — Success")
success = p1.run()
p1.summary()
result = "🎉 تم النشر على Production!" if success else "🛑 النشر موقوف"
print(f"\n  النتيجة: {result}")

# ─── سيناريو 2: فشل بوابة الجودة ─────────────────────────
print(f"\n\n🔴 سيناريو 2: دقة النموذج أقل من 90%")
p2 = build_ml_pipeline("ML CI/CD — Gate Failure")
fail = p2.run(fail_at="بوابة جودة النموذج")
p2.summary()
print(f"\n  النتيجة: 🛑 النشر موقوف — يجب تحسين الدقة")

# ─── DVC Commands ──────────────────────────────────────────
print(f"\n\n📦 DVC — إصدار البيانات والنماذج:")
dvc_cmds = [
    ("تهيئة DVC",        "dvc init"),
    ("تتبع البيانات",    "dvc add data/train.csv"),
    ("تتبع النموذج",     "dvc add models/classifier.pkl"),
    ("حفظ في S3",        "dvc push"),
    ("استرجاع من S3",    "dvc pull"),
    ("تشغيل Pipeline",   "dvc repro"),
    ("مقارنة تجربتين",   "dvc metrics diff main feature/new-model"),
]
for desc, cmd in dvc_cmds:
    print(f"  # {desc}")
    print(f"  $ {cmd}")
    print()