data_manager.py

import os
import random
import numpy as np
from PIL import Image
from collections import defaultdict
from sklearn.model_selection import StratifiedShuffleSplit
import torch
from torch.utils.data import Dataset
from torchvision import transforms
from torchvision.transforms import InterpolationMode, functional as F

# Constants
IMG_EXTS = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}
DEFAULT_MEAN = [0.485, 0.456, 0.406]
DEFAULT_STD = [0.229, 0.224, 0.225]

def set_seed(seed=42):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)

class SnakeDataset(Dataset):
    def __init__(self, image_paths, labels, transform=None):
        self.image_paths = image_paths
        self.labels = labels
        self.transform = transform

    def __len__(self):
        return len(self.image_paths)

    def __getitem__(self, idx):
        img_path = self.image_paths[idx]
        label = self.labels[idx]
        
        try:
            image = Image.open(img_path).convert('RGB')
            if self.transform:
                image = self.transform(image)
            return image, label
        except Exception as e:
            print(f"Error loading {img_path}: {e}")
            # Return a dummy tensor or handle appropriately. 
            # For now, just reraise to be safe, or return None and handle in collate_fn (advanced)
            # Re-raising is safer for debugging.
            raise e

def scan_dataset(root_dir):
    """
    Scans the directory for images.
    Structure expected: root_dir/class_name/image.jpg
    """
    species_to_files = defaultdict(list)
    for class_name in os.listdir(root_dir):
        class_dir = os.path.join(root_dir, class_name)
        if not os.path.isdir(class_dir):
            continue
        
        for f in os.listdir(class_dir):
            if os.path.splitext(f.lower())[1] in IMG_EXTS:
                species_to_files[class_name].append(os.path.join(class_dir, f))
    return species_to_files

def filter_and_split_data(root_dir, threshold, seed=42):
    """
    Filters classes with < threshold images.
    Splits remaining data:
      1. 80% (Train+Val) / 20% (Test)
      2. Of the 80% Train+Val: 80% Train / 20% Val (Which is 64% total Train, 16% total Val)
    Returns:
      (train_paths, train_labels), (val_paths, val_labels), (test_paths, test_labels), class_to_idx
    """
    species_to_files = scan_dataset(root_dir)
    
    # Filter by threshold
    valid_species = sorted([sp for sp, files in species_to_files.items() if len(files) >= threshold])
    if not valid_species:
        return None, None, None, None

    class_to_idx = {sp: i for i, sp in enumerate(valid_species)}
    
    all_paths = []
    all_labels = []
    
    for sp in valid_species:
        files = species_to_files[sp]
        all_paths.extend(files)
        all_labels.extend([class_to_idx[sp]] * len(files))
        
    all_paths = np.array(all_paths)
    all_labels = np.array(all_labels)

    # First Split: 80% Train+Val, 20% Test
    sss_test = StratifiedShuffleSplit(n_splits=1, test_size=0.2, random_state=seed)
    train_val_idx, test_idx = next(sss_test.split(all_paths, all_labels))
    
    X_train_val, y_train_val = all_paths[train_val_idx], all_labels[train_val_idx]
    X_test, y_test = all_paths[test_idx], all_labels[test_idx]
    
    # Second Split: Of Train+Val, split 80% Train, 20% Val
    # Note: 20% of 80% is 16% of total. 80% of 80% is 64% of total.
    # The prompt says: "training portion was further split 8:2 into training and validation"
    
    sss_val = StratifiedShuffleSplit(n_splits=1, test_size=0.2, random_state=seed)
    train_idx, val_idx = next(sss_val.split(X_train_val, y_train_val))
    
    X_train, y_train = X_train_val[train_idx], y_train_val[train_idx]
    X_val, y_val = X_train_val[val_idx], y_train_val[val_idx]
    
    return (X_train, y_train), (X_val, y_val), (X_test, y_test), class_to_idx

class RandomDiscreteTransform:
    """Apply a transform with a given probability."""
    def __init__(self, transform, p=0.5):
        self.transform = transform
        self.p = p

    def __call__(self, img):
        if random.random() < self.p:
            return self.transform(img)
        return img

