https://github.com/GPflow/GPflow
Tip revision: bd04738eff4c8fa2c05ec14b69bffe32357d02f4 authored by vdutor on 25 June 2018, 14:04:02 UTC
Fix the global step for summary tasks
Fix the global step for summary tasks
Tip revision: bd04738
test_saver.py
# Copyright 2018 the GPflow authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import copy
import os
import tempfile
import numpy as np
import pytest
import tensorflow as tf
from numpy.testing import assert_allclose
import gpflow as gp
from gpflow.test_util import session_context, session_tf
# ==============================
# Fixtures and data definitions.
# ==============================
class Data:
@staticmethod
def deep_structure():
a = gp.Param(1)
b = gp.Param(2)
c_a = gp.Param(3)
c_b = gp.Param(4)
with gp.defer_build():
p = gp.Parameterized()
p.c = gp.Parameterized()
p.c.c = gp.Parameterized()
p.c.c.a = gp.Param(3)
p.c.c.b = gp.Param(4)
p.a = a
p.b = b
p.c.a = c_a
p.c.b = c_b
p.compile()
return p
@staticmethod
def model():
x = Data.x_new()
y = np.random.rand(10, 1)
kernel = gp.kernels.RBF(2)
m = gp.models.GPR(x, y, kernel)
return m
@staticmethod
def x_new():
return np.random.rand(10, 2)
@pytest.fixture
def filename(request):
with tempfile.NamedTemporaryFile() as file:
yield file.name
@pytest.fixture
def deep_structure(session_tf):
return Data.deep_structure()
@pytest.fixture
def model():
return Data.model()
def simple_type_values():
return [0, 0., 10, 10.0, np.array(1),
np.array(10.), np.float16(2),
np.int(10), np.float32(10),
"test", "", None, True, False,
tf.exp]
def list_type_values():
return [
[1, 2, 3],
["", "artem", 1, np.float(32), False],
["", 1, np.array([1,2,3]), True, None],
np.array([10]),
np.array([10, 20, 30]),
np.array([10, 20, 30], dtype=np.float32),
]
def collection_type_values():
return [
{'a': 1, 'b': 'test', 'c': None},
{'a': np.array([1])},
{'a': np.array([1, 2, 3])},
{'a': np.array([1, 2, 3]), 'b': ""}
]
# ======
# Tests.
# ======
@pytest.mark.parametrize('value', simple_type_values())
def test_encode_decode_simple_types(value):
d = encode_decode(value)
assert value == d
@pytest.mark.parametrize('value', list_type_values())
def test_encode_decode_list_types(value):
d = encode_decode(value)
def equal(x):
a, b = x
eq = a == b
if isinstance(eq, np.ndarray) and eq.shape:
return all(eq)
return eq
assert all(list(map(equal, zip(value, d))))
@pytest.mark.parametrize('value', collection_type_values())
def test_encode_decode_collection_types(value):
d = encode_decode(value)
assert value.keys() == d.keys()
for k, v in value.items():
if isinstance(v, np.ndarray) and v.shape:
assert all(d[k] == v)
else:
assert d[k] == v
def test_saving_deep_parameterized_object(session_tf, filename, deep_structure):
sess_a = session_tf
gp.Saver().save(filename, deep_structure)
with session_context() as sess_b:
copy = gp.Saver().load(filename)
equal_params(deep_structure.a, copy.a, session_a=sess_a, session_b=sess_b)
equal_params(deep_structure.b, copy.b, session_a=sess_a, session_b=sess_b)
equal_params(deep_structure.c.a, copy.c.a, session_a=sess_a, session_b=sess_b)
equal_params(deep_structure.c.b, copy.c.b, session_a=sess_a, session_b=sess_b)
equal_params(deep_structure.c.c.a, copy.c.c.a, session_a=sess_a, session_b=sess_b)
equal_params(deep_structure.c.c.b, copy.c.c.b, session_a=sess_a, session_b=sess_b)
def test_saving_gpflow_model(session_tf, filename, model):
x_new = Data.x_new()
predict_origin = model.predict_f(x_new)
gp.Saver().save(filename, model)
with session_context() as session:
loaded = gp.Saver().load(filename)
predict_loaded = loaded.predict_f(x_new)
assert_allclose(predict_origin, predict_loaded)
def test_loading_without_autocompile(session_tf, filename, model):
gp.Saver().save(filename, model)
with session_context() as session:
context = gp.SaverContext(autocompile=False)
loaded = gp.Saver().load(filename, context=context)
assert loaded.is_built(session_tf.graph) == gp.Build.NO
assert loaded.is_built(session.graph) == gp.Build.NO
assert not any(loaded.trainable_tensors)
def test_loading_into_specific_session(session_tf, filename, model):
x_new = Data.x_new()
predict_origin = model.predict_f(x_new)
gp.Saver().save(filename, model)
with session_context() as session:
context = gp.SaverContext(session=session)
loaded = gp.Saver().load(filename, context=context)
predict_loaded = loaded.predict_f(x_new, session=session)
assert_allclose(predict_origin, predict_loaded)
# ========
# Helpers.
# ========
def encode_decode(value):
ctx = gp.SaverContext()
e = gp.saver.CoderDispatcher(ctx).encode(value)
return gp.saver.CoderDispatcher(ctx).decode(e)
def equal_params(a, b, session_a=None, session_b=None):
assert a.name == b.name
assert a.pathname == b.pathname
assert a.tf_pathname == b.tf_pathname
assert a.tf_name_scope == b.tf_name_scope
val_a = a.read_value(session=session_a)
val_b = b.read_value(session=session_b)
val_a_back = a.transform.backward(val_a)
val_b_back = b.transform.backward(val_b)
assert_allclose(val_a_back, val_b_back)