"""Build pseudo-sequences from the extracted GTSRB archive and precompute VGG16 features. The archive is treated as an ordered image corpus, not as a video dataset: - images are read from `archive/Train//` - files are sorted within each class folder - fixed-length windows are sliced with stride 2 - each sequence is stored with metadata as a `.npz` - VGG16 features are then computed for each saved sequence """ from __future__ import annotations import json import shutil import tempfile from pathlib import Path from typing import Dict, Iterable, List, Tuple import numpy as np from PIL import Image from src.config import DatasetConfig from src.detection.feature_extractor_vgg16 import VGG16FeatureExtractor ARCHIVE_TRAIN_DIR = Path("archive/Train") OUTPUT_SEQ_DIR = Path("data/preprocessed_sequences") FEATURES_OUTPUT_DIR = Path("cache/vgg16_sequence_features") SPLIT = "train" SEQ_LEN = DatasetConfig.SEQUENCE_LENGTH STRIDE = DatasetConfig.FRAME_STRIDE RESIZE = DatasetConfig.IMAGE_SIZE FPS = 30.0 def _split_counts(total: int, train_ratio: float, val_ratio: float) -> Tuple[int, int, int]: train_count = int(total * train_ratio) val_count = int(total * val_ratio) test_count = total - train_count - val_count if total >= 3: if train_count == 0: train_count = 1 if val_count == 0: val_count = 1 test_count = total - train_count - val_count if test_count <= 0: test_count = 1 train_count = max(train_count - 1, 1) val_count = max(val_count - 1, 1) return train_count, val_count, test_count def _load_image(path: Path) -> np.ndarray: image = Image.open(path).convert("RGB") if RESIZE: image = image.resize(RESIZE) return np.asarray(image, dtype=np.uint8) class SequencePreprocessor: """Create pseudo-sequences from a class folder of ordered still images.""" def __init__(self, output_dir: Path = OUTPUT_SEQ_DIR, seq_length: int = SEQ_LEN, stride: int = STRIDE): self.output_dir = Path(output_dir) self.seq_length = seq_length self.stride = stride def process_class_folder(self, class_dir: Path, class_label: int, split: str = SPLIT) -> int: class_dir = Path(class_dir) out_dir = self.output_dir / split / f"class_{class_label:02d}" out_dir.mkdir(parents=True, exist_ok=True) image_files = sorted( [path for path in class_dir.iterdir() if path.suffix.lower() in (".ppm", ".png", ".jpg", ".jpeg")] ) count = 0 for start in range(0, len(image_files) - self.seq_length + 1, self.stride or 1): end = start + self.seq_length sequence_paths = image_files[start:end] frames = [_load_image(path) for path in sequence_paths] frames_array = np.stack(frames) metadata = { "sequence_id": f"class_{class_label:02d}_{start:06d}", "video_source": class_dir.name, "start_frame": start, "end_frame": end - 1, "frame_count": self.seq_length, "class_label": class_label, "timestamps": [index / FPS for index in range(start, end)], "fps": FPS, } out_path = out_dir / f"{metadata['sequence_id']}.npz" np.savez_compressed(out_path, frames=frames_array, metadata=json.dumps(metadata)) count += 1 print(f"Saved {count} sequences for class {class_label} to {out_dir}") return count def preprocess_all(self, archive_dir: Path = ARCHIVE_TRAIN_DIR, split: str = SPLIT) -> int: archive_dir = Path(archive_dir) total = 0 for class_dir in sorted([path for path in archive_dir.iterdir() if path.is_dir()]): try: class_label = int(class_dir.name) except ValueError: continue total += self.process_class_folder(class_dir, class_label, split=split) print(f"Total sequences created: {total}") return total def _load_metadata_value(metadata_value): if isinstance(metadata_value, np.ndarray): metadata_value = metadata_value.item() if isinstance(metadata_value, bytes): metadata_value = metadata_value.decode("utf-8") return json.loads(metadata_value) class SequenceFeaturePrecomputer: """Precompute VGG16 features for saved sequence `.npz` files.""" def __init__(self, sequence_dir: Path = OUTPUT_SEQ_DIR, output_dir: Path = FEATURES_OUTPUT_DIR, device: str = "cuda"): self.sequence_dir = Path(sequence_dir) self.output_dir = Path(output_dir) self.device = device self.extractor = VGG16FeatureExtractor(device=device) def precompute_sequences(self, split: str = SPLIT) -> int: split_dir = self.sequence_dir / split if not split_dir.exists(): print(f"Sequence directory not found: {split_dir}") return 0 sequence_files = sorted(split_dir.rglob("*.npz")) if not sequence_files: print(f"No sequence files found in {split_dir}") return 0 saved = 0 for seq_file in sequence_files: data = np.load(seq_file, allow_pickle=False) frames = data["frames"] metadata = _load_metadata_value(data["metadata"]) if "metadata" in data else {} frame_images = [Image.fromarray(frame.astype(np.uint8)) for frame in frames] features = self.extractor.extract_sequence(frame_images).astype(np.float32) class_folder = seq_file.parent.name output_dir = self.output_dir / split / class_folder output_dir.mkdir(parents=True, exist_ok=True) output_file = output_dir / f"{seq_file.stem}_features.npz" np.savez_compressed(output_file, features=features, metadata=json.dumps(metadata)) saved += 1 print(f"Saved {saved} feature files to {self.output_dir / split}") return saved def precompute_all_splits(self, splits: Iterable[str] = ("train", "val", "test")) -> Dict[str, int]: saved_by_split: Dict[str, int] = {} for split in splits: saved_by_split[split] = self.precompute_sequences(split=split) return saved_by_split def regenerate_grouped_feature_splits( features_root: Path = FEATURES_OUTPUT_DIR, input_split: str = "train", sequence_length: int = SEQ_LEN, group_size_sequences: int = 5, train_ratio: float = 0.7, val_ratio: float = 0.15, seed: int = 42, ) -> Dict[str, int]: """Create leakage-safe train/val/test feature splits from a single cached split. Steps: - keep only non-overlapping windows (`start_frame % sequence_length == 0`) - group windows by source segment (video_source + chunk id) - split by group so related sequences stay in one split """ features_root = Path(features_root) source_dir = features_root / input_split if not source_dir.exists(): raise FileNotFoundError(f"Input split not found: {source_dir}") stage_dir: Path | None = None read_root = source_dir if input_split in {"train", "val", "test"}: stage_dir = Path(tempfile.mkdtemp(prefix="safe_split_stage_")) for feature_file in sorted(source_dir.rglob("*_features.npz")): rel_path = feature_file.relative_to(source_dir) target_path = stage_dir / rel_path target_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(feature_file, target_path) read_root = stage_dir split_dirs = { "train": features_root / "train", "val": features_root / "val", "test": features_root / "test", } for split_dir in split_dirs.values(): if split_dir.exists(): shutil.rmtree(split_dir) split_dir.mkdir(parents=True, exist_ok=True) entries_by_class: Dict[int, List[Tuple[Path, Dict]]] = {} feature_files = sorted(read_root.rglob("*_features.npz")) for feature_file in feature_files: data = np.load(feature_file, allow_pickle=False) metadata = _load_metadata_value(data["metadata"]) if "metadata" in data else {} class_label = int(metadata.get("class_label", int(feature_file.parent.name.split("_")[1]))) start_frame = int(metadata.get("start_frame", 0)) # Drop overlapping windows to prevent cross-split frame reuse leakage. if sequence_length > 0 and (start_frame % sequence_length) != 0: continue entries_by_class.setdefault(class_label, []).append((feature_file, metadata)) rng = np.random.default_rng(seed) copied_counts = {"train": 0, "val": 0, "test": 0} for class_label, entries in sorted(entries_by_class.items()): groups: Dict[str, List[Tuple[Path, Dict]]] = {} for feature_file, metadata in entries: start_frame = int(metadata.get("start_frame", 0)) source = str(metadata.get("video_source", f"class_{class_label:02d}")) segment_id = start_frame // max(sequence_length * group_size_sequences, 1) group_key = f"{source}::segment_{segment_id}" groups.setdefault(group_key, []).append((feature_file, metadata)) group_keys = list(groups.keys()) rng.shuffle(group_keys) train_n, val_n, _ = _split_counts(len(group_keys), train_ratio, val_ratio) split_by_group: Dict[str, str] = {} for index, group_key in enumerate(group_keys): if index < train_n: split_by_group[group_key] = "train" elif index < train_n + val_n: split_by_group[group_key] = "val" else: split_by_group[group_key] = "test" for group_key, group_entries in groups.items(): split = split_by_group[group_key] class_dir = split_dirs[split] / f"class_{class_label:02d}" class_dir.mkdir(parents=True, exist_ok=True) for feature_file, _metadata in group_entries: shutil.copy2(feature_file, class_dir / feature_file.name) copied_counts[split] += 1 print("[Split] Regenerated grouped non-overlapping feature cache:") print(f" train: {copied_counts['train']}") print(f" val: {copied_counts['val']}") print(f" test: {copied_counts['test']}") if stage_dir and stage_dir.exists(): shutil.rmtree(stage_dir, ignore_errors=True) return copied_counts def main() -> None: preprocessor = SequencePreprocessor(output_dir=OUTPUT_SEQ_DIR, seq_length=SEQ_LEN, stride=STRIDE) sequence_count = preprocessor.preprocess_all() if sequence_count == 0: print("No sequences created. Check that archive/Train has class folders with images.") return print("Precomputing VGG16 features for sequences...") precomputer = SequenceFeaturePrecomputer(sequence_dir=OUTPUT_SEQ_DIR, output_dir=FEATURES_OUTPUT_DIR) precomputer.precompute_sequences(split=SPLIT) print("Feature precomputation complete.") if __name__ == "__main__": main()