def get_transforms(intensity='none', input_size=224):
    """
    Returns training and validation transforms.
    Intensity profiles: none, low, medium, high.
    Base augmentation params (from previous codebase logic):
      - rotation: 10
      - shifts: 0.1
      - zoom: 0.1
      - shear: 0.1
      - flip: True (Horizontal)
    
    Multipliers:
      - low: 0.5
      - medium: 1.0
      - high: 1.5
    """
    
    # Validation/Test is always just Resize + Norm
    val_transforms = transforms.Compose([
        transforms.Resize((input_size, input_size), interpolation=InterpolationMode.BICUBIC),
        transforms.ToTensor(),
        transforms.Normalize(mean=DEFAULT_MEAN, std=DEFAULT_STD)
    ])
    
    if intensity == 'none':
        return val_transforms, val_transforms

    # Base params
    base_deg = 10
    base_trans = 0.1
    base_scale = 0.1
    base_shear = 10 # Approx 0.1 shear ~ 10 degrees or 0.1 value in affine
    # Note: torchvision shear is in degrees. 0.1 shear in Keras is tricky, usually means shear intensity. 
    # Let's map 0.1 shear to ~10 degrees for simplicity or keep small. 
    # Keras RandomShear uses 'shear_intensity' (shear angle in radians is not exactly it, it's a shear transformation matrix).
    # Actually, Keras: "Shear angle in counter-clockwise direction in degrees." -> Wait, Keras doc says 'shear_range': Float. Shear Intensity (Shear angle in counter-clockwise direction in degrees)
    # So 0.1 in older Keras might have been radians? No, usually degrees if large, but 0.1 degree is tiny.
    # Let's assume standard "shear" usually implies a visible distortion. 
    # In the provided `new_normal_experiment.py`, `shear_range=0.1`.
    # In `new_train_transformer...py`, it mimics this.
    # I will stick to the logic: shear=0.1 -> maybe 10 degrees? Or 0.1 radians (~5.7 deg). Let's use proportional scaling.
    
    multipliers = {'low': 0.5, 'medium': 1.0, 'high': 1.5}
    m = multipliers.get(intensity, 1.0)
    
    # Keras semantics from DeiT file
    deg = 10 * m
    tx = 0.1 * m
    ty = 0.1 * m
    zm = 0.1 * m
    shr = 0.1 * m # Radians
    sh_deg = np.degrees(shr) if shr > 0 else 0.0
    
    # Validation/Test is always just Resize + Norm
    val_transforms = transforms.Compose([
        transforms.Resize((input_size, input_size), interpolation=InterpolationMode.BILINEAR),
        transforms.ToTensor(),
        transforms.Normalize(mean=DEFAULT_MEAN, std=DEFAULT_STD)
    ])
    
    if intensity == 'none':
        return val_transforms, val_transforms

    # Train transforms with exact Keras emulation
    translate = (max(0.0, min(tx, 0.499)), max(0.0, min(ty, 0.499)))
    scale = (max(0.0, 1.0 - zm), 1.0 + zm)
    
    # Emulate nearest fill mode: edge-pad -> affine -> crop
    pad_px = int(round(0.12 * input_size))
    
    train_ops = [
        transforms.Resize((input_size, input_size), interpolation=InterpolationMode.BILINEAR),
        transforms.RandomHorizontalFlip(p=0.5),
        transforms.Pad(pad_px, padding_mode="edge"),
        transforms.RandomAffine(
            degrees=(-deg, deg),
            translate=translate,
            scale=scale,
            shear=(-sh_deg, sh_deg),
            interpolation=InterpolationMode.BILINEAR
        ),
        transforms.CenterCrop((input_size, input_size)),
        transforms.ToTensor(),
        transforms.Normalize(mean=DEFAULT_MEAN, std=DEFAULT_STD)
    ]
    
    train_transforms = transforms.Compose(train_ops)
    
    return train_transforms, val_transforms