#!/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)