提交 d6c62e85 编写于 作者: Y Yi Wang

Rewrite mnist.py and add mnist_test.py

上级 91115ab6
import sklearn.datasets.mldata
import sklearn.model_selection
import paddle.v2.dataset.common
import subprocess
import numpy
from common import DATA_HOME
__all__ = ['train_creator', 'test_creator']
URL_PREFIX = 'http://yann.lecun.com/exdb/mnist/'
TEST_IMAGE_URL = URL_PREFIX + 't10k-images-idx3-ubyte.gz'
TEST_IMAGE_MD5 = '25e3cc63507ef6e98d5dc541e8672bb6'
def __mnist_reader_creator__(data, target):
def reader():
n_samples = data.shape[0]
for i in xrange(n_samples):
yield (data[i] / 255.0).astype(numpy.float32), int(target[i])
TEST_LABEL_URL = URL_PREFIX + 't10k-labels-idx1-ubyte.gz'
TEST_LABEL_MD5 = '4e9511fe019b2189026bd0421ba7b688'
TRAIN_IMAGE_URL = URL_PREFIX + 'train-images-idx3-ubyte.gz'
TRAIN_IMAGE_MD5 = 'f68b3c2dcbeaaa9fbdd348bbdeb94873'
return reader
TRAIN_LABEL_URL = URL_PREFIX + 'train-labels-idx1-ubyte.gz'
TRAIN_LABEL_MD5 = 'd53e105ee54ea40749a09fcbcd1e9432'
TEST_SIZE = 10000
def reader_creator(image_filename, label_filename, buffer_size):
def reader():
# According to http://stackoverflow.com/a/38061619/724872, we
# cannot use standard package gzip here.
m = subprocess.Popen(["zcat", image_filename], stdout=subprocess.PIPE)
m.stdout.read(16) # skip some magic bytes
l = subprocess.Popen(["zcat", label_filename], stdout=subprocess.PIPE)
l.stdout.read(8) # skip some magic bytes
data = sklearn.datasets.mldata.fetch_mldata(
"MNIST original", data_home=DATA_HOME)
X_train, X_test, y_train, y_test = sklearn.model_selection.train_test_split(
data.data, data.target, test_size=TEST_SIZE, random_state=0)
while True:
labels = numpy.fromfile(
l.stdout, 'ubyte', count=buffer_size
).astype("int")
if labels.size != buffer_size:
break # numpy.fromfile returns empty slice after EOF.
def train_creator():
return __mnist_reader_creator__(X_train, y_train)
images = numpy.fromfile(
m.stdout, 'ubyte', count=buffer_size * 28 * 28
).reshape((buffer_size, 28 * 28)
).astype('float32')
images = images / 255.0 * 2.0 - 1.0
def test_creator():
return __mnist_reader_creator__(X_test, y_test)
for i in xrange(buffer_size):
yield images[i, :], labels[i]
m.terminate()
l.terminate()
def unittest():
assert len(list(test_creator()())) == TEST_SIZE
return reader()
def train():
return reader_creator(
paddle.v2.dataset.common.download(
TRAIN_IMAGE_URL, 'mnist', TRAIN_IMAGE_MD5),
paddle.v2.dataset.common.download(
TRAIN_LABEL_URL, 'mnist', TRAIN_LABEL_MD5),
100)
if __name__ == '__main__':
unittest()
def test():
return reader_creator(
paddle.v2.dataset.common.download(
TEST_IMAGE_URL, 'mnist', TEST_IMAGE_MD5),
paddle.v2.dataset.common.download(
TEST_LABEL_URL, 'mnist', TEST_LABEL_MD5),
100)
import paddle.v2.dataset.mnist
import unittest
class TestMNIST(unittest.TestCase):
def check_reader(self, reader):
sum = 0
for l in reader:
self.assertEqual(l[0].size, 784)
self.assertEqual(l[1].size, 1)
self.assertLess(l[1], 10)
self.assertGreaterEqual(l[1], 0)
sum += 1
return sum
def test_train(self):
self.assertEqual(
self.check_reader(paddle.v2.dataset.mnist.train()),
60000)
def test_test(self):
self.assertEqual(
self.check_reader(paddle.v2.dataset.mnist.test()),
10000)
if __name__ == '__main__':
unittest.main()
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册