test_model.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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.
# =============================================================================

from __future__ import division

import os
import math
import unittest
import numpy as np

from singa import singa_wrap as singa_api
from singa.tensor import Tensor
from singa import autograd
from singa import tensor
from singa import device
from singa import layer
from singa import model
from singa import opt

from cuda_helper import gpu_dev, cpu_dev


class DoubleLinear(layer.Layer):

    def __init__(self, a, b, c):
        super(DoubleLinear, self).__init__()
        self.l1 = layer.Linear(a, b)
        self.l2 = layer.Linear(b, c)

    def forward(self, x):
        y = self.l1(x)
        y = self.l2(y)
        return y


class MyModel(model.Model):

    def __init__(self):
        super(MyModel, self).__init__()
        self.conv1 = layer.Conv2d(2, 2)
        self.bn1 = layer.BatchNorm2d(2)
        self.doublelinear1 = DoubleLinear(2, 4, 2)
        self.optimizer = opt.SGD()

    def forward(self, x):
        y = self.conv1(x)
        y = self.bn1(y)
        y = autograd.reshape(y, (y.shape[0], -1))
        y = self.doublelinear1(y)
        return y

    def train_one_batch(self, x, y):
        y_ = self.forward(x)
        l = self.loss(y_, y)
        self.optim(l)
        return y_, l

    def loss(self, out, ty):
        return autograd.softmax_cross_entropy(out, ty)

    def optim(self, loss):
        self.optimizer(loss)


class MLP(model.Model):

    def __init__(self, data_size=10, perceptron_size=100, num_classes=10):
        super(MLP, self).__init__()
        self.num_classes = num_classes
        self.dimension = 2

        self.relu = layer.ReLU()
        self.linear1 = layer.Linear(perceptron_size)
        self.linear2 = layer.Linear(num_classes)
        self.softmax_cross_entropy = layer.SoftMaxCrossEntropy()

    def forward(self, inputs):
        y = self.linear1(inputs)
        y = self.relu(y)
        y = self.linear2(y)
        return y

    def train_one_batch(self, x, y):
        out = self.forward(x)
        loss = self.softmax_cross_entropy(out, y)
        self.optimizer(loss)
        return out, loss

    def set_optimizer(self, optimizer):
        self.optimizer = optimizer


# lstm testing
class LSTMModel3(model.Model):

    def __init__(self, hidden_size):
        super(LSTMModel3, self).__init__()
        self.lstm = layer.CudnnRNN(
            hidden_size=hidden_size,
            batch_first=True,
            #    return_sequences=True,
            use_mask=True)
        self.l1 = layer.Linear(2)
        self.optimizer = opt.SGD(0.1)

    def forward(self, x, seq_lengths):
        y = self.lstm(x, seq_lengths=seq_lengths)
        y = autograd.reshape(y, (y.shape[0], -1))
        y = self.l1(y)
        return y


class LSTMModel2(model.Model):

    def __init__(self, hidden_size, bidirectional, num_layers):
        super(LSTMModel2, self).__init__()
        self.lstm = layer.CudnnRNN(hidden_size=hidden_size,
                                   num_layers=num_layers,
                                   bidirectional=bidirectional,
                                   return_sequences=False,
                                   rnn_mode='lstm',
                                   batch_first=True)
        self.optimizer = opt.SGD(0.1)

    def forward(self, x):
        return self.lstm(x)


class LSTMModel(model.Model):

    def __init__(self, hidden_size, seq_length, batch_size, bidirectional,
                 num_layers, return_sequences, rnn_mode, batch_first):
        super(LSTMModel, self).__init__()
        self.hidden_size = hidden_size
        self.seq_length = seq_length
        self.return_sequences = return_sequences

        self.lstm = layer.CudnnRNN(hidden_size=hidden_size,
                                   num_layers=num_layers,
                                   bidirectional=bidirectional,
                                   return_sequences=return_sequences,
                                   rnn_mode=rnn_mode,
                                   batch_first=batch_first)
        self.optimizer = opt.SGD(0.1)

    def forward(self, x):
        y = self.lstm(x)
        if self.return_sequences:
            y = autograd.reshape(y, (-1, self.seq_length * self.hidden_size))
        return y


