Vertex AI

from dataclasses import dataclass, field
from typing import List, Dict, Any
import random

# ─── محاكاة Vertex AI SDK ──────────────────────────────────
@dataclass
class VertexModel:
    display_name:  str
    framework:     str
    artifact_uri:  str
    serving_image: str
    labels:        Dict[str, str] = field(default_factory=dict)

@dataclass
class VertexEndpoint:
    display_name: str
    model_name:   str
    machine_type: str
    min_replicas: int = 1
    max_replicas: int = 5
    traffic:      int = 100  # %
    latency_ms:   float = 0.0

    def predict(self, instances: List[Dict]) -> List[Dict]:
        results = []
        for inst in instances:
            text   = inst.get("text", "")
            score  = round(random.uniform(0.6, 0.99), 4)
            label  = "positive" if score > 0.5 else "negative"
            lat    = round(random.uniform(15, self.latency_ms + 30), 1)
            results.append({"label": label, "score": score, "latency_ms": lat})
        return results

class VertexAI:
    MODEL_GARDEN = {
        "gemini-1.5-pro":  {"type": "LLM",  "ctx": "1M",  "provider": "Google"},
        "gemini-1.5-flash":{"type": "LLM",  "ctx": "1M",  "provider": "Google"},
        "llama-3-70b":     {"type": "LLM",  "ctx": "8K",  "provider": "Meta"},
        "mistral-7b":      {"type": "LLM",  "ctx": "32K", "provider": "Mistral"},
        "text-bison":      {"type": "LLM",  "ctx": "8K",  "provider": "Google"},
        "imagetext":       {"type": "VLM",  "ctx": "-",   "provider": "Google"},
    }
    MACHINE_TYPES = {
        "n1-standard-4":   {"vCPUs": 4,  "RAM": "15GB"},
        "n1-highmem-8":    {"vCPUs": 8,  "RAM": "52GB"},
        "g2-standard-4":   {"vCPUs": 4,  "RAM": "16GB", "GPU": "L4"},
        "a2-highgpu-1g":   {"vCPUs": 12, "RAM": "85GB", "GPU": "A100"},
    }

    def __init__(self, project: str, region: str = "us-central1"):
        self.project  = project
        self.region   = region
        self.models:   Dict[str, VertexModel]    = {}
        self.endpoints:Dict[str, VertexEndpoint] = {}

    def upload_model(self, name: str, framework: str, uri: str) -> VertexModel:
        m = VertexModel(name, framework, uri, f"gcr.io/vertex-ai/{framework}")
        self.models[name] = m
        print(f"  ✅ Model uploaded: {name} ({framework})")
        return m

    def create_endpoint(self, name: str, model: str, machine: str,
                        min_r: int = 1, max_r: int = 3) -> VertexEndpoint:
        ep = VertexEndpoint(name, model, machine, min_r, max_r,
                            latency_ms=random.uniform(20, 60))
        self.endpoints[name] = ep
        mt = self.MACHINE_TYPES.get(machine, {})
        gpu_s = mt.get("GPU", "None")
        print(f"  ✅ Endpoint: {name} ({machine}, GPU={gpu_s})")
        return ep

    def show_model_garden(self):
        print(f"\n🌿 Model Garden (مختارات):")
        print(f"  {'Model':<22} {'Type':>5} {'Context':>8} {'Provider':>10}")
        print("  " + "-"*50)
        for m, info in self.MODEL_GARDEN.items():
            print(f"  {m:<22} {info['type']:>5} {info['ctx']:>8} {info['provider']:>10}")

    def batch_predict(self, endpoint: VertexEndpoint, texts: List[str]) -> List[Dict]:
        predictions = endpoint.predict([{"text": t} for t in texts])
        return predictions

# ─── استخدام Vertex AI ─────────────────────────────────────
random.seed(77)
print("🌐 Vertex AI — نشر نموذج تصنيف النصوص:")
print("=" * 55)

vertex = VertexAI("my-ai-prod", "us-central1")

# Model Garden
vertex.show_model_garden()

# رفع نموذج مخصص
print(f"\n\n📦 رفع نموذج مخصص:")
model = vertex.upload_model(
    "sentiment-v2-roberta",
    "pytorch",
    "gs://my-ai-prod/models/sentiment-v2/",
)

# إنشاء Endpoints
print(f"\n🚀 إنشاء Endpoints:")
ep_prod = vertex.create_endpoint("sentiment-prod", model.display_name, "g2-standard-4", 2, 10)
ep_test = vertex.create_endpoint("sentiment-test", model.display_name, "n1-standard-4", 1, 2)

# Online Prediction
print(f"\n⚡ Online Prediction:")
test_texts = [
    "المنتج رائع وجودة ممتازة!",
    "خدمة العملاء سيئة جداً",
    "تجربة لا بأس بها، مقبولة",
]
preds = vertex.batch_predict(ep_prod, test_texts)
for text, pred in zip(test_texts, preds):
    icon  = "🟢" if pred["label"] == "positive" else "🔴"
    lat_s = f"{pred['latency_ms']:.1f}ms"
    print(f"  {icon} [{pred['label']:<10}] {pred['score']:.3f} | {lat_s} | {text[:35]}")

# Autoscaling
print(f"\n\n📈 Autoscaling Configuration:")
for name, ep in vertex.endpoints.items():
    print(f"  [{name}]")
    print(f"   Machine : {ep.machine_type}")
    print(f"   Replicas: {ep.min_replicas} → {ep.max_replicas}")
    lat_s = f"{ep.latency_ms:.1f}ms"
    print(f"   Latency : {lat_s}")
print(f"\n✅ Vertex AI Endpoint جاهز للإنتاج!")