Unittests

Download test_dbnnodes.py.

import unittest
import mdp
import bimdp
import mdp.nodes as mn

from dbn_nodes import DBNLayerNode
import dbn_binodes

from mdp.test import testing_tools
from mdp.test.testing_tools import assert_array_almost_equal, assert_array_equal, \
     assert_almost_equal, assert_equal, assert_array_almost_equal_diff, \
     assert_type_equal

class DBNLayerTestCase(unittest.TestCase):
    def _test_updown_stability(self):
        """Test that the updown phase does not change optimal greedy weights.
        """

        # number of visible and hidden units
        I, J = 8, 2

        # create DBNLayer node
        node = DBNLayerNode(J, I)
        node._rbm._init_weights()
        # init to random model
        node._rbm.w = mdp.utils.random_rot(max(I,J), dtype='d')[:I, :J]
        node._rbm.bv = mdp.numx_rand.randn(I)
        node._rbm.bh = mdp.numx_rand.randn(J)

        # Gibbs sample to reach the equilibrium distribution
        N = 1e4
        v = mdp.numx_rand.randint(0,2,(N,I)).astype('d')
        for k in range(100):
            #if k%5==0: spinner()
            p, h = node._rbm._sample_h(v)
            p, v = node._rbm._sample_v(h)

        # greedy learning phase (it shouldn't change the weights by much,
        # since the input is already taken from the equilibrium distr)
        for k in range(100):
            #if k%5==0: spinner()
            node.train(v)
        node.stop_training()
        
        # save original weights
        real_w = node._rbm.w.copy()
        real_bv = node._rbm.bv.copy()
        real_bh = node._rbm.bh.copy()

        # up-down training
        node._init_updown()
        for k in range(100):
            h, ph, deltah = node._up_pass(v)
            _, _, deltav = node._down_pass(h)
            #print k, deltah, deltav

        assert_array_almost_equal(real_w, node.w_rec, 2)
        assert_array_almost_equal(real_w, node.w_gen, 2)
        assert_array_almost_equal(real_bv, node.bv, 2)
        assert_array_almost_equal(real_bh, node.bh, 2)

    def _test_updown_learning(self):
        """Test that DBNLayer is able to learn by up-down passes alone."""
        
        # number of visible and hidden units
        I, J = 4, 2
        
        node = DBNLayerNode(J, I)

        # the observations consist of two disjunct patterns that
        # never appear together
        N = 10000
        v = mdp.numx.zeros((N,I))
        for n in range(N):
            r = mdp.numx_rand.random()
            if r>0.666: v[n,:] = [0,1,0,1]
            elif r>0.333: v[n,:] = [1,0,1,0]

        # fake to train the node with a very short greedy phase
        node.train(v[:1,:])
        node.stop_training()
        # start up-down phase
        node._init_updown()

        for k in range(1500):
            #if k%5==0: spinner()
            if k>5:
                mom = 0.9
                eps = 0.2
            else:
                mom = 0.5
                eps = 0.5
            h, ph, deltah = node._up_pass(v, epsilon=eps, momentum=mom)
            rec_v, rec_pv, deltav = node._down_pass(h, epsilon=eps, momentum=mom)
            train_err = float(((v-rec_v)**2.).sum())
            #print k, train_err, train_err/N, deltah, deltav
            if train_err/N<0.1: break

        assert train_err/N<0.1

class DBNTestCase(unittest.TestCase):
    # TODO: what to do with this test?
#     def test_all_dbnlayer(self):
#         """Test that all nodes are DBNLayers"""
#         flow = [DBNLayerNode(10),
#                 mn.PCANode(input_dim=10, output_dim=20),
#                 DBNLayerNode(30)]
#         # this works
#         mdp.Flow(flow)
#         # this doesn't
#         self.assertRaises(DBNFlowException, DBNFlow, flow)

    # TODO: test updown phase where the iterable runs out

    # TODO: test updown phase for stop criterions becoming true

    # TODO: test convergence of epsilon, decay, momentum

    # TODO: test learning with multiple layers

    def test_factory_arguments(self):
        self.assertRaises(dbn_binodes.DBNException,
                          dbn_binodes.get_DBN_flow, 3, [1,2])
        self.assertRaises(dbn_binodes.DBNException,
                          dbn_binodes.get_DBN_flow, 3, [1,2,3,4])
        self.assertRaises(dbn_binodes.DBNException,
                          dbn_binodes.get_DBN_flow, 1, 2)

    def test_one_node_updown(self):
        """Test flow updown phase with just one node."""
        flow = dbn_binodes.get_DBN_flow(1, hidden_dims=[2])
        N = 10000
        x = mdp.numx.zeros((N, 4))
        for i in range(N):
            r = mdp.numx.rand()
            if r>0.666:
                x[i,:] = [0.,1.,0.,1.]
            elif r>0.333:
                x[i,:] = [1.,0.,1.,0.]

        def data_gen():
            for _ in range(100):
                # (data, learning rate, decay, momentum)
                yield (x, 0.5, 0., 0.5)

        flow.train([data_gen()])
        print flow[0].w_gen

        flow.updown_phase([x], epsilon=0.5,
                          decay=0., momentum=0.5, n_updates=3,
                          max_iter=300, min_error=1e-2)
        print flow[0].w_gen
        
        error = flow.updown_phase([x], epsilon=0.1,
                                  decay=0., momentum=0.9, n_updates=3,
                                  max_iter=1000, min_error=0.05)

        assert error < 0.05

    def test_training_finished_before_updown(self):
        flow = DBNFlow([DBNLayerNode(5), DBNLayerNode(2)])
        self.assertRaises(DBNFlowException, flow.updown_phase, None)        

    def test_updown_phase_functionality(self):
        N, I, J = 100, 5, 2
        flow = DBNFlow([DBNLayerNode(I), DBNLayerNode(J)])
        
        v = mdp.numx_rand.randint(0,2,size=(N,I)).astype('d')
        flow.train(v)
        flow.updown_phase(v, max_iter=10)
        h = flow.execute(v)
        v = flow.inverse(h)

if __name__ == "__main__":
    unittest.main()