class TestModelMethods(unittest.TestCase):

    @unittest.skipIf(not singa_api.USE_CUDA, 'CUDA is not enabled')
    def test_rnn_with_seq_lengths(self, dev=gpu_dev):
        bs = 2
        seq_length = 3
        hidden_size = 2
        em_size = 2
        x_np = np.array([[[0.1, 0.1], [0.2, 0.2], [0.3, 0.3]],
                         [[0.3, 0.3], [0.4, 0.4], [0.0,
                                                   0.0]]]).astype(np.float32)
        y_np = np.array([[0.4, 0.4], [0.5, 0.5]]).astype(np.float32)
        seq_lengths_np = np.array([3, 2]).astype(np.int32)

        x = tensor.from_numpy(x_np)
        x.to_device(dev)
        y = tensor.from_numpy(y_np)
        y.to_device(dev)
        seq_lengths = tensor.from_numpy(seq_lengths_np)

        m = LSTMModel3(hidden_size)
        m.compile([x, seq_lengths],
                  is_train=True,
                  use_graph=False,
                  sequential=False)
        m.train()
        for i in range(10):
            out = m.forward(x, seq_lengths)
            loss = autograd.mse_loss(out, y)
            print("train l:", tensor.to_numpy(loss))
            m.optimizer(loss)
        m.eval()
        out = m.forward(x, seq_lengths)
        loss = autograd.mse_loss(out, y)
        print(" eval l:", tensor.to_numpy(loss))

    @unittest.skipIf(not singa_api.USE_CUDA, 'CUDA is not enabled')
    def test_lstm_model(self, dev=gpu_dev):
        hidden_size = 3
        seq_length = 2
        batch_size = 4
        feature_size = 3
        bidirectional = False
        directions = 2 if bidirectional else 1
        num_layers = 2
        out_size = hidden_size
        return_sequences = False
        batch_first = True
        rnn_mode = "lstm"

        # manual test case
        x_data = np.array([[[0, 0, 1], [0, 1, 0]], [[0, 1, 0], [1, 0, 0]],
                           [[0, 0, 1], [0, 1, 0]], [[1, 0, 0], [0, 0, 1]]],
                          dtype=np.float32).reshape(batch_size, seq_length,
                                                    hidden_size)  # bs, seq, fea
        if return_sequences:
            y_data = np.array(
                [[[0, 1, 0], [1, 0, 0]], [[1, 0, 0], [0, 0, 1]],
                 [[0, 1, 0], [1, 0, 0]], [[0, 0, 1], [0, 1, 0]]],
                dtype=np.float32).reshape(batch_size, seq_length,
                                          hidden_size)  # bs, hidden
            y_data.reshape(batch_size, -1)
        else:
            y_data = np.array([[1, 0, 0], [0, 0, 1], [1, 0, 0], [0, 1, 0]],
                              dtype=np.float32).reshape(
                                  batch_size, hidden_size)  # bs, hidden

        x = tensor.Tensor(device=dev, data=x_data)
        y_t = tensor.Tensor(device=dev, data=y_data)

        m = LSTMModel(hidden_size, seq_length, batch_size, bidirectional,
                      num_layers, return_sequences, rnn_mode, batch_first)
        m.compile([x], is_train=True, use_graph=False, sequential=False)

        m.train()
        for i in range(1000):
            y = m.forward(x)
            assert y.shape == y_t.shape
            loss = autograd.softmax_cross_entropy(y, y_t)
            if i % 100 == 0:
                print("loss", loss)
            m.optimizer(loss)

        m.eval()
        y = m.forward(x)
        loss = autograd.softmax_cross_entropy(y, y_t)
        print("eval loss", loss)


