Buckets:
| import torch | |
| from torch.utils.data import Dataset, DataLoader | |
| import torchvision.transforms as transforms | |
| from PIL import Image | |
| import json | |
| import os | |
| from typing import List, Dict, Tuple, Optional, Any | |
| import random | |
| from collections import Counter | |
| import numpy as np | |
| from ..utils.base_classes import BaseDataset, VQASample | |
| class VQADataset(BaseDataset): | |
| def __init__( | |
| self, | |
| data_dir: str, | |
| split: str = "train", | |
| image_size: int = 224, | |
| max_question_length: int = 20, | |
| ): | |
| self.data_dir = data_dir | |
| self.split = split | |
| self.image_size = image_size | |
| self.max_question_length = max_question_length | |
| self.image_transforms = transforms.Compose( | |
| [ | |
| transforms.Resize((image_size, image_size)), | |
| transforms.ToTensor(), | |
| transforms.Normalize( | |
| mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225] | |
| ), | |
| ] | |
| ) | |
| self.samples = self._load_samples() | |
| self.question_types = self._get_question_types() | |
| self.answer_vocab = self._get_answer_vocabulary() | |
| def _load_samples(self) -> List[VQASample]: | |
| samples = [] | |
| sample_data = [ | |
| { | |
| "image_id": "000001", | |
| "image_path": "images/train/000001.jpg", | |
| "question": "What color is the car?", | |
| "answer": "red", | |
| "question_type": "attribute", | |
| }, | |
| { | |
| "image_id": "000002", | |
| "image_path": "images/train/000002.jpg", | |
| "question": "How many people are there?", | |
| "answer": "3", | |
| "question_type": "counting", | |
| }, | |
| { | |
| "image_id": "000003", | |
| "image_path": "images/train/000003.jpg", | |
| "question": "Is there a dog in the image?", | |
| "answer": "yes", | |
| "question_type": "yes_no", | |
| }, | |
| { | |
| "image_id": "000004", | |
| "image_path": "images/train/000004.jpg", | |
| "question": "What is the person doing?", | |
| "answer": "walking", | |
| "question_type": "activity", | |
| }, | |
| { | |
| "image_id": "000005", | |
| "image_path": "images/train/000005.jpg", | |
| "question": "Where is the ball?", | |
| "answer": "on the ground", | |
| "question_type": "spatial", | |
| }, | |
| ] | |
| for data in sample_data: | |
| sample = VQASample( | |
| image_id=data["image_id"], | |
| image_path=data["image_path"], | |
| question=data["question"], | |
| answer=data["answer"], | |
| question_type=data["question_type"], | |
| confidence=1.0, | |
| metadata={"split": self.split}, | |
| ) | |
| samples.append(sample) | |
| return samples | |
| def _get_question_types(self) -> List[str]: | |
| types = set() | |
| for sample in self.samples: | |
| if sample.question_type: | |
| types.add(sample.question_type) | |
| return list(types) | |
| def _get_answer_vocabulary(self) -> List[str]: | |
| answers = [sample.answer for sample in self.samples] | |
| return list(set(answers)) | |
| def __len__(self) -> int: | |
| return len(self.samples) | |
| def __getitem__(self, idx: int) -> VQASample: | |
| return self.samples[idx] | |
| def get_sample_by_id(self, sample_id: str) -> Optional[VQASample]: | |
| for sample in self.samples: | |
| if sample.image_id == sample_id: | |
| return sample | |
| return None | |
| def get_question_types(self) -> List[str]: | |
| return self.question_types | |
| def get_answer_vocabulary(self) -> List[str]: | |
| return self.answer_vocab | |
| class SyntheticVQADataset(BaseDataset): | |
| def __init__(self, num_samples: int = 1000, image_size: int = 224): | |
| self.num_samples = num_samples | |
| self.image_size = image_size | |
| self.image_transforms = transforms.Compose( | |
| [ | |
| transforms.Resize((image_size, image_size)), | |
| transforms.ToTensor(), | |
| transforms.Normalize( | |
| mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225] | |
| ), | |
| ] | |
| ) | |
| self.samples = self._generate_samples() | |
| self.question_types = self._get_question_types() | |
| self.answer_vocab = self._get_answer_vocabulary() | |
| def _generate_samples(self) -> List[VQASample]: | |
| samples = [] | |
| question_templates = { | |
| "object_recognition": [ | |
| "What is in the image?", | |
| "What object do you see?", | |
| "What is this?", | |
| ], | |
| "attribute": [ | |
| "What color is the {object}?", | |
| "How big is the {object}?", | |
| "What shape is the {object}?", | |
| ], | |
| "counting": [ | |
| "How many {object}s are there?", | |
| "Count the {object}s", | |
| "What is the number of {object}s?", | |
| ], | |
| "spatial": [ | |
| "Where is the {object}?", | |
| "What is to the left of the {object}?", | |
| "What is above the {object}?", | |
| ], | |
| "activity": [ | |
| "What is the person doing?", | |
| "What activity is happening?", | |
| "What action is being performed?", | |
| ], | |
| "yes_no": [ | |
| "Is there a {object} in the image?", | |
| "Can you see a {object}?", | |
| "Does the image contain a {object}?", | |
| ], | |
| } | |
| objects = ["car", "person", "dog", "house", "tree", "ball", "cat", "bird"] | |
| colors = ["red", "blue", "green", "yellow", "white", "black", "brown", "gray"] | |
| sizes = ["small", "medium", "large", "tiny", "huge"] | |
| shapes = ["round", "square", "rectangular", "oval", "triangular"] | |
| activities = [ | |
| "walking", | |
| "running", | |
| "sitting", | |
| "standing", | |
| "jumping", | |
| "sleeping", | |
| ] | |
| spatial_relations = ["left", "right", "above", "below", "near", "far"] | |
| for i in range(self.num_samples): | |
| question_type = random.choice(list(question_templates.keys())) | |
| template = random.choice(question_templates[question_type]) | |
| if question_type == "object_recognition": | |
| question = template | |
| answer = random.choice(objects) | |
| elif question_type == "attribute": | |
| obj = random.choice(objects) | |
| question = template.format(object=obj) | |
| if "color" in template: | |
| answer = random.choice(colors) | |
| elif "big" in template: | |
| answer = random.choice(sizes) | |
| elif "shape" in template: | |
| answer = random.choice(shapes) | |
| else: | |
| answer = random.choice(colors) | |
| elif question_type == "counting": | |
| obj = random.choice(objects) | |
| question = template.format(object=obj) | |
| count = random.randint(1, 5) | |
| answer = str(count) | |
| elif question_type == "spatial": | |
| obj = random.choice(objects) | |
| question = template.format(object=obj) | |
| answer = random.choice(spatial_relations) | |
| elif question_type == "activity": | |
| question = template | |
| answer = random.choice(activities) | |
| elif question_type == "yes_no": | |
| obj = random.choice(objects) | |
| question = template.format(object=obj) | |
| answer = random.choice(["yes", "no"]) | |
| sample = VQASample( | |
| image_id=f"synthetic_{i:06d}", | |
| image_path=f"synthetic_images/{i:06d}.jpg", | |
| question=question, | |
| answer=answer, | |
| question_type=question_type, | |
| confidence=1.0, | |
| metadata={"synthetic": True, "generated_index": i}, | |
| ) | |
| samples.append(sample) | |
| return samples | |
| def _get_question_types(self) -> List[str]: | |
| types = set() | |
| for sample in self.samples: | |
| if sample.question_type: | |
| types.add(sample.question_type) | |
| return list(types) | |
| def _get_answer_vocabulary(self) -> List[str]: | |
| answers = [sample.answer for sample in self.samples] | |
| return list(set(answers)) | |
| def __len__(self) -> int: | |
| return len(self.samples) | |
| def __getitem__(self, idx: int) -> VQASample: | |
| return self.samples[idx] | |
| def get_sample_by_id(self, sample_id: str) -> Optional[VQASample]: | |
| for sample in self.samples: | |
| if sample.image_id == sample_id: | |
| return sample | |
| return None | |
| def get_question_types(self) -> List[str]: | |
| return self.question_types | |
| def get_answer_vocabulary(self) -> List[str]: | |
| return self.answer_vocab | |
| class VQADataModule: | |
| def __init__(self, config: Dict[str, Any]): | |
| self.config = config | |
| self.batch_size = config.get("batch_size", 32) | |
| self.num_workers = config.get("num_workers", 4) | |
| self.train_dataset = None | |
| self.val_dataset = None | |
| self.test_dataset = None | |
| def setup_datasets(self): | |
| data_config = self.config.get("data", {}) | |
| if data_config.get("use_synthetic", True): | |
| self.train_dataset = SyntheticVQADataset( | |
| num_samples=data_config.get("train_samples", 1000), | |
| image_size=data_config.get("image_size", 224), | |
| ) | |
| else: | |
| self.train_dataset = VQADataset( | |
| data_dir=data_config.get("data_dir", "data/vqa"), | |
| split="train", | |
| image_size=data_config.get("image_size", 224), | |
| ) | |
| if data_config.get("use_synthetic", True): | |
| self.val_dataset = SyntheticVQADataset( | |
| num_samples=data_config.get("val_samples", 200), | |
| image_size=data_config.get("image_size", 224), | |
| ) | |
| else: | |
| self.val_dataset = VQADataset( | |
| data_dir=data_config.get("data_dir", "data/vqa"), | |
| split="val", | |
| image_size=data_config.get("image_size", 224), | |
| ) | |
| if data_config.get("use_synthetic", True): | |
| self.test_dataset = SyntheticVQADataset( | |
| num_samples=data_config.get("test_samples", 200), | |
| image_size=data_config.get("image_size", 224), | |
| ) | |
| else: | |
| self.test_dataset = VQADataset( | |
| data_dir=data_config.get("data_dir", "data/vqa"), | |
| split="test", | |
| image_size=data_config.get("image_size", 224), | |
| ) | |
| def get_dataloaders(self) -> Tuple[DataLoader, DataLoader, DataLoader]: | |
| if self.train_dataset is None: | |
| self.setup_datasets() | |
| train_loader = DataLoader( | |
| self.train_dataset, | |
| batch_size=self.batch_size, | |
| shuffle=True, | |
| num_workers=self.num_workers, | |
| pin_memory=True, | |
| ) | |
| val_loader = DataLoader( | |
| self.val_dataset, | |
| batch_size=self.batch_size, | |
| shuffle=False, | |
| num_workers=self.num_workers, | |
| pin_memory=True, | |
| ) | |
| test_loader = DataLoader( | |
| self.test_dataset, | |
| batch_size=self.batch_size, | |
| shuffle=False, | |
| num_workers=self.num_workers, | |
| pin_memory=True, | |
| ) | |
| return train_loader, val_loader, test_loader | |
| def get_dataset_stats(self) -> Dict[str, Any]: | |
| if self.train_dataset is None: | |
| self.setup_datasets() | |
| stats = { | |
| "train_samples": len(self.train_dataset), | |
| "val_samples": len(self.val_dataset), | |
| "test_samples": len(self.test_dataset), | |
| "total_samples": len(self.train_dataset) | |
| + len(self.val_dataset) | |
| + len(self.test_dataset), | |
| "question_types": self.train_dataset.get_question_types(), | |
| "answer_vocab_size": len(self.train_dataset.get_answer_vocabulary()), | |
| } | |
| return stats | |
| def create_mock_image(image_path: str, size: Tuple[int, int] = (224, 224)): | |
| image = Image.new( | |
| "RGB", | |
| size, | |
| color=(random.randint(0, 255), random.randint(0, 255), random.randint(0, 255)), | |
| ) | |
| os.makedirs(os.path.dirname(image_path), exist_ok=True) | |
| image.save(image_path) | |
| return image | |
| def demonstrate_data_handling(): | |
| print("=== نمایش مدیریت داده VQA ===") | |
| config = { | |
| "data": { | |
| "use_synthetic": True, | |
| "train_samples": 100, | |
| "val_samples": 20, | |
| "test_samples": 20, | |
| "image_size": 224, | |
| }, | |
| "batch_size": 8, | |
| "num_workers": 2, | |
| } | |
| data_module = VQADataModule(config) | |
| data_module.setup_datasets() | |
| stats = data_module.get_dataset_stats() | |
| print(f"\nآمار مجموعه داده:") | |
| print(f"نمونههای آموزشی: {stats['train_samples']}") | |
| print(f"نمونههای اعتبارسنجی: {stats['val_samples']}") | |
| print(f"نمونههای تست: {stats['test_samples']}") | |
| print(f"کل نمونهها: {stats['total_samples']}") | |
| print(f"انواع سوالات: {stats['question_types']}") | |
| print(f"اندازه واژگان پاسخ: {stats['answer_vocab_size']}") | |
| train_loader, val_loader, test_loader = data_module.get_dataloaders() | |
| print(f"\nDataLoader ها:") | |
| print(f"آموزش: {len(train_loader)} batch") | |
| print(f"اعتبارسنجی: {len(val_loader)} batch") | |
| print(f"تست: {len(test_loader)} batch") | |
| print(f"\nنمونههایی از مجموعه داده آموزشی:") | |
| for i, sample in enumerate(train_loader.dataset): | |
| if i >= 3: | |
| break | |
| print(f" نمونه {i+1}:") | |
| print(f" ID: {sample.image_id}") | |
| print(f" سوال: {sample.question}") | |
| print(f" پاسخ: {sample.answer}") | |
| print(f" نوع: {sample.question_type}") | |
| print() | |
| if __name__ == "__main__": | |
| demonstrate_data_handling() | |
Xet Storage Details
- Size:
- 14.6 kB
- Xet hash:
- b9ba78db6430db139dd8eb031f617b4f86813aa170e5fba5ed9e673e77c5e253
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.