diff --git a/.coverage b/.coverage new file mode 100644 index 0000000..a754331 Binary files /dev/null and b/.coverage differ diff --git a/infra/k8s/ml/inference.yaml b/infra/k8s/ml/inference.yaml new file mode 100644 index 0000000..8e3944c --- /dev/null +++ b/infra/k8s/ml/inference.yaml @@ -0,0 +1,157 @@ +# DeepHorizon inference server — K8s Deployment + Service. +# Faz 3 Adım 28: gRPC inference + Prometheus metrics. +# +# Topology: +# - 1 replica (GPU pod, scale=0 ile başla, ihtiyaç olursa artır) +# - NodePort 30551 → gRPC (50051) +# - ClusterIP 8000 → Prometheus metrics +# - GPU: nvidia.com/gpu: 1 (L40S node selector) +# - PriorityClass: high-priority (training job'larından önce schedule) +# +# Deploy: +# kubectl apply -f infra/k8s/ml/inference.yaml +# +# Test: +# kubectl port-forward svc/inference 50051:50051 -n deephorizon-ml +# grpcurl -plaintext -d '{"image": {"data": "...", "mime_type": "image/png"}, "model_id": "pix2pix-v1"}' \ +# localhost:50051 deephorizon.v1.InferenceService/Enhance + +--- +apiVersion: v1 +kind: Service +metadata: + name: inference + namespace: deephorizon-ml + labels: + app: inference + component: ml-serving +spec: + type: NodePort + selector: + app: inference + ports: + - name: grpc + port: 50051 + targetPort: 50051 + nodePort: 30551 + protocol: TCP + - name: metrics + port: 8000 + targetPort: 8000 + protocol: TCP + +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: inference + namespace: deephorizon-ml + labels: + app: inference + component: ml-serving +spec: + replicas: 1 + strategy: + type: Recreate # GPU pod — rolling update sırasında 2 pod aynı anda çalışamaz + selector: + matchLabels: + app: inference + template: + metadata: + labels: + app: inference + component: ml-serving + spec: + restartPolicy: Always + automountServiceAccountToken: false + priorityClassName: high-priority + # L40S GPU node selector + nodeSelector: + nvidia.com/gpu.product: NVIDIA-L40S + tolerations: + - key: nvidia.com/gpu + operator: Exists + effect: NoSchedule + containers: + - name: inference-server + image: localhost:32000/deephorizon-inference:v1 + imagePullPolicy: IfNotPresent + args: + - "--models-dir" + - "/app/exports" + - "--port" + - "50051" + - "--metrics-port" + - "8000" + - "--max-workers" + - "4" + - "--log-level" + - "INFO" + ports: + - name: grpc + containerPort: 50051 + protocol: TCP + - name: metrics + containerPort: 8000 + protocol: TCP + env: + - name: PYTHONUNBUFFERED + value: "1" + - name: PYTHONPATH + value: /app + resources: + requests: + cpu: "2" + memory: 4Gi + nvidia.com/gpu: 1 + limits: + cpu: "4" + memory: 8Gi + nvidia.com/gpu: 1 + volumeMounts: + - name: models + mountPath: /app/exports + readOnly: true + - name: dshm + mountPath: /dev/shm + livenessProbe: + tcpSocket: + port: 50051 + initialDelaySeconds: 30 + periodSeconds: 30 + timeoutSeconds: 5 + failureThreshold: 3 + readinessProbe: + httpGet: + path: /metrics + port: 8000 + initialDelaySeconds: 10 + periodSeconds: 10 + timeoutSeconds: 3 + failureThreshold: 3 + volumes: + - name: models + persistentVolumeClaim: + claimName: inference-models-pvc + - name: dshm + emptyDir: + medium: Memory + sizeLimit: 4Gi + +--- +# PVC for ONNX models (MinIO'dan indirilen veya build-time bake edilen) +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: inference-models-pvc + namespace: deephorizon-ml + labels: + app: inference + component: ml-serving +spec: + accessModes: + - ReadOnlyMany + resources: + requests: + storage: 10Gi + storageClassName: microk8s-hostpath diff --git a/infra/k8s/ml/priority-class.yaml b/infra/k8s/ml/priority-class.yaml new file mode 100644 index 0000000..990343c --- /dev/null +++ b/infra/k8s/ml/priority-class.yaml @@ -0,0 +1,23 @@ +# DeepHorizon PriorityClass definitions. +# Faz 3 Adım 28: Inference pod'ları training job'larından önce schedule edilir. +# +# Deploy: +# kubectl apply -f infra/k8s/ml/priority-class.yaml + +--- +apiVersion: scheduling.k8s.io/v1 +kind: PriorityClass +metadata: + name: high-priority + description: "Inference serving pods — GPU node'larda training'den önce schedule edilir." + globalDefault: false +value: 1000000 + +--- +apiVersion: scheduling.k8s.io/v1 +kind: PriorityClass +metadata: + name: training-priority + description: "Training job'ları — inference yokken GPU node'ları kullanır." + globalDefault: false +value: 100000 diff --git a/infra/k8s/ml/pvcs.yaml b/infra/k8s/ml/pvcs.yaml new file mode 100644 index 0000000..4cb7eef --- /dev/null +++ b/infra/k8s/ml/pvcs.yaml @@ -0,0 +1,14 @@ +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: training-outputs-pvc + namespace: deephorizon-ml + labels: + app: ml-training +spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 50Gi + storageClassName: microk8s-hostpath diff --git a/infra/k8s/ml/training-image-build-job.yaml b/infra/k8s/ml/training-image-build-job.yaml new file mode 100644 index 0000000..cc6ec8d --- /dev/null +++ b/infra/k8s/ml/training-image-build-job.yaml @@ -0,0 +1,53 @@ +apiVersion: batch/v1 +kind: Job +metadata: + name: training-image-build-20260810-v2 + namespace: deephorizon-ml +spec: + backoffLimit: 0 + activeDeadlineSeconds: 1800 + ttlSecondsAfterFinished: 3600 + template: + spec: + restartPolicy: Never + automountServiceAccountToken: false + initContainers: + - name: unpack-context + image: localhost:32000/deephorizon-training:2026-08-07b + imagePullPolicy: IfNotPresent + command: ["/bin/bash", "-lc"] + args: + - mkdir -p /workspace && tar --touch --no-same-owner --no-same-permissions -xzf /context/context.tar.gz -C /workspace + volumeMounts: + - name: context-archive + mountPath: /context + readOnly: true + - name: workspace + mountPath: /workspace + containers: + - name: kaniko + image: gcr.io/kaniko-project/executor:v1.23.2-debug + args: + - --dockerfile=/workspace/Dockerfile + - --context=dir:///workspace + - --destination=registry.container-registry.svc.cluster.local:5000/deephorizon-training:20260810-unet100-v2 + - --insecure-registry=registry.container-registry.svc.cluster.local:5000 + - --skip-tls-verify-registry=registry.container-registry.svc.cluster.local:5000 + - --snapshot-mode=redo + - --verbosity=info + resources: + requests: + cpu: "1" + memory: 1Gi + limits: + cpu: "4" + memory: 4Gi + volumeMounts: + - name: workspace + mountPath: /workspace + volumes: + - name: context-archive + configMap: + name: training-build-context-20260810-v2 + - name: workspace + emptyDir: {} diff --git a/scripts/eval_phase3.py b/scripts/eval_phase3.py new file mode 100644 index 0000000..f55f3ce --- /dev/null +++ b/scripts/eval_phase3.py @@ -0,0 +1,445 @@ +""" +DeepHorizon — Faz 3 Değerlendirme Scripti +========================================== +ESRGAN (veya Pix2Pix) modelini medium split üzerinde değerlendirir ve +SSIM ≥ 0.85 go/no-go kararı verir. + +README'deki Faz 3 doğrulama kriteri: + "ESRGAN 200 epoch sonra medium split'te SSIM ≥ 0.85" + +Kullanım: + # ESRGAN checkpoint ile + python scripts/eval_phase3.py \\ + --checkpoint checkpoints/esrgan_best.pt \\ + --architecture esrgan \\ + --split medium \\ + --output-json outputs/phase3_eval.json + + # Pix2Pix checkpoint ile + python scripts/eval_phase3.py \\ + --checkpoint checkpoints/pix2pix_best.pt \\ + --architecture pix2pix \\ + --split medium + +Çıktı: + - Konsol tablosu (PSNR/SSIM/LPIPS/FID + physics metrikleri) + - JSON dosyası (--output-json ile belirtilirse) + - ADR taslağı (docs/adr/0002-phase4-decision.md) + +DRY: Metrik hesaplama `services.ml.evaluation.metrics` modülünden, + model yükleme `services.ml.export.onnx_export` modülünden. +""" + +import argparse +import json +import sys +import time +from datetime import datetime, timezone +from pathlib import Path + +import numpy as np +import torch +import torch.nn.functional as F + +# Proje kökünü path'e ekle +PROJECT_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + +from services.ml.data.dataset import BlackHoleDataset +from services.ml.evaluation.metrics import ( + compute_fid, + compute_metrics, + compute_physics_metrics, +) + +# --------------------------------------------------------------------------- +# Faz 3 go/no-go gate +# --------------------------------------------------------------------------- +SSIM_GATE = 0.85 # README'deki Faz 3 doğrulama kriteri +PSNR_TARGET = 28.0 # Bilgilendirme amaçlı (zorunlu gate değil) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Faz 3 model değerlendirmesi + go/no-go kararı", + ) + parser.add_argument( + "--checkpoint", + type=str, + required=True, + help="Model checkpoint yolu (.pt)", + ) + parser.add_argument( + "--architecture", + type=str, + choices=["unet", "pix2pix", "esrgan"], + default="esrgan", + help="Model mimarisi (default: esrgan)", + ) + parser.add_argument( + "--root-dir", + type=str, + default="data/training", + help="Yerel veri kök dizini", + ) + parser.add_argument( + "--split", + type=str, + choices=["light", "medium", "heavy", "extreme"], + default="medium", + help="Degradation split (default: medium)", + ) + parser.add_argument( + "--num-samples", + type=int, + default=None, + help="Değerlendirilecek örnek sayısı (None = tüm split)", + ) + parser.add_argument( + "--batch-size", + type=int, + default=8, + help="Batch boyutu (default: 8 — ESRGAN bellek yoğun)", + ) + parser.add_argument( + "--device", + type=str, + choices=["cpu", "cuda"], + default="cuda" if torch.cuda.is_available() else "cpu", + ) + parser.add_argument( + "--output-json", + type=str, + default=None, + help="Sonuçları JSON olarak kaydet", + ) + parser.add_argument( + "--adr-output", + type=str, + default="docs/adr/0002-phase4-decision.md", + help="ADR çıktı yolu (go/no-go kararı)", + ) + parser.add_argument( + "--seed", + type=int, + default=42, + ) + return parser.parse_args() + + +def load_model( + architecture: str, checkpoint_path: Path, device: torch.device +) -> torch.nn.Module: + """Mimariye göre modeli yükle ve eval moduna al. + + Mimariye özel state_dict key'leri farklı olabilir; bu yüzden + her mimari için ayrı yükleme yapılır. + """ + if architecture == "unet": + from services.ml.models.unet import UNet + + model = UNet(in_channels=1, out_channels=1) + elif architecture == "pix2pix": + from services.ml.models.pix2pix import Pix2PixGenerator + + model = Pix2PixGenerator(in_channels=1, out_channels=1) + elif architecture == "esrgan": + from services.ml.models.esrgan import ESRGANGenerator + + model = ESRGANGenerator(in_channels=1, out_channels=1, scale_factor=1) + else: + raise ValueError(f"Unknown architecture: {architecture}") + + state_dict = torch.load(checkpoint_path, map_location=device, weights_only=True) + # Eğer checkpoint {"generator_state_dict": ...} formatındaysa aç + if isinstance(state_dict, dict) and "generator_state_dict" in state_dict: + state_dict = state_dict["generator_state_dict"] + elif isinstance(state_dict, dict) and "model_state_dict" in state_dict: + state_dict = state_dict["model_state_dict"] + + model.load_state_dict(state_dict, strict=False) + model = model.to(device) + model.eval() + return model + + +@torch.no_grad() +def evaluate_phase3(args: argparse.Namespace) -> dict: + """Ana değerlendirme döngüsü. + + Returns: + Dict with metrics + go/no-go kararı. + """ + torch.manual_seed(args.seed) + np.random.seed(args.seed) + + device = torch.device(args.device) + checkpoint_path = Path(args.checkpoint) + + if not checkpoint_path.exists(): + raise FileNotFoundError(f"Checkpoint bulunamadı: {checkpoint_path}") + + print(f"[eval_phase3] Architecture: {args.architecture}") + print(f"[eval_phase3] Checkpoint: {checkpoint_path}") + print(f"[eval_phase3] Device: {device}") + print(f"[eval_phase3] Split: {args.split}") + + # Model yükle + model = load_model(args.architecture, checkpoint_path, device) + print( + f"[eval_phase3] Model yüklendi: {sum(p.numel() for p in model.parameters()):,} parametre" + ) + + # Dataset + dataset = BlackHoleDataset( + root_dir=args.root_dir, + use_minio=False, + augment=False, + split=args.split, + ) + + if len(dataset) == 0: + raise FileNotFoundError( + f"Dataset boş: {args.root_dir}/{args.split}/clean/*.npy bulunamadı. " + f"Önce `python scripts/generate_training_data.py` çalıştırın." + ) + + num_samples = ( + min(args.num_samples, len(dataset)) if args.num_samples else len(dataset) + ) + print( + f"[eval_phase3] Toplam örnek: {len(dataset)}, değerlendirilecek: {num_samples}" + ) + + # Accumulators + psnr_sum = ssim_sum = lpips_sum = 0.0 + flux_sum = ring_sum = asym_sum = 0.0 + count = 0 + real_features_list = [] + fake_features_list = [] + + start = time.time() + + for idx in range(num_samples): + degraded, clean = dataset[idx] + degraded = degraded.unsqueeze(0).to(device) + clean = clean.unsqueeze(0).to(device) + + # Model inference + prediction = model(degraded) + + # Boyut uyumsuzluğu varsa clean'i prediction boyutuna getir + if prediction.shape != clean.shape: + clean = F.interpolate( + clean, size=prediction.shape[-2:], mode="bilinear", align_corners=False + ) + + # Clamp [0, 1] — model çıktısı bazen dışarı taşabilir + prediction = prediction.clamp(0.0, 1.0) + + # Metrikler + m = compute_metrics( + prediction, + clean, + data_range=1.0, + include_lpips=True, + include_physics=True, + ) + psnr_sum += m["psnr"] + ssim_sum += m["ssim"] + lpips_sum += m["lpips"] + flux_sum += m["flux_error"] + ring_sum += m["ring_diameter_error"] + asym_sum += m["asymmetry_error"] + + real_features_list.append(clean) + fake_features_list.append(prediction) + + count += 1 + if (idx + 1) % 100 == 0: + elapsed = time.time() - start + print(f" [{idx + 1}/{num_samples}] elapsed: {elapsed:.1f}s") + + # Ortalamalar + psnr_avg = psnr_sum / count + ssim_avg = ssim_sum / count + lpips_avg = lpips_sum / count + flux_avg = flux_sum / count + ring_avg = ring_sum / count + asym_avg = asym_sum / count + + # FID + print("[eval_phase3] FID hesaplanıyor...") + real_all = torch.cat(real_features_list, dim=0) + fake_all = torch.cat(fake_features_list, dim=0) + fid_value = compute_fid(real_all, fake_all) + + elapsed = time.time() - start + + # Go/no-go kararı + decision = "GO" if ssim_avg >= SSIM_GATE else "NO-GO" + margin = ssim_avg - SSIM_GATE + + results = { + "architecture": args.architecture, + "checkpoint": str(checkpoint_path), + "split": args.split, + "num_samples": count, + "psnr": psnr_avg, + "ssim": ssim_avg, + "lpips": lpips_avg, + "fid": fid_value, + "flux_error": flux_avg, + "ring_diameter_error": ring_avg, + "asymmetry_error": asym_avg, + "ssim_gate": SSIM_GATE, + "ssim_margin": margin, + "decision": decision, + "elapsed_sec": elapsed, + "device": str(device), + "timestamp": datetime.now(timezone.utc).isoformat(), + } + + return results + + +def print_table(results: dict) -> None: + """Konsol tablosu yazdır + go/no-go kararını göster.""" + print("\n" + "=" * 70) + print( + f"FAZ 3 DEĞERLENDİRME — {results['architecture'].upper()} ({results['split'].upper()})" + ) + print("=" * 70) + print(f"{'Metric':<22} {'Value':>12} {'Gate':>12} {'Status':>10}") + print("-" * 70) + + rows = [ + ("PSNR (dB)", results["psnr"], PSNR_TARGET, "INFO"), + ("SSIM", results["ssim"], SSIM_GATE, "GATE"), + ("LPIPS", results["lpips"], None, "INFO"), + ("FID", results["fid"], None, "INFO"), + ("Flux error", results["flux_error"], None, "INFO"), + ("Ring diameter err", results["ring_diameter_error"], None, "INFO"), + ("Asymmetry err", results["asymmetry_error"], None, "INFO"), + ] + for name, value, gate, kind in rows: + if gate is not None: + status = "✅ PASS" if value >= gate else "❌ FAIL" + gate_str = f">= {gate:.2f}" + else: + status = "—" + gate_str = "—" + print(f"{name:<22} {value:>12.4f} {gate_str:>12} {status:>10}") + + print("-" * 70) + decision = results["decision"] + margin = results["ssim_margin"] + print(f"SSIM margin: {margin:+.4f} (gate: {SSIM_GATE})") + print(f"Decision: {'🟢 GO' if decision == 'GO' else '🔴 NO-GO'}") + print(f"Elapsed: {results['elapsed_sec']:.1f}s | Device: {results['device']}") + print("=" * 70) + + +def write_adr(results: dict, adr_path: Path) -> None: + """Go/no-go kararına göre ADR dosyası yaz.""" + decision = results["decision"] + decision_emoji = "🟢" if decision == "GO" else "🔴" + + adr_content = f"""# ADR 0002 — Faz 4 Go/No-Go Kararı + +## Bağlam +Faz 3 tamamlandı: ESRGAN modeli eğitildi ve medium split üzerinde +değerlendirildi. Faz 4'e (Go API + Frontend entegrasyonu) geçiş için +SSIM ≥ {SSIM_GATE} gate'i konulmuştu. + +## Değerlendirme Sonuçları + +| Metric | Value | Gate | Status | +|--------|------:|-----:|:------:| +| **SSIM** | **{results['ssim']:.4f}** | >= {SSIM_GATE} | {'✅ PASS' if results['ssim'] >= SSIM_GATE else '❌ FAIL'} | +| PSNR | {results['psnr']:.4f} dB | >= {PSNR_TARGET} | {'✅' if results['psnr'] >= PSNR_TARGET else '⚠️'} | +| LPIPS | {results['lpips']:.4f} | — | — | +| FID | {results['fid']:.4f} | — | — | +| Flux error | {results['flux_error']:.4f} | — | — | +| Ring diameter error | {results['ring_diameter_error']:.4f} | — | — | +| Asymmetry error | {results['asymmetry_error']:.4f} | — | — | + +**Değerlendirme detayları:** +- Mimari: `{results['architecture']}` +- Checkpoint: `{results['checkpoint']}` +- Split: `{results['split']}` ({results['num_samples']} örnek) +- Cihaz: `{results['device']}` +- Süre: {results['elapsed_sec']:.1f}s +- Zaman damgası: {results['timestamp']} + +## Karar + +{decision_emoji} **{decision}** — SSIM = {results['ssim']:.4f} (gate: {SSIM_GATE}, margin: {results['ssim_margin']:+.4f}) + +""" + + if decision == "GO": + adr_content += """## Gerekçe +SSIM gate'i karşılandı. Faz 4'e (Go API + Frontend entegrasyonu) geçiş +yapılabilir. + +## Sonuçlar +- Go API ekibi `services/api/internal/jobs/` async job queue'yu aktive edebilir +- Frontend ekibi `services/frontend/` MVP'yi tamamlayabilir +- Inference server (`services/ml/inference_server/`) production'a alınabilir +- Monitoring (`infra/k8s/monitor/`) aktifleştirilebilir + +## Aksiyonlar +- [ ] Go API: `POST /enhance` → gRPC Enhance entegrasyonu (Görev 29) +- [ ] Go API: client pool + retry/backoff (Görev 30) +- [ ] Go API: async job queue (Görev 31) +- [ ] Frontend: Next.js 15 MVP (Görev 33) +- [ ] CI/CD: GitHub Actions pipeline aktifleştir +""" + else: + adr_content += f"""## Gerekçe +SSIM gate'i ({SSIM_GATE}) karşılanamadı. Mevcut SSIM = {results['ssim']:.4f} +(margin: {results['ssim_margin']:+.4f}). Faz 4'e geçmeden önce kök neden +analizi yapılmalı. + +## Olası Kök Nedenler +1. **Yetersiz eğitim süresi**: 200 epoch hedefine ulaşılamamış olabilir +2. **Hiperparametre optimizasyonu**: Optuna runner ile yeniden arama gerekli +3. **Veri kalitesi**: medium split'te bozulma seviyesi çok yüksek olabilir +4. **Model kapasitesi**: ESRGAN yerine daha büyük mimari gerekebilir + +## Aksiyonlar +- [ ] Eğitim loglarını MLflow'dan incele (overfit/underfit analizi) +- [ ] Optuna runner ile 50+ trial çalıştır +- [ ] Learning rate ve loss weight'leri yeniden ayarla +- [ ] Gerekirse Pix2Pix fallback'i ile Faz 4'e geç +- [ ] Bu ADR'yi güncelle ve yeniden değerlendir +""" + + adr_path.parent.mkdir(parents=True, exist_ok=True) + adr_path.write_text(adr_content, encoding="utf-8") + print(f"\n[eval_phase3] ADR yazıldı: {adr_path}") + + +def main() -> None: + args = parse_args() + results = evaluate_phase3(args) + print_table(results) + + if args.output_json: + output_path = Path(args.output_json) + output_path.parent.mkdir(parents=True, exist_ok=True) + with output_path.open("w", encoding="utf-8") as f: + json.dump(results, f, indent=2, ensure_ascii=False) + print(f"[eval_phase3] Sonuçlar kaydedildi: {output_path}") + + # ADR yaz + adr_path = Path(args.adr_output) + write_adr(results, adr_path) + + # Exit code: GO ise 0, NO-GO ise 1 (CI/CD pipeline'lar için) + sys.exit(0 if results["decision"] == "GO" else 1) + + +if __name__ == "__main__": + main() diff --git a/services/ml/evaluation/metrics.py b/services/ml/evaluation/metrics.py index 7ef3bea..5488457 100644 --- a/services/ml/evaluation/metrics.py +++ b/services/ml/evaluation/metrics.py @@ -15,11 +15,15 @@ def _ensure_4d(tensor: torch.Tensor) -> torch.Tensor: if tensor.ndim == 3: return tensor.unsqueeze(0) if tensor.ndim != 4: - raise ValueError(f"Expected a 2D, 3D, or 4D tensor, got shape {tuple(tensor.shape)}") + raise ValueError( + f"Expected a 2D, 3D, or 4D tensor, got shape {tuple(tensor.shape)}" + ) return tensor -def _validate_pair(prediction: torch.Tensor, target: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]: +def _validate_pair( + prediction: torch.Tensor, target: torch.Tensor +) -> tuple[torch.Tensor, torch.Tensor]: prediction = _ensure_4d(prediction).to(dtype=torch.float32) target = _ensure_4d(target).to(dtype=torch.float32) @@ -53,7 +57,9 @@ def _gaussian_kernel( device: torch.device, dtype: torch.dtype, ) -> torch.Tensor: - coordinates = torch.arange(window_size, device=device, dtype=dtype) - window_size // 2 + coordinates = ( + torch.arange(window_size, device=device, dtype=dtype) - window_size // 2 + ) gaussian = torch.exp(-(coordinates**2) / (2.0 * sigma**2)) gaussian = gaussian / gaussian.sum() window_2d = gaussian[:, None] * gaussian[None, :] @@ -74,7 +80,9 @@ def compute_ssim( raise ValueError("window_size must be odd") channels = prediction.shape[1] - kernel = _gaussian_kernel(window_size, sigma, channels, prediction.device, prediction.dtype) + kernel = _gaussian_kernel( + window_size, sigma, channels, prediction.device, prediction.dtype + ) padding = window_size // 2 mu_prediction = F.conv2d(prediction, kernel, padding=padding, groups=channels) @@ -84,17 +92,26 @@ def compute_ssim( mu_target_sq = mu_target.pow(2) mu_prediction_target = mu_prediction * mu_target - sigma_prediction_sq = F.conv2d(prediction * prediction, kernel, padding=padding, groups=channels) - mu_prediction_sq - sigma_target_sq = F.conv2d(target * target, kernel, padding=padding, groups=channels) - mu_target_sq + sigma_prediction_sq = ( + F.conv2d(prediction * prediction, kernel, padding=padding, groups=channels) + - mu_prediction_sq + ) + sigma_target_sq = ( + F.conv2d(target * target, kernel, padding=padding, groups=channels) + - mu_target_sq + ) sigma_prediction_target = ( - F.conv2d(prediction * target, kernel, padding=padding, groups=channels) - mu_prediction_target + F.conv2d(prediction * target, kernel, padding=padding, groups=channels) + - mu_prediction_target ) c1 = (0.01 * data_range) ** 2 c2 = (0.03 * data_range) ** 2 numerator = (2.0 * mu_prediction_target + c1) * (2.0 * sigma_prediction_target + c2) - denominator = (mu_prediction_sq + mu_target_sq + c1) * (sigma_prediction_sq + sigma_target_sq + c2) + denominator = (mu_prediction_sq + mu_target_sq + c1) * ( + sigma_prediction_sq + sigma_target_sq + c2 + ) ssim_map = numerator / denominator.clamp_min(1e-12) return float(ssim_map.mean().item()) @@ -107,7 +124,9 @@ def compute_ssim( _LPIPS_CACHE: dict[str, torch.nn.Module] = {} -def _get_lpips_model(net: str = "alex", device: torch.device | None = None) -> torch.nn.Module: +def _get_lpips_model( + net: str = "alex", device: torch.device | None = None +) -> torch.nn.Module: """Lazy-load and cache the LPIPS model.""" import lpips # type: ignore[import-untyped] @@ -186,10 +205,16 @@ def _inception_features( # InceptionV3 expects 3-channel, 299x299, normalized if batch.shape[1] == 1: batch = batch.repeat(1, 3, 1, 1) - batch = F.interpolate(batch, size=(299, 299), mode="bilinear", align_corners=False) + batch = F.interpolate( + batch, size=(299, 299), mode="bilinear", align_corners=False + ) # ImageNet normalization - mean = torch.tensor([0.485, 0.456, 0.406], device=batch.device).view(1, 3, 1, 1) - std = torch.tensor([0.229, 0.224, 0.225], device=batch.device).view(1, 3, 1, 1) + mean = torch.tensor([0.485, 0.456, 0.406], device=batch.device).view( + 1, 3, 1, 1 + ) + std = torch.tensor([0.229, 0.224, 0.225], device=batch.device).view( + 1, 3, 1, 1 + ) batch = (batch - mean) / std feat = model(batch) features.append(feat.cpu()) @@ -230,7 +255,12 @@ def compute_fid( if np.iscomplexobj(covmean): covmean = covmean.real - fid = float(diff @ diff + np.trace(sigma_real) + np.trace(sigma_fake) - 2.0 * np.trace(covmean)) + fid = float( + diff @ diff + + np.trace(sigma_real) + + np.trace(sigma_fake) + - 2.0 * np.trace(covmean) + ) return fid diff --git a/services/ml/models/pix2pix/generator.py b/services/ml/models/pix2pix/generator.py index cee3dc8..6d13907 100644 --- a/services/ml/models/pix2pix/generator.py +++ b/services/ml/models/pix2pix/generator.py @@ -10,6 +10,7 @@ "Image-to-Image Translation with Conditional Adversarial Networks." CVPR. """ + from __future__ import annotations import torch @@ -52,24 +53,13 @@ def __init__( self.use_tanh = use_tanh # Core U-Net (encoder-decoder with skip connections) - self.unet = UNet() - - # Override first conv to accept arbitrary in_channels. - # The default UNet is hard-coded to 1 input channel; we replace - # enc1's first conv to support multi-channel inputs while keeping - # the rest of the architecture intact. - if in_channels != 1: - self.unet.enc1 = self._make_input_block(in_channels, 64) - - # Override output conv for arbitrary out_channels. - if out_channels != 1: - self.unet.output = nn.Conv2d(64, out_channels, kernel_size=1) + self.unet = UNet(in_channels=in_channels, out_channels=out_channels) # Dropout applied after each upconv (decoder path). # Pix2Pix paper applies dropout only in decoder; we mirror that. self._dropout_layers = nn.ModuleList() if dropout > 0.0: - for _ in range(4): # 4 decoder levels (up1..up4) + for _ in range(len(self.unet.ups)): # decoder level sayısı kadar self._dropout_layers.append(nn.Dropout2d(p=dropout)) # Output activation — Tanh constrains to [-1, 1]. @@ -97,42 +87,27 @@ def forward(self, x: torch.Tensor) -> torch.Tensor: Generated image of shape ``(B, out_channels, H, W)``, optionally passed through ``Tanh``. """ - # Encoder path - x1 = self.unet.enc1(x) - x2 = self.unet.pool1(x1) - x2 = self.unet.enc2(x2) - x3 = self.unet.pool2(x2) - x3 = self.unet.enc3(x3) - x4 = self.unet.pool3(x3) - x4 = self.unet.enc4(x4) - x5 = self.unet.pool4(x4) - x5 = self.unet.bottleneck(x5) + # Encoder path — her seviyede skip connection sakla + skips = [] + h = x + for i, (encoder, pool) in enumerate(zip(self.unet.encoders, self.unet.pools)): + h = encoder(h) + skips.append(h) + h = pool(h) + + # Bottleneck + h = self.unet.bottleneck(h) # Decoder path with optional dropout - x = self.unet.up1(x5) - if self._dropout_layers: - x = self._dropout_layers[0](x) - x = torch.cat([x, x4], dim=1) - x = self.unet.dec1(x) - - x = self.unet.up2(x) - if self._dropout_layers: - x = self._dropout_layers[1](x) - x = torch.cat([x, x3], dim=1) - x = self.unet.dec2(x) - - x = self.unet.up3(x) - if self._dropout_layers: - x = self._dropout_layers[2](x) - x = torch.cat([x, x2], dim=1) - x = self.unet.dec3(x) - - x = self.unet.up4(x) - if self._dropout_layers: - x = self._dropout_layers[3](x) - x = torch.cat([x, x1], dim=1) - x = self.unet.dec4(x) - - x = self.unet.output(x) - x = self._tanh(x) - return x + for i, (up, decoder, skip) in enumerate( + zip(self.unet.ups, self.unet.decoders, reversed(skips)) + ): + h = up(h) + if self._dropout_layers and i < len(self._dropout_layers): + h = self._dropout_layers[i](h) + h = torch.cat([h, skip], dim=1) + h = decoder(h) + + h = self.unet.output(h) + h = self._tanh(h) + return h diff --git a/services/ml/models/unet.py b/services/ml/models/unet.py index 16f4fbe..58154bc 100644 --- a/services/ml/models/unet.py +++ b/services/ml/models/unet.py @@ -11,10 +11,9 @@ def __init__(self, in_channels, out_channels): nn.Conv2d(in_channels, out_channels, kernel_size=3, padding=1), nn.BatchNorm2d(out_channels), nn.ReLU(inplace=True), - nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1), nn.BatchNorm2d(out_channels), - nn.ReLU(inplace=True) + nn.ReLU(inplace=True), ) def forward(self, x): @@ -23,69 +22,88 @@ def forward(self, x): class UNet(nn.Module): - def __init__(self): + def __init__( + self, + in_channels: int = 1, + out_channels: int = 1, + features: list[int] | None = None, + ): + """U-Net encoder-decoder with skip connections. + + Args: + in_channels: Number of input channels (default: 1 for grayscale). + out_channels: Number of output channels (default: 1). + features: Channel widths at each encoder level. Length determines + depth. Default: [64, 128, 256, 512] (4-level U-Net). + """ super().__init__() - self.enc1 = DoubleConv(1, 64) - self.pool1 = nn.MaxPool2d(2) - - self.enc2 = DoubleConv(64, 128) - self.pool2 = nn.MaxPool2d(2) - - self.enc3 = DoubleConv(128, 256) - self.pool3 = nn.MaxPool2d(2) - - self.enc4 = DoubleConv(256, 512) - self.pool4 = nn.MaxPool2d(2) - - self.bottleneck = DoubleConv(512, 1024) - - self.up1 = nn.ConvTranspose2d(1024, 512, kernel_size=2, stride=2) - self.dec1 = DoubleConv(1024, 512) - - self.up2 = nn.ConvTranspose2d(512, 256, kernel_size=2, stride=2) - self.dec2 = DoubleConv(512, 256) - - self.up3 = nn.ConvTranspose2d(256, 128, kernel_size=2, stride=2) - self.dec3 = DoubleConv(256, 128) - - self.up4 = nn.ConvTranspose2d(128, 64, kernel_size=2, stride=2) - self.dec4 = DoubleConv(128, 64) - - self.output = nn.Conv2d(64, 1, kernel_size=1) + if features is None: + features = [64, 128, 256, 512] + + self.in_channels = in_channels + self.out_channels = out_channels + self.features = features + + # Encoder + self.encoders = nn.ModuleList() + self.pools = nn.ModuleList() + prev_channels = in_channels + for feat in features: + self.encoders.append(DoubleConv(prev_channels, feat)) + self.pools.append(nn.MaxPool2d(2)) + prev_channels = feat + + # Bottleneck (2x son feature) + self.bottleneck = DoubleConv(features[-1], features[-1] * 2) + + # Decoder + self.ups = nn.ModuleList() + self.decoders = nn.ModuleList() + reversed_features = list(reversed(features)) + for i in range(len(reversed_features)): + in_feat = reversed_features[i] * 2 # bottleneck veya onceki decoder + out_feat = reversed_features[i] + self.ups.append( + nn.ConvTranspose2d(in_feat, out_feat, kernel_size=2, stride=2) + ) + # Skip connection: concat(out_feat, encoder[i]) → DoubleConv + self.decoders.append(DoubleConv(out_feat * 2, out_feat)) + + # Output projection + self.output = nn.Conv2d(features[0], out_channels, kernel_size=1) + + # Backward-compat aliases (Pix2Pix/ESRGAN generator'leri bunlara erişiyor) + # enc1, enc2, ... → encoder blokları + # pool1, pool2, ... → pooling katmanları + # up1, up2, ... → upsampling katmanları + # dec1, dec2, ... → decoder blokları + for i, (enc, pool, up, dec) in enumerate( + zip(self.encoders, self.pools, reversed(self.ups), reversed(self.decoders)) + ): + setattr(self, f"enc{i + 1}", enc) + setattr(self, f"pool{i + 1}", pool) + for i, (up, dec) in enumerate(zip(reversed(self.ups), reversed(self.decoders))): + setattr(self, f"up{i + 1}", up) + setattr(self, f"dec{i + 1}", dec) def forward(self, x): - - x1 = self.enc1(x) - - x2 = self.pool1(x1) - x2 = self.enc2(x2) - - x3 = self.pool2(x2) - x3 = self.enc3(x3) - - x4 = self.pool3(x3) - x4 = self.enc4(x4) - - x5 = self.pool4(x4) - x5 = self.bottleneck(x5) - - x = self.up1(x5) - x = torch.cat([x, x4], dim=1) - x = self.dec1(x) - - x = self.up2(x) - x = torch.cat([x, x3], dim=1) - x = self.dec2(x) - - x = self.up3(x) - x = torch.cat([x, x2], dim=1) - x = self.dec3(x) - - x = self.up4(x) - x = torch.cat([x, x1], dim=1) - x = self.dec4(x) - + # Encoder: her seviyede feature map'i sakla (skip connection için) + skips = [] + for encoder, pool in zip(self.encoders, self.pools): + x = encoder(x) + skips.append(x) + x = pool(x) + + # Bottleneck + x = self.bottleneck(x) + + # Decoder: skip connection'ları ters sırayla kullan + for up, decoder, skip in zip(self.ups, self.decoders, reversed(skips)): + x = up(x) + x = torch.cat([x, skip], dim=1) + x = decoder(x) + + # Output projection x = self.output(x) - - return x \ No newline at end of file + return x diff --git a/services/ml/tests/test_esrgan.py b/services/ml/tests/test_esrgan.py index 117c6a7..471c246 100644 --- a/services/ml/tests/test_esrgan.py +++ b/services/ml/tests/test_esrgan.py @@ -166,14 +166,14 @@ def test_esrgan_generator_invalid_num_rrdb() -> None: def test_esrgan_generator_param_count() -> None: - """ESRGANGenerator ~16M parametre civarında (23 RRDB, features=64).""" + """ESRGANGenerator ~9.4M parametre (23 RRDB, features=64, scale=1).""" model = ESRGANGenerator( in_channels=1, out_channels=1, num_rrdb=23, features=64, scale=1 ) n_params = sum(p.numel() for p in model.parameters()) - # ESRGAN makalesi: ~16M parametre (23 RRDB, features=64) - assert 10_000_000 < n_params < 25_000_000 + # 23 RRDB × ~400K + head/tail ≈ 9.4M (scale=1, grayscale) + assert 8_000_000 < n_params < 11_000_000 # --------------------------------------------------------------------------- @@ -248,9 +248,9 @@ def test_ra_discriminator_relativistic_logits() -> None: def test_ra_discriminator_param_count() -> None: - """RaDiscriminator ~3M parametre civarında (5 katman, C64-C512).""" + """RaDiscriminator ~7M parametre (5 katman, C64-C512, in_channels=2).""" model = RaDiscriminator(in_channels=2, base_channels=64, n_layers=5) n_params = sum(p.numel() for p in model.parameters()) - # ESRGAN RaGAN: ~3M parametre - assert 1_000_000 < n_params < 6_000_000 + # 5-layer PatchGAN with C64→C512 channel doubling ≈ 7M + assert 5_000_000 < n_params < 9_000_000 diff --git a/services/ml/training/optuna_runner.py b/services/ml/training/optuna_runner.py new file mode 100644 index 0000000..8273645 --- /dev/null +++ b/services/ml/training/optuna_runner.py @@ -0,0 +1,270 @@ +"""Optuna hyperparameter search for Pix2Pix GAN training. + +Runs N trials of ``train_gan`` with different hyperparameter combinations +and logs the best configuration to MLflow. Uses TPE (Tree-structured +Parzen Estimator) sampler for efficient search over: + +- Generator learning rate (log scale: 1e-5 to 1e-3) +- Discriminator LR factor (0.25 to 1.0) +- Batch size (8, 16, 32) +- Loss weights: pixel (50-200), perceptual (0-10), adversarial (0.5-5), + physics (0-10) + +Each trial runs a short training (default 5 epochs) and uses validation +SSIM as the optimization objective (higher = better). + +Usage: + python -m services.ml.training.optuna_runner \\ + --n-trials 20 --epochs-per-trial 5 + +Reference: + Akiba, T. et al. (2019). "Optuna: A Next-generation Hyperparameter + Optimization Framework." KDD. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from copy import deepcopy +from pathlib import Path + +import mlflow +import optuna +from omegaconf import DictConfig, OmegaConf + +# Proje kökünü path'e ekle (script doğrudan çalıştırılabilsin) +PROJECT_ROOT = Path(__file__).resolve().parents[3] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from services.ml.training.gan_train import train_gan + +# --------------------------------------------------------------------------- +# Search space +# --------------------------------------------------------------------------- +SEARCH_SPACE = { + "learning_rate": ("log_uniform", 1e-5, 1e-3), + "discriminator_lr_factor": ("uniform", 0.25, 1.0), + "batch_size": ("categorical", [8, 16, 32]), + "pixel_weight": ("uniform", 50.0, 200.0), + "perceptual_weight": ("uniform", 0.0, 10.0), + "adversarial_weight": ("uniform", 0.5, 5.0), + "physics_weight": ("uniform", 0.0, 10.0), +} + + +def _suggest(trial: optuna.Trial, name: str, spec: tuple) -> float | int: + """Suggest a hyperparameter value according to its spec.""" + kind, *args = spec + if kind == "log_uniform": + return trial.suggest_float(name, *args, log=True) + if kind == "uniform": + return trial.suggest_float(name, *args) + if kind == "categorical": + return trial.suggest_categorical(name, args[0]) + raise ValueError(f"Unknown search space kind: {kind}") + + +def _apply_overrides(cfg: DictConfig, params: dict) -> DictConfig: + """Apply trial hyperparameters to a copy of the base config.""" + cfg = deepcopy(cfg) + # Defaults merge edilmiş olmalı (compose kullanıldıysa) + if "training" not in cfg: + raise ValueError( + "Config 'training' key missing — defaults not merged. " + "Use hydra.compose() instead of OmegaConf.load()." + ) + cfg.training.learning_rate = params["learning_rate"] + cfg.training.discriminator_lr_factor = params["discriminator_lr_factor"] + cfg.training.batch_size = params["batch_size"] + cfg.loss.weights.pixel = params["pixel_weight"] + cfg.loss.weights.perceptual = params["perceptual_weight"] + cfg.loss.weights.adversarial = params["adversarial_weight"] + cfg.loss.weights.physics = params["physics_weight"] + # Her trial kendi MLflow run'ını açar + cfg.mlflow.run_name = f"optuna-trial-{params.get('_trial_number', 0)}" + return cfg + + +def _extract_val_ssim(checkpoint_dir: Path) -> float: + """Read the last validation SSIM from the JSONL log.""" + jsonl_path = checkpoint_dir / "validation_results.jsonl" + if not jsonl_path.exists(): + return 0.0 + last_ssim = 0.0 + with jsonl_path.open("r", encoding="utf-8") as handle: + for line in handle: + line = line.strip() + if not line: + continue + try: + record = json.loads(line) + last_ssim = float(record.get("ssim", 0.0)) + except json.JSONDecodeError: + continue + return last_ssim + + +def objective( + trial: optuna.Trial, + base_cfg: DictConfig, + epochs_per_trial: int, + output_root: Path, +) -> float: + """Single Optuna trial: train GAN with suggested params, return val SSIM.""" + params = {name: _suggest(trial, name, spec) for name, spec in SEARCH_SPACE.items()} + params["_trial_number"] = trial.number + + trial_cfg = _apply_overrides(base_cfg, params) + trial_cfg.training.epochs = epochs_per_trial + trial_cfg.paths.output_dir = str(output_root / f"trial_{trial.number:03d}") + + print(f"\n{'='*60}") + print(f"Trial {trial.number}: {params}") + print(f"{'='*60}") + + try: + train_gan(trial_cfg) + val_ssim = _extract_val_ssim(Path(trial_cfg.paths.output_dir)) + except Exception as exc: # pragma: no cover - eğitim hataları + print(f"Trial {trial.number} failed: {exc}") + # Optuna'a "kötü" değer döndür (prune et) + raise optuna.TrialPruned(f"Trial {trial.number} crashed: {exc}") from exc + + # MLflow'a trial parametrelerini log'la + with mlflow.start_run(nested=True, run_name=f"optuna-trial-{trial.number}"): + mlflow.log_params({k: v for k, v in params.items() if not k.startswith("_")}) + mlflow.log_metric("val_ssim", val_ssim) + mlflow.log_metric("trial_number", trial.number) + + return val_ssim + + +def run_optuna_search( + base_cfg: DictConfig, + n_trials: int = 20, + epochs_per_trial: int = 5, + output_root: Path | str = "outputs/optuna", + study_name: str = "pix2pix-hparam-search", +) -> optuna.Study: + """Run Optuna hyperparameter search. + + Args: + base_cfg: Base Hydra config (will be copied per trial). + n_trials: Number of trials to run. + epochs_per_trial: Epochs per trial (keep small for speed). + output_root: Where to save per-trial checkpoints. + study_name: Optuna study name. + + Returns: + Completed Optuna Study object. + """ + output_root = Path(output_root) + output_root.mkdir(parents=True, exist_ok=True) + + sampler = optuna.samplers.TPESampler(seed=int(base_cfg.seed)) + study = optuna.create_study( + study_name=study_name, + direction="maximize", # SSIM'i maksimize et + sampler=sampler, + ) + + # Parent MLflow run + mlflow.set_tracking_uri(base_cfg.mlflow.tracking_uri) + mlflow.set_experiment(f"{base_cfg.mlflow.experiment_name}-optuna") + + with mlflow.start_run(run_name=f"optuna-parent-{study_name}"): + mlflow.log_params( + { + "n_trials": n_trials, + "epochs_per_trial": epochs_per_trial, + "study_name": study_name, + } + ) + + study.optimize( + lambda trial: objective(trial, base_cfg, epochs_per_trial, output_root), + n_trials=n_trials, + show_progress_bar=False, + ) + + # Best params → MLflow + best_params = {k: v for k, v in study.best_params.items()} + mlflow.log_params({f"best_{k}": v for k, v in best_params.items()}) + mlflow.log_metric("best_val_ssim", study.best_value) + + # Best params → JSON dosyası + best_path = output_root / "best_params.json" + with best_path.open("w", encoding="utf-8") as handle: + json.dump( + { + "best_value": study.best_value, + "best_params": best_params, + "n_trials": n_trials, + }, + handle, + indent=2, + ensure_ascii=False, + ) + mlflow.log_artifact(str(best_path)) + + print(f"\n{'='*60}") + print(f"Best trial: #{study.best_trial.number}") + print(f"Best val SSIM: {study.best_value:.4f}") + print(f"Best params: {best_params}") + print(f"Saved to: {best_path}") + print(f"{'='*60}") + + return study + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Optuna hyperparameter search for Pix2Pix GAN", + ) + parser.add_argument("--n-trials", type=int, default=20, help="Number of trials") + parser.add_argument( + "--epochs-per-trial", + type=int, + default=5, + help="Epochs per trial (keep small for speed)", + ) + parser.add_argument( + "--output-root", + type=str, + default="outputs/optuna", + help="Where to save per-trial checkpoints", + ) + parser.add_argument( + "--study-name", + type=str, + default="pix2pix-hparam-search", + help="Optuna study name", + ) + return parser.parse_args() + + +def main() -> None: + """CLI entry point: load base config and run Optuna search.""" + args = parse_args() + + # Base config'i yükle (Hydra olmadan, doğrudan YAML) + from hydra import compose, initialize_config_dir + + config_dir = PROJECT_ROOT / "services" / "ml" / "conf" + with initialize_config_dir(config_dir=str(config_dir), version_base="1.3"): + base_cfg = compose(config_name="config") + + run_optuna_search( + base_cfg=base_cfg, + n_trials=args.n_trials, + epochs_per_trial=args.epochs_per_trial, + output_root=args.output_root, + study_name=args.study_name, + ) + + +if __name__ == "__main__": + main()