class TestModelSaveMethods(unittest.TestCase):

    def _save_states_load_states_helper(self, dev, graph_flag="False"):
        x_shape = (2, 2, 2, 2)
        x = tensor.PlaceHolder(x_shape, device=dev)

        m = MyModel()
        m.compile([x], is_train=True, use_graph=graph_flag, sequential=False)

        states = {
            "conv1.W":
                tensor.Tensor((2, 2, 2, 2), device=dev).set_value(0.1),
            "conv1.b":
                tensor.Tensor((2,), device=dev).set_value(0.2),
            "bn1.scale":
                tensor.Tensor((2,), device=dev).set_value(0.3),
            "bn1.bias":
                tensor.Tensor((2,), device=dev).set_value(0.4),
            "bn1.running_mean":
                tensor.Tensor((2,), device=dev).set_value(0.5),
            "bn1.running_var":
                tensor.Tensor((2,), device=dev).set_value(0.6),
            "doublelinear1.l1.W":
                tensor.Tensor((2, 4), device=dev).set_value(0.7),
            "doublelinear1.l1.b":
                tensor.Tensor((4,), device=dev).set_value(0.8),
            "doublelinear1.l2.W":
                tensor.Tensor((4, 2), device=dev).set_value(0.9),
            "doublelinear1.l2.b":
                tensor.Tensor((2,), device=dev).set_value(1.0)
        }

        m.set_states(states)
        states2 = m.get_states()
        for k in states2.keys():
            np.testing.assert_array_almost_equal(tensor.to_numpy(states[k]),
                                                 tensor.to_numpy(states2[k]))

        opt_state1 = tensor.Tensor((2, 10), device=dev).gaussian(1, 0.1)
        opt_state2 = tensor.Tensor((20, 2), device=dev).gaussian(0.1, 1)
        aux = {"opt1": opt_state1, "opt2": opt_state2}

        # save snapshot1
        zip_fp = 'snapshot1_%s.zip' % self._testMethodName
        if os.path.exists(zip_fp):
            os.remove(zip_fp)
        m.save_states(zip_fp, aux)

        # do some training, states changes
        cx = tensor.Tensor(x_shape, device=dev).gaussian(1, 1)
        cy = tensor.Tensor((2, 2), device=dev).gaussian(1, 1)
        mini_batch_size = 10
        for i in range(mini_batch_size):
            m.train_one_batch(cx, cy)

        # restore snapshot
        aux2 = m.load_states(zip_fp)
        np.testing.assert_array_almost_equal(tensor.to_numpy(aux2["opt1"]),
                                             tensor.to_numpy(aux["opt1"]))
        np.testing.assert_array_almost_equal(tensor.to_numpy(aux2["opt2"]),
                                             tensor.to_numpy(aux["opt2"]))

        # snapshot states
        states3 = m.get_states()
        for k in states3.keys():
            np.testing.assert_array_almost_equal(tensor.to_numpy(states[k]),
                                                 tensor.to_numpy(states3[k]))

    @unittest.skipIf(not singa_api.USE_CUDA, 'CUDA is not enabled')
    def test_save_states_load_states_gpu(self):
        self._save_states_load_states_helper(gpu_dev, graph_flag=False)
        self._save_states_load_states_helper(gpu_dev, graph_flag=True)

    def test_save_states_load_states_cpu(self):
        self._save_states_load_states_helper(cpu_dev, graph_flag=False)
        self._save_states_load_states_helper(cpu_dev, graph_flag=True)


