cloud_orchestrator.py

#!/usr/bin/env python3
"""
MeshAI Cloud Orchestrator - Nomadic Training & Continuous Checkpoint Sync Engine
This script manages the global fine-tuning loop on cloud GPUs (A100/H100),
automatically resuming from the latest checkpoint and pushing weights to Hugging Face.
"""
from __future__ import annotations

import os
import shutil
import subprocess
import sys
import argparse
from datetime import datetime
from pathlib import Path

# [Kesin] Windows/Linux terminal cikti dilini UTF-8 olarak zorla
if sys.platform == "win32":
    import io

    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")

# [Kesin] Ortam Degiskenleri Kontrolu
HF_TOKEN = os.getenv("HF_TOKEN")  # Sizin Write yetkili hf_... tokeniniz
HF_REPO = os.getenv("HF_REPO")  # Orn: "HayrettinIscan/MeshAI-Base-Models"
HF_REPO_TYPE = os.getenv("HF_REPO_TYPE", "model")
ORCHESTRATOR_VERSION = "v1.1-auto-resume-model-repo"

ROOT = Path(__file__).resolve().parent
LOG_DIR = ROOT / "logs"
LOG_DIR.mkdir(exist_ok=True)
CHECKPOINT_DIR = ROOT / "checkpoints"
CHECKPOINT_DIR.mkdir(exist_ok=True)


def _log(msg: str) -> None:
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    satir = f"[{timestamp}] [Orchestrator] {msg}"
    print(satir, flush=True)
    with open(LOG_DIR / "cloud_orchestrator.log", "a", encoding="utf-8") as f:
        f.write(satir + "\n")


def check_env() -> None:
    """[Kesin] Ortam degiskenlerini ve Hugging Face baglantisini dogrular."""
    _log(f"Orchestrator surumu: {ORCHESTRATOR_VERSION}")
    if not HF_TOKEN or not HF_REPO:
        _log("[Kesin] HATA: HF_TOKEN veya HF_REPO ortam degiskenleri eksik! Deploy baslatilamaz.")
        sys.exit(1)

    try:
        from huggingface_hub import HfApi

        api = HfApi(token=HF_TOKEN)
        api.create_repo(repo_id=HF_REPO, repo_type=HF_REPO_TYPE, exist_ok=True, token=HF_TOKEN)
        _log(f"[Kesin] HF depo hazir: {HF_REPO} ({HF_REPO_TYPE})")
    except Exception as exc:
        _log(f"[Kesin] HATA: HF deposu hazirlanamadi: {exc}")
        sys.exit(1)

    lock_file = ROOT / ".orchestrator.lock"
    if lock_file.exists():
        _log("[Kesin] HATA: Baska bir orchestrator zaten calisiyor. Tekrar baslatmayin.")
        sys.exit(1)
    lock_file.write_text(str(os.getpid()), encoding="utf-8")


def download_latest_checkpoint() -> bool:
    """[Kesin] Parcali sunucu korumasi: Depodaki en guncel egitilmis agirligi indirir."""
    try:
        from huggingface_hub import HfApi, hf_hub_download

        api = HfApi(token=HF_TOKEN)
        _log(f"{HF_REPO} deposu kontrol ediliyor...")

        files = api.list_repo_files(repo_id=HF_REPO, repo_type=HF_REPO_TYPE)
        ckpt_files = sorted(
            [
                f
                for f in files
                if f == "checkpoints/latest_model.pt"
                or (f.startswith("checkpoints/checkpoint_") and f.endswith(".pt"))
                or (f.startswith("checkpoint_") and f.endswith(".pt"))
            ]
        )

        if not ckpt_files:
            _log("[Kesin] Depoda eski kayit bulunamadi. Sifirdan (Base Model) egitim baslatilacak.")
            return False

        latest_ckpt = "checkpoints/latest_model.pt" if "checkpoints/latest_model.pt" in ckpt_files else ckpt_files[-1]
        _log(f"[Kesin] En guncel kayit bulundu: {latest_ckpt}. Buluttan indiriliyor...")

        downloaded = hf_hub_download(
            repo_id=HF_REPO,
            filename=latest_ckpt,
            local_dir=str(CHECKPOINT_DIR),
            repo_type=HF_REPO_TYPE,
            token=HF_TOKEN,
        )

        shutil.copy2(downloaded, CHECKPOINT_DIR / "latest_model.pt")
        _log("[Kesin] Indirme tamamlandi. Egitim kalindigi yerden devam edecek.")
        return True
    except Exception as e:
        _log(f"[Tahmin] HF baglanti hatasi veya bos depo: {e}. Egitim ilk adimdan baslatiliyor.")
        return False


