Я поставил два Dataset конвейера для поезда / теста = 9: 1, установленных в моем графике, и управляю потоком в течение tf.cond. Я столкнулся с проблемой, что во время обучения оба конвейера активируются на каждом этапе. Тестовый набор закончился перед поездом, так как его меньше во время тренировки.
OutOfRangeError (см. Выше для отслеживания): конец последовательности
Сначала вложите входной конвейер в функцию:
def input_pipeline(*args):
...
# construct iterator
it = batch.make_initializable_iterator()
iter_init_op = it.initializer
# get next img and label
X_it, y_it = it.get_next()
inputs = {'img': X_it, 'label': y_it, 'iterator_init_op': iter_init_op}
return inputs
Затем инициируйте:
train_input = input_pipeline(args)
test_input = input_pipeline(args)
В модели мы поместили заполнитель для заполнения условия с помощью feed_dict, который не снизит производительность:
...
def f1(): return train_input
def f2(): return test_input
cond_pl = tf.placeholder(tf.string, name='cond_pl')
input = tf.cond(tf.equal(cond_pl, 'train'), lambda: f1(), lambda: f2())
...
В рамках сессии:
for ep in range(nb_ep):
...
for step in range(ep_len):
print('step:{}\r'.format(step))
try:
sess.run([train_op], feed_dict={cond_pl: 'train'})
if step % step_len == (step_len - 1):
sess.run([test_op], feed_dict={cond_pl: 'test'})
except tf.errors.OutOfRangeError:
raise('drop the remainder')
...
Как разрешить вызов входного конвейера get_next(), только если условие подходит?
Обновление фрагмента для игры на основе ответа @sharky:
def write_h5(args):
x, is_training = args
with h5py.File('./{}_{}.h5'.format('train' if is_training else 'test', x), 'w') as f:
h = w = np.arange(-1, 1, 0.02)
hh, _ = np.meshgrid(h, w)
a = hh ** 2
b = np.add(a + 1, np.random.randn(100, 100)) #do something and add gaussian noise
f.create_dataset('X', shape=(100, 100), dtype='float32', data=a)
f.create_dataset('y', shape=(100, 100), dtype='float32', data=b)
def input_pipeline(window_size, batch_size, is_train=True, ncores=mp.cpu_count()):
flist = []
for dirpath, _, fnames in os.walk('./'):
for fname in fnames:
if fname.startswith('train' if is_train else 'test') and fname.endswith('.h5'):
print(fname)
flist.append((os.path.abspath(os.path.join(dirpath, fname)), str(window_size)))
f_len = len(flist)
print(f_len)
# init list of files
batch = tf.data.Dataset.from_tensor_slices((tf.constant(flist)))
batch = batch.map(_pyfn_wrapper, num_parallel_calls=ncores)
batch = batch.shuffle(batch_size).batch(batch_size, drop_remainder=True).prefetch(ncores).repeat()
# construct iterator
it = batch.make_initializable_iterator()
iter_init_op = it.initializer
# get next img and label
X_it, y_it = it.get_next()
inputs = {'img': X_it, 'label': y_it, 'iterator_init_op': iter_init_op}
return inputs, f_len
def _pyfn_wrapper(args):
return tf.py_func(parse_h5, #wrapped pythonic function
[args],
[tf.float32, tf.float32] #[input, output] dtype
)
def parse_h5(args):
name, window_size = args
window_size = int(window_size.decode('utf-8'))
with h5py.File(name, 'r') as f:
X = f['X'][:].reshape(window_size, window_size, 1)
y = f['y'][:].reshape(window_size, window_size, 1)
return X, y
# init data
p = mp.Pool(mp.cpu_count())
p.map(write_h5, zip(range(9000), repeat(True)))
p.map(write_h5, zip(range(1000), repeat(False)))
# hparam
ep_len = 90
step_len = 9 # run test_op after 9 steps
# create tf.data.Dataset
train_input, train_len = input_pipeline(100, 5, is_train=True)
test_input, test_len = input_pipeline(100, 5, is_train=False)
# draw graph
def f1(): return train_input
def f2(): return test_input
cond_pl = tf.placeholder(tf.string, shape=None, name='cond_pl')
input = tf.cond(tf.equal(cond_pl, 'train'), lambda: f1(), lambda: f2()) # I thou
with tf.name_scope("Conv1"):
W = tf.get_variable("W", shape=[3, 3, 1, 1],
initializer=tf.contrib.layers.xavier_initializer())
b = tf.get_variable("b", shape=[1], initializer=tf.contrib.layers.xavier_initializer())
layer1 = tf.nn.conv2d(input['img'], W, strides=[1, 1, 1, 1], padding='SAME') + b
logits = tf.nn.relu(layer1)
loss = tf.reduce_mean(tf.losses.mean_squared_error(labels=input['label'], predictions=logits))
train_op = tf.train.AdamOptimizer(learning_rate=0.0001).minimize(loss)
test_op = print(loss)
#
# session
with tf.Session() as sess:
sess.run(tf.global_variables_initializer())
for ep in range(5):
print('ep:{}'.format(ep))
sess.run(input['iterator_init_op'], feed_dict={cond_pl: 'train'})
sess.run(input['iterator_init_op'], feed_dict={cond_pl: 'test'})
for step in range(ep_len):
print('step:{}\r'.format(step))
try:
sess.run([train_op], feed_dict={cond_pl: 'train'})
if step % step_len == (step_len - 1):
sess.run([test_op], feed_dict={cond_pl: 'test'})
except tf.errors.OutOfRangeError:
raise('drop the remainder')