class TestPythonModule(unittest.TestCase):

    def to_categorical(self, y, num_classes):
        y = np.array(y, dtype="int")
        n = y.shape[0]
        categorical = np.zeros((n, num_classes))
        categorical[np.arange(n), y] = 1
        return categorical

    def generate_data(self, dev, num=400):
        f = lambda x: (5 * x + 1)

        x = np.random.uniform(-1, 1, num)
        y = f(x) + 2 * np.random.randn(len(x))

        self.label = np.asarray([5 * a + 1 > b for (a, b) in zip(x, y)])
        self.data = np.array([[a, b] for (a, b) in zip(x, y)], dtype=np.float32)
        self.label = self.to_categorical(self.label, 2).astype(np.float32)

        self.inputs = Tensor(data=self.data, device=dev)
        self.target = Tensor(data=self.label, device=dev)

    def get_params(self, model):
        params = model.get_params()
        self.w0 = params['linear1.W']
        self.b0 = params['linear1.b']
        self.w1 = params['linear2.W']
        self.b1 = params['linear2.b']

        self.W0 = tensor.to_numpy(self.w0)
        self.B0 = tensor.to_numpy(self.b0)
        self.W1 = tensor.to_numpy(self.w1)
        self.B1 = tensor.to_numpy(self.b1)

    def numpy_forward(self, inputs):
        self.x1 = np.matmul(inputs, self.W0)
        self.x2 = np.add(self.x1, self.B0)
        self.x3 = np.maximum(self.x2, 0)
        self.x4 = np.matmul(self.x3, self.W1)
        self.x5 = np.add(self.x4, self.B1)
        return self.x5

    def numpy_train_one_batch(self, inputs, y):
        # forward propagation
        out = self.numpy_forward(inputs)

        # softmax cross entropy loss
        exp_out = np.exp(out - np.max(out, axis=-1, keepdims=True))
        self.softmax = exp_out / np.sum(exp_out, axis=-1, keepdims=True)
        loss = np.sum(y * np.log(self.softmax)) / -self.softmax.shape[0]

        # optimize
        # calculate gradients
        label_sum = np.sum(self.label, axis=-1)
        dloss = self.softmax - self.label / label_sum.reshape(
            label_sum.shape[0], 1)
        dloss /= self.softmax.shape[0]

        dx5 = dloss
        db1 = np.sum(dloss, 0)

        dx4 = np.matmul(dx5, self.W1.T)
        dw1 = np.matmul(self.x3.T, dx5)

        dx3 = dx4 * (self.x3 > 0)

        dx2 = dx3
        db0 = np.sum(dx3, 0)

        dx1 = np.matmul(dx2, self.W0.T)
        dw0 = np.matmul(self.data.T, dx2)

        # update all the params
        self.W0 -= 0.05 * dw0
        self.B0 -= 0.05 * db0
        self.W1 -= 0.05 * dw1
        self.B1 -= 0.05 * db1
        return out, loss

    def setUp(self):
        self.sgd = opt.SGD(lr=0.05)

        cpu_dev.ResetGraph()
        if singa_api.USE_CUDA:
            gpu_dev.ResetGraph()

    def tearDown(self):
        cpu_dev.ResetGraph()
        if singa_api.USE_CUDA:
            gpu_dev.ResetGraph()

    def _forward_helper(self, dev, is_train, use_graph, sequential):
        self.generate_data(dev)
        model = MLP(self.sgd)
        model.compile([self.inputs],
                      is_train=is_train,
                      use_graph=use_graph,
                      sequential=sequential)

        self.get_params(model)

        out = model(self.inputs)
        np_out = self.numpy_forward(self.data)

        np.testing.assert_array_almost_equal(tensor.to_numpy(out), np_out)

    def _train_one_batch_helper(self, dev, is_train, use_graph, sequential):
        self.generate_data(dev)
        model = MLP(num_classes=2)
        model.set_optimizer(self.sgd)
        model.compile([self.inputs],
                      is_train=is_train,
                      use_graph=use_graph,
                      sequential=sequential)

        self.get_params(model)

        out, loss = model(self.inputs, self.target)
        np_out, np_loss = self.numpy_train_one_batch(self.data, self.label)

        np.testing.assert_array_almost_equal(tensor.to_numpy(out), np_out)
        np.testing.assert_array_almost_equal(tensor.to_numpy(loss), np_loss)
        np.testing.assert_array_almost_equal(tensor.to_numpy(self.w0), self.W0)
        np.testing.assert_array_almost_equal(tensor.to_numpy(self.b0), self.B0)
        np.testing.assert_array_almost_equal(tensor.to_numpy(self.w1), self.W1)
        np.testing.assert_array_almost_equal(tensor.to_numpy(self.b1), self.B1)

    def test_forward_cpu(self):
        self._forward_helper(cpu_dev, False, True, False)

    @unittest.skipIf(not singa_api.USE_CUDA, 'CUDA is not enabled')
    def test_forward_gpu(self):
        self._forward_helper(gpu_dev, False, True, False)

    def test_evaluate_cpu(self):
        self._forward_helper(cpu_dev, False, False, False)

    @unittest.skipIf(not singa_api.USE_CUDA, 'CUDA is not enabled')
    def test_evaluate_gpu(self):
        self._forward_helper(gpu_dev, False, False, False)

    def test_train_one_batch_cpu(self):
        self._train_one_batch_helper(cpu_dev, True, True, False)

    @unittest.skipIf(not singa_api.USE_CUDA, 'CUDA is not enabled')
    def test_train_one_batch_gpu(self):
        self._train_one_batch_helper(gpu_dev, True, True, False)

    def test_without_graph_cpu(self):
        self._train_one_batch_helper(cpu_dev, True, False, False)

    @unittest.skipIf(not singa_api.USE_CUDA, 'CUDA is not enabled')
    def test_without_graph_gpu(self):
        self._train_one_batch_helper(gpu_dev, True, False, False)

    def test_run_in_serial_cpu(self):
        self._train_one_batch_helper(cpu_dev, True, True, True)

    @unittest.skipIf(not singa_api.USE_CUDA, 'CUDA is not enabled')
    def test_run_in_serial_gpu(self):
        self._train_one_batch_helper(gpu_dev, True, True, True)


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