Spaces:
Running on Zero
Running on Zero
File size: 10,180 Bytes
be88765 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 | """
Preview utilities for data splitting in training sample generation.
This module contains functions for previewing what data splits would look like
before actually processing the data.
"""
import random
import logging
from typing import List, Tuple, Optional, Dict
logger = logging.getLogger(__name__)
def preview_data_splits(sequence_samples: Dict[str, int],
train_ratio: float,
split_by_sequence: bool,
random_seed: int,
data_loader=None,
val_sequences: Optional[List[str]] = None):
"""Preview what the data splits would look like without actually creating them."""
if not sequence_samples:
return
# Check for predefined splits
using_predefined_splits = False
predefined_val_sequences = []
if val_sequences is None and data_loader is not None and hasattr(data_loader, 'predefined_split_available') and data_loader.predefined_split_available:
predefined_val_sequences = data_loader.get_predefined_val_sequences()
# Only use predefined sequences that would be processed
predefined_val_sequences = [seq for seq in predefined_val_sequences if seq in sequence_samples]
if predefined_val_sequences:
val_sequences = predefined_val_sequences
using_predefined_splits = True
if using_predefined_splits:
_preview_predefined_splits(sequence_samples, val_sequences)
elif val_sequences is not None:
_preview_manual_splits(sequence_samples, val_sequences)
else:
_preview_automatic_splits(sequence_samples, train_ratio, split_by_sequence, random_seed)
def _preview_predefined_splits(sequence_samples: Dict[str, int], val_sequences: List[str]):
"""Preview predefined splits."""
logger.info("DATA SPLITTING PREVIEW (predefined):")
logger.info(" Using predefined validation sequences from dataset")
# Calculate splits based on predefined validation sequences
train_sequences = [seq for seq in sequence_samples.keys() if seq not in val_sequences]
val_sequences_actual = [seq for seq in sequence_samples.keys() if seq in val_sequences]
train_samples_count = sum(sequence_samples[seq] for seq in train_sequences)
val_samples_count = sum(sequence_samples[seq] for seq in val_sequences_actual)
total_samples = train_samples_count + val_samples_count
actual_train_ratio = train_samples_count / total_samples if total_samples > 0 else 0
logger.info(f" Split method: predefined")
logger.info(f" Predefined train samples: {train_samples_count}")
logger.info(f" Predefined val samples: {val_samples_count}")
logger.info(f" Predefined actual ratio: {actual_train_ratio:.3f}")
logger.info(f" Train sequences: {sorted(train_sequences)}")
logger.info(f" Val sequences: {sorted(val_sequences_actual)}")
logger.info(f" Data leakage: None (sequences kept separate)")
# Show detailed sequence assignments
logger.info(" Sequence assignments:")
for sequence in sorted(sequence_samples.keys()):
num_samples = sequence_samples[sequence]
assignment = "VAL" if sequence in val_sequences_actual else "TRAIN"
logger.info(f" {sequence}: {num_samples:3d} samples → {assignment}")
def _preview_manual_splits(sequence_samples: Dict[str, int], val_sequences: List[str]):
"""Preview manual splits."""
logger.info("DATA SPLITTING PREVIEW (manual):")
logger.info(f" Using manually specified validation sequences: {val_sequences}")
# Calculate splits based on manual validation sequences
train_sequences = [seq for seq in sequence_samples.keys() if seq not in val_sequences]
val_sequences_actual = [seq for seq in sequence_samples.keys() if seq in val_sequences]
train_samples_count = sum(sequence_samples[seq] for seq in train_sequences)
val_samples_count = sum(sequence_samples[seq] for seq in val_sequences_actual)
total_samples = train_samples_count + val_samples_count
actual_train_ratio = train_samples_count / total_samples if total_samples > 0 else 0
logger.info(f" Split method: manual")
logger.info(f" Manual train samples: {train_samples_count}")
logger.info(f" Manual val samples: {val_samples_count}")
logger.info(f" Manual actual ratio: {actual_train_ratio:.3f}")
logger.info(f" Train sequences: {sorted(train_sequences)}")
logger.info(f" Val sequences: {sorted(val_sequences_actual)}")
logger.info(f" Data leakage: None (sequences kept separate)")
# Show detailed sequence assignments
logger.info(" Sequence assignments:")
for sequence in sorted(sequence_samples.keys()):
num_samples = sequence_samples[sequence]
assignment = "VAL" if sequence in val_sequences_actual else "TRAIN"
logger.info(f" {sequence}: {num_samples:3d} samples → {assignment}")
def _preview_automatic_splits(sequence_samples: Dict[str, int], train_ratio: float, split_by_sequence: bool, random_seed: int):
"""Preview automatic splits."""
split_method = "sequence-based" if split_by_sequence else "sample-based"
logger.info(f"DATA SPLITTING PREVIEW ({split_method}):")
# Convert sequence_samples (counts) to mock sample lists for splitting simulation
mock_sequence_samples = {}
for sequence, num_samples in sequence_samples.items():
mock_sequence_samples[sequence] = [f"mock_sample_{i}" for i in range(num_samples)]
if split_by_sequence:
# Use the actual sequence-based splitting logic
train_samples, val_samples, train_sequences, val_sequences = _preview_sequence_split(
mock_sequence_samples, train_ratio, random_seed
)
else:
# Simulate sample-based splitting
train_samples, val_samples = _preview_sample_split(
mock_sequence_samples, train_ratio, random_seed
)
train_sequences = list(sequence_samples.keys()) # All sequences appear in both
val_sequences = list(sequence_samples.keys())
total_samples = sum(sequence_samples.values())
actual_train_ratio = len(train_samples) / total_samples if total_samples > 0 else 0
logger.info(f" Split method: {split_method}")
logger.info(f" Target train ratio: {train_ratio:.3f}")
logger.info(f" Predicted train samples: {len(train_samples)}")
logger.info(f" Predicted val samples: {len(val_samples)}")
logger.info(f" Predicted actual ratio: {actual_train_ratio:.3f}")
logger.info(f" Ratio deviation: ±{abs(actual_train_ratio - train_ratio):.3f}")
if split_by_sequence:
logger.info(f" Train sequences: {sorted(train_sequences)}")
logger.info(f" Val sequences: {sorted(val_sequences)}")
logger.info(f" Data leakage: None (sequences kept separate)")
# Show detailed sequence assignments
logger.info(" Sequence assignments:")
for sequence in sorted(sequence_samples.keys()):
num_samples = sequence_samples[sequence]
assignment = "TRAIN" if sequence in train_sequences else "VAL"
logger.info(f" {sequence}: {num_samples:3d} samples → {assignment}")
else:
# Count potential data leakage for sample-based splitting
logger.info(f" Potential data leakage: All {len(sequence_samples)} sequences")
logger.info(f" (samples from each sequence likely in both train/val)")
def _preview_sequence_split(sequence_samples: Dict[str, List[str]],
train_ratio: float,
random_seed: int) -> Tuple[List[str], List[str], List[str], List[str]]:
"""Preview sequence-based splitting (simplified version of the actual logic)."""
import random
random.seed(random_seed)
# Calculate total samples and target
total_samples = sum(len(samples) for samples in sequence_samples.values())
target_train_samples = int(total_samples * train_ratio)
# Create sequence info and sort by size
sequences = list(sequence_samples.items())
sequences.sort(key=lambda x: len(x[1]), reverse=True)
# Add some randomness while keeping deterministic behavior
random.shuffle(sequences)
sequences.sort(key=lambda x: len(x[1]), reverse=True)
# Greedy assignment
train_samples = []
val_samples = []
train_sequences = []
val_sequences = []
current_train_count = 0
for seq_name, samples in sequences:
num_samples = len(samples)
if current_train_count + num_samples <= target_train_samples:
train_samples.extend(samples)
train_sequences.append(seq_name)
current_train_count += num_samples
else:
# Check which assignment gives better ratio
train_ratio_if_added = (current_train_count + num_samples) / total_samples
train_ratio_if_not = current_train_count / total_samples
diff_if_added = abs(train_ratio_if_added - train_ratio)
diff_if_not = abs(train_ratio_if_not - train_ratio)
if diff_if_added < diff_if_not:
train_samples.extend(samples)
train_sequences.append(seq_name)
current_train_count += num_samples
else:
val_samples.extend(samples)
val_sequences.append(seq_name)
return train_samples, val_samples, train_sequences, val_sequences
def _preview_sample_split(sequence_samples: Dict[str, List[str]],
train_ratio: float,
random_seed: int) -> Tuple[List[str], List[str]]:
"""Preview sample-based splitting."""
import random
random.seed(random_seed)
# Collect all samples
all_samples = []
for samples in sequence_samples.values():
all_samples.extend(samples)
# Calculate split
num_train = int(len(all_samples) * train_ratio)
train_samples = all_samples[:num_train]
val_samples = all_samples[num_train:]
return train_samples, val_samples |