app/engine/validation.py

import os
import torch
import torch.nn as nn
import numpy as np

# Same architecture as training script
class AstroNet1D(nn.Module):
    def __init__(self):
        super(AstroNet1D, self).__init__()
        self.conv1 = nn.Conv1d(1, 16, kernel_size=5, stride=1, padding=2)
        self.conv2 = nn.Conv1d(16, 32, kernel_size=5, stride=2, padding=2)
        self.conv3 = nn.Conv1d(32, 64, kernel_size=5, stride=2, padding=2)
        
        self.pool = nn.MaxPool1d(2)
        self.relu = nn.ReLU()
        self.dropout = nn.Dropout(0.3)
        
        self.fc1 = nn.Linear(64 * 31, 128)
        self.fc2 = nn.Linear(128, 1)
        self.sigmoid = nn.Sigmoid()

    def forward(self, x):
        x = self.relu(self.pool(self.conv1(x)))
        x = self.relu(self.pool(self.conv2(x)))
        x = self.relu(self.pool(self.conv3(x)))
        
        x = x.view(x.size(0), -1)
        x = self.dropout(self.relu(self.fc1(x)))
        x = self.sigmoid(self.fc2(x))
        return x

# Singleton for loading the model once
_MODEL = None

def load_model():
    global _MODEL
    if _MODEL is not None:
        return _MODEL
        
    model_path = os.path.join(os.path.dirname(__file__), "..", "..", "data_cache", "models", "astronet_v1.pt")
    if not os.path.exists(model_path):
        return None
        
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    _MODEL = AstroNet1D().to(device)
    _MODEL.load_state_dict(torch.load(model_path, map_location=device, weights_only=True))
    _MODEL.eval()
    return _MODEL

def bin_lightcurve(phase, flux, bins=1000):
    """Sorts and bins phase-folded data into a fixed size array of 1000"""
    # Sort by phase
    sort_idx = np.argsort(phase)
    p_sorted = np.array(phase)[sort_idx]
    f_sorted = np.array(flux)[sort_idx]
    
    # Create bin edges from min to max phase
    bins_edges = np.linspace(np.min(p_sorted), np.max(p_sorted), bins + 1)
    
    # Digitize phase
    bin_indices = np.digitize(p_sorted, bins_edges)
    
    binned_flux = np.ones(bins)
    for i in range(1, bins + 1):
        mask = bin_indices == i
        if np.any(mask):
            binned_flux[i-1] = np.median(f_sorted[mask])
            
    # Normalize to mean 1
    if np.nanmean(binned_flux) != 0:
        binned_flux /= np.nanmean(binned_flux)
    
    # Fill any NaNs remaining (empty bins) with 1.0 (baseline)
    binned_flux[np.isnan(binned_flux)] = 1.0
        
    return binned_flux

def validate_candidate(phase: list, flux: list):
    """
    CNN Validation Layer using PyTorch AstroNet V1.
    """
    model = load_model()
    if model is None:
        return {
            "status": "Unavailable",
            "cnn_confidence": None,
            "message": "AstroNet model weights not found in data_cache/models."
        }
        
    try:
        # 1. Preprocess: Bin into 1000 elements array
        binned_flux = bin_lightcurve(phase, flux, bins=1000)
        
        # 2. Convert to tensor: (batch=1, channels=1, length=1000)
        device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        tensor = torch.tensor(binned_flux, dtype=torch.float32).unsqueeze(0).unsqueeze(0).to(device)
        
        # 3. Inference
        with torch.no_grad():
            output = model(tensor)
            
        confidence = output.item() * 100.0 # Convert 0-1 to 0-100%
        
        return {
            "status": "PASS" if confidence > 50 else "FAIL",
            "cnn_confidence": float(confidence),
            "message": f"AstroNet predicts {confidence:.1f}% confidence of planetary transit."
        }
    except Exception as e:
        print(f"Validation error: {e}")
        return {
            "status": "Unavailable",
            "cnn_confidence": None,
            "message": "Validation encountered an error."
        }