Source code for flowvision.datasets.mnist

"""
Modified from https://github.com/pytorch/vision/blob/main/torchvision/datasets/mnist.py
"""
import warnings
from PIL import Image
import os
import os.path
import numpy as np
import codecs
from typing import Any, Callable, Dict, Optional, Tuple
from urllib.error import URLError

import oneflow as flow
from .vision import VisionDataset
from .utils import download_and_extract_archive, check_integrity
from oneflow.framework.tensor import Tensor


[docs]class MNIST(VisionDataset): r""" `MNIST <http://yann.lecun.com/exdb/mnist/>`_ Dataset. Args: root (string): Root directory of dataset where ``MNIST/processed/training.pt`` and ``MNIST/processed/test.pt`` exist. train (bool, optional): If True, creates dataset from ``training.pt``, otherwise from ``test.pt``. download (bool, optional): If true, downloads the dataset from the internet and puts it in root directory. If dataset is already downloaded, it is not downloaded again. transform (callable, optional): A function/transform that takes in an PIL image and returns a transformed version. E.g, ``transforms.RandomCrop`` target_transform (callable, optional): A function/transform that takes in the target and transforms it. """ mirrors = [ "http://yann.lecun.com/exdb/mnist/", "https://ossci-datasets.s3.amazonaws.com/mnist/", ] resources = [ ("train-images-idx3-ubyte.gz", "f68b3c2dcbeaaa9fbdd348bbdeb94873"), ("train-labels-idx1-ubyte.gz", "d53e105ee54ea40749a09fcbcd1e9432"), ("t10k-images-idx3-ubyte.gz", "9fb629c4189551a2d022fa330f9573f3"), ("t10k-labels-idx1-ubyte.gz", "ec29112dd5afa0611ce80d1b7f02629c"), ] training_file = "training.pt" test_file = "test.pt" classes = [ "0 - zero", "1 - one", "2 - two", "3 - three", "4 - four", "5 - five", "6 - six", "7 - seven", "8 - eight", "9 - nine", ] @property def train_labels(self): warnings.warn("train_labels has been renamed targets") return self.targets @property def test_labels(self): warnings.warn("test_labels has been renamed targets") return self.targets @property def train_data(self): warnings.warn("train_data has been renamed data") return self.data @property def test_data(self): warnings.warn("test_data has been renamed data") return self.data def __init__( self, root: str, train: bool = True, transform: Optional[Callable] = None, target_transform: Optional[Callable] = None, download: bool = False, source_url: Optional[str] = None, ) -> None: super(MNIST, self).__init__( root, transform=transform, target_transform=target_transform ) self.train = train # training set or test set if source_url is not None: self.mirrors = [source_url] if self._check_legacy_exist(): self.data, self.targets = self._load_legacy_data() return if download: self.download() if not self._check_exists(): raise RuntimeError( "Dataset not found." + " You can use download=True to download it" ) self.data, self.targets = self._load_data() def _check_legacy_exist(self): processed_folder_exists = os.path.exists(self.processed_folder) if not processed_folder_exists: return False return all( check_integrity(os.path.join(self.processed_folder, file)) for file in (self.training_file, self.test_file) ) def _load_legacy_data(self): # This is for BC only. We no longer cache the data in a custom binary, but simply read from the raw data # directly. data_file = self.training_file if self.train else self.test_file return flow.load(os.path.join(self.processed_folder, data_file)) def _load_data(self): image_file = f"{'train' if self.train else 't10k'}-images-idx3-ubyte" data = read_image_file(os.path.join(self.raw_folder, image_file)) label_file = f"{'train' if self.train else 't10k'}-labels-idx1-ubyte" targets = read_label_file(os.path.join(self.raw_folder, label_file)) return data, targets def __getitem__(self, index: int) -> Tuple[Any, Any]: """ Args: index (int): Index Returns: tuple: (image, target) where target is index of the target class. """ img, target = self.data[index], int(self.targets[index].numpy()) # doing this so that it is consistent with all other datasets # to return a PIL Image img = Image.fromarray(img.numpy(), mode="L") if self.transform is not None: img = self.transform(img) if self.target_transform is not None: target = self.target_transform(target) return img, target def __len__(self) -> int: return self.data.size(0) @property def raw_folder(self) -> str: return os.path.join(self.root, self.__class__.__name__, "raw") @property def processed_folder(self) -> str: return os.path.join(self.root, self.__class__.__name__, "processed") @property def class_to_idx(self) -> Dict[str, int]: return {_class: i for i, _class in enumerate(self.classes)} def _check_exists(self) -> bool: return all( check_integrity( os.path.join( self.raw_folder, os.path.splitext(os.path.basename(url))[0] ) ) for url, _ in self.resources )
[docs] def download(self) -> None: """Download the MNIST data if it doesn't exist already.""" if self._check_exists(): return os.makedirs(self.raw_folder, exist_ok=True) # download files for filename, md5 in self.resources: for mirror in self.mirrors: url = "{}{}".format(mirror, filename) try: print("Downloading {}".format(url)) download_and_extract_archive( url, download_root=self.raw_folder, filename=filename, md5=md5 ) except URLError as error: print("Failed to download (trying next):\n{}".format(error)) continue finally: print() break else: raise RuntimeError("Error downloading {}".format(filename))
def extra_repr(self) -> str: return "Split: {}".format("Train" if self.train is True else "Test")
[docs]class FashionMNIST(MNIST): r""" `Fashion-MNIST <https://github.com/zalandoresearch/fashion-mnist>`_ Dataset. Args: root (string): Root directory of dataset where ``FashionMNIST/processed/training.pt`` and ``FashionMNIST/processed/test.pt`` exist. train (bool, optional): If True, creates dataset from ``training.pt``, otherwise from ``test.pt``. download (bool, optional): If true, downloads the dataset from the internet and puts it in root directory. If dataset is already downloaded, it is not downloaded again. transform (callable, optional): A function/transform that takes in an PIL image and returns a transformed version. E.g, ``transforms.RandomCrop`` target_transform (callable, optional): A function/transform that takes in the target and transforms it. """ mirrors = ["http://fashion-mnist.s3-website.eu-central-1.amazonaws.com/"] resources = [ ("train-images-idx3-ubyte.gz", "8d4fb7e6c68d591d4c3dfef9ec88bf0d"), ("train-labels-idx1-ubyte.gz", "25c81989df183df01b3e8a0aad5dffbe"), ("t10k-images-idx3-ubyte.gz", "bef4ecab320f06d8554ea6380940ec79"), ("t10k-labels-idx1-ubyte.gz", "bb300cfdad3c16e7a12a480ee83cd310"), ] classes = [ "T-shirt/top", "Trouser", "Pullover", "Dress", "Coat", "Sandal", "Shirt", "Sneaker", "Bag", "Ankle boot", ]
SN3_PASCALVINCENT_TYPEMAP = { 8: (flow.uint8, np.uint8, np.uint8), 9: (flow.int8, np.int8, np.int8), 11: (flow.int32, np.dtype(">i2"), "i2"), 12: (flow.int32, np.dtype(">i4"), "i4"), 13: (flow.float32, np.dtype(">f4"), "f4"), 14: (flow.float64, np.dtype(">f8"), "f8"), } def get_int(b: bytes) -> int: return int(codecs.encode(b, "hex"), 16) def read_sn3_pascalvincent_tensor(path: str, strict: bool = True) -> Tensor: """Read a SN3 file in "Pascal Vincent" format (Lush file 'libidx/idx-io.lsh'). Argument may be a filename, compressed filename, or file object. """ # read with open(path, "rb") as f: data = f.read() # parse magic = get_int(data[0:4]) nd = magic % 256 ty = magic // 256 assert 1 <= nd <= 3 assert 8 <= ty <= 14 m = SN3_PASCALVINCENT_TYPEMAP[ty] s = [get_int(data[4 * (i + 1) : 4 * (i + 2)]) for i in range(nd)] parsed = np.frombuffer(data, dtype=m[1], offset=(4 * (nd + 1))) assert parsed.shape[0] == np.prod(s) or not strict return flow.reshape(flow.tensor(parsed.astype(m[2]), dtype=m[0]), shape=s) def read_label_file(path: str) -> Tensor: x = read_sn3_pascalvincent_tensor(path, strict=False) assert x.dtype == flow.uint8 assert x.ndimension() == 1 x = x.to(dtype=flow.int64) return x def read_image_file(path: str) -> Tensor: x = read_sn3_pascalvincent_tensor(path, strict=False) assert x.dtype == flow.uint8 assert x.ndimension() == 3 return x