def run_training_pipeline(epochs: int, validation_every: int) -> None:
    """[Kesin] Hibrit egitim motorunu (train_pipeline.py) tetikler."""
    has_checkpoint = download_latest_checkpoint()

    cmd = [
        sys.executable,
        "-u",
        "train_pipeline.py",
        "--epochs",
        str(epochs),
        "--validation-every",
        str(validation_every),
    ]
    if has_checkpoint:
        cmd.extend(["--resume_from", str(CHECKPOINT_DIR / "latest_model.pt")])

    _log(f"[Kesin] Egitim motoru baslatiliyor: {' '.join(cmd)}")

    env = os.environ.copy()
    env["PYTHONUNBUFFERED"] = "1"

    process = subprocess.Popen(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
        universal_newlines=True,
        env=env,
        cwd=str(ROOT),
    )

    epoch_counter = 1
    uploads_done = 0
    if process.stdout:
        for line in process.stdout:
            print(line, end="", flush=True)
            if "epoch" in line.lower() and "tamamlandi" in line.lower():
                _log(f"Epoch {epoch_counter} sinyali yakalandi. Otomatik bulut yedeklemesi tetikleniyor...")
                _upload_checkpoint_to_hf(epoch_counter)
                uploads_done += 1
                epoch_counter += 1

    process.wait()
    if process.returncode != 0:
        _log(
            f"[Kesin] HATA: Egitim motoru beklenmedik bir sekilde coktu! Hata kodu: {process.returncode}"
        )
        sys.exit(process.returncode)

    latest = CHECKPOINT_DIR / "latest_model.pt"
    if uploads_done == 0 and latest.exists():
        _log("[Tahmin] Canli log sinyali kacirildi; son checkpoint yedeklemesi tetikleniyor...")
        _upload_checkpoint_to_hf(max(epoch_counter - 1, 1))


def _upload_checkpoint_to_hf(epoch: int) -> None:
    """[Kesin] Epoch sonu uretilen agirligi arka planda Hugging Face'e muhurler."""
    try:
        from huggingface_hub import HfApi

        api = HfApi(token=HF_TOKEN)
        local_file = CHECKPOINT_DIR / "latest_model.pt"

        if not local_file.exists():
            _log("[Kesin] HATA: latest_model.pt bulunamadi, yedekleme atlandi.")
            return

        _log(f"[Muhtemel] latest_model.pt Hugging Face {HF_REPO} deposuna aktariliyor...")

        api.upload_file(
            path_or_fileobj=str(local_file),
            path_in_repo="checkpoints/latest_model.pt",
            repo_id=HF_REPO,
            repo_type=HF_REPO_TYPE,
            token=HF_TOKEN,
            commit_message=f"Auto checkpoint epoch {epoch}",
        )

        epoch_file = f"checkpoints/checkpoint_epoch_{epoch:03d}.pt"
        api.upload_file(
            path_or_fileobj=str(local_file),
            path_in_repo=epoch_file,
            repo_id=HF_REPO,
            repo_type=HF_REPO_TYPE,
            token=HF_TOKEN,
            commit_message=f"Archive checkpoint epoch {epoch}",
        )

        for local_name, remote_name in [
            ("training_progress.log", "logs/training_progress.log"),
            ("training_status.json", "logs/training_status.json"),
            ("logs/cloud_orchestrator.log", "logs/cloud_orchestrator.log"),
        ]:
            path = ROOT / local_name
            if path.exists():
                api.upload_file(
                    path_or_fileobj=str(path),
                    path_in_repo=remote_name,
                    repo_id=HF_REPO,
                    repo_type=HF_REPO_TYPE,
                    token=HF_TOKEN,
                    commit_message=f"Auto logs epoch {epoch}",
                )

        _log(f"[Kesin] YEDEKLEME BASARILI: latest_model.pt ve {epoch_file} bulutta guvende.")
    except Exception as e:
        _log(f"[Kesin] CRITICAL HATA: Yedekleme basarisiz oldu! {e}")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="MeshAI nomadic training orchestrator")
    parser.add_argument("--epochs", type=int, default=5, help="Bu oturumda kac epoch calissin")
    parser.add_argument("--validation-every", type=int, default=500, help="Kac stepte bir validation")
    return parser.parse_args()


if __name__ == "__main__":
    try:
        args = parse_args()
        check_env()
        run_training_pipeline(epochs=args.epochs, validation_every=args.validation_every)
    finally:
        lock_file = ROOT / ".orchestrator.lock"
        if lock_file.exists():
            lock_file.unlink(missing_ok=True)