text: from __future__ import absolute_import
  from __future__ import print_function
  import pytest
  import numpy as np
  
  from keras import backend as K
  from keras.utils.test_utils import get_test_data
  from keras.models import Sequential, Model
  from keras.layers import Dense, Activation, GRU, TimeDistributed, Input
  from keras.utils import np_utils
  from keras.utils.test_utils import keras_test
  from numpy.testing import assert_almost_equal, assert_array_almost_equal
  
  num_classes = 10
  batch_size = 128
  epochs = 15
  weighted_class = 5
  high_weight = 10
  train_samples = 5000
  test_samples = 1000
  timesteps = 3
  input_dim = 10
  loss = 'mse'
  loss_full_name = 'mean_squared_error'
  standard_weight = 1
  standard_score_sequential = 0.5
  
  decimal_precision = {
      'cntk': 2,
      'theano': 6,
      'tensorflow': 6
  }
  
  
  def _get_test_data():
      np.random.seed(1337)
      (x_train, y_train), (x_test, y_test) = get_test_data(num_train=train_samples,
                                                           num_test=test_samples,
                                                           input_shape=(input_dim,),
                                                           classification=True,
                                                           num_classes=num_classes)
      int_y_test = y_test.copy()
      int_y_train = y_train.copy()
      # convert class vectors to binary class matrices
      y_train = np_utils.to_categorical(y_train, num_classes)
      y_test = np_utils.to_categorical(y_test, num_classes)
      test_ids = np.where(int_y_test == np.array(weighted_class))[0]
  
      class_weight = dict([(i, standard_weight) for i in range(num_classes)])
      class_weight[weighted_class] = high_weight
  
      sample_weight = np.ones((y_train.shape[0])) * standard_weight
      sample_weight[int_y_train == weighted_class] = high_weight
  
      return (x_train, y_train), (x_test, y_test), (sample_weight, class_weight, test_ids)
  
  
  def create_sequential_model():
      model = Sequential()
      model.add(Dense(32, input_shape=(input_dim,)))
      model.add(Activation('relu'))
      model.add(Dense(num_classes))
      model.add(Activation('softmax'))
      return model
  
  
  def create_temporal_sequential_model():
      model = Sequential()
      model.add(GRU(32, input_shape=(timesteps, input_dim), return_sequences=True))
      model.add(TimeDistributed(Dense(num_classes)))
      model.add(Activation('softmax'))
      return model
  
  
  @keras_test
  def test_sequential_class_weights():
      model = create_sequential_model()
      model.compile(loss=loss, optimizer='rmsprop')
  
      (x_train, y_train), (x_test, y_test), (sample_weight, class_weight, test_ids) = _get_test_data()
  
      model.fit(x_train, y_train, batch_size=batch_size,
                epochs=epochs // 3, verbose=0,
                class_weight=class_weight,
                validation_data=(x_train, y_train, sample_weight))
      model.fit(x_train, y_train, batch_size=batch_size,
                epochs=epochs // 2, verbose=0,
                class_weight=class_weight)
      model.fit(x_train, y_train, batch_size=batch_size,
                epochs=epochs // 2, verbose=0,
                class_weight=class_weight,
                validation_split=0.1)
  
      model.train_on_batch(x_train[:32], y_train[:32],
                           class_weight=class_weight)
      score = model.evaluate(x_test[test_ids, :], y_test[test_ids, :], verbose=0)
      assert(score < standard_score_sequential)
  
  
  @keras_test
  def test_sequential_sample_weights():
      model = create_sequential_model()
      model.compile(loss=loss, optimizer='rmsprop')
  
      (x_train, y_train), (x_test, y_test), (sample_weight, class_weight, test_ids) = _get_test_data()
  
      model.fit(x_train, y_train, batch_size=batch_size,
                epochs=epochs // 3, verbose=0,
                sample_weight=sample_weight)
      model.fit(x_train, y_train, batch_size=batch_size,
                epochs=epochs // 3, verbose=0,
                sample_weight=sample_weight,
                validation_split=0.1)
  
      model.train_on_batch(x_train[:32], y_train[:32],
                           sample_weight=sample_weight[:32])
      model.test_on_batch(x_train[:32], y_train[:32],
                          sample_weight=sample_weight[:32])
      score = model.evaluate(x_test[test_ids, :], y_test[test_ids, :], verbose=0)
      assert(score < standard_score_sequential)
  
  
  @keras_test
  def test_sequential_temporal_sample_weights():
      (x_train, y_train), (x_test, y_test), (sample_weight, class_weight, test_ids) = _get_test_data()
  
      temporal_x_train = np.reshape(x_train, (len(x_train), 1, x_train.shape[1]))
      temporal_x_train = np.repeat(temporal_x_train, timesteps, axis=1)
      temporal_x_test = np.reshape(x_test, (len(x_test), 1, x_test.shape[1]))
      temporal_x_test = np.repeat(temporal_x_test, timesteps, axis=1)
  
      temporal_y_train = np.reshape(y_train, (len(y_train), 1, y_train.shape[1]))
      temporal_y_train = np.repeat(temporal_y_train, timesteps, axis=1)
      temporal_y_test = np.reshape(y_test, (len(y_test), 1, y_test.shape[1]))
      temporal_y_test = np.repeat(temporal_y_test, timesteps, axis=1)
  
      temporal_sample_weight = np.reshape(sample_weight, (len(sample_weight), 1))
      temporal_sample_weight = np.repeat(temporal_sample_weight, timesteps, axis=1)
  
      model = create_temporal_sequential_model()
      model.compile(loss=loss, optimizer='rmsprop',
                    sample_weight_mode='temporal')
  
      model.fit(temporal_x_train, temporal_y_train, batch_size=batch_size,
                epochs=epochs // 3, verbose=0,
                sample_weight=temporal_sample_weight)
      model.fit(temporal_x_train, temporal_y_train, batch_size=batch_size,
                epochs=epochs // 3, verbose=0,
                sample_weight=temporal_sample_weight,
                validation_split=0.1)
  
      model.train_on_batch(temporal_x_train[:32], temporal_y_train[:32],
                           sample_weight=temporal_sample_weight[:32])
      model.test_on_batch(temporal_x_train[:32], temporal_y_train[:32],
                          sample_weight=temporal_sample_weight[:32])
      score = model.evaluate(temporal_x_test[test_ids], temporal_y_test[test_ids], verbose=0)
      assert(score < standard_score_sequential)
  
  
  @keras_test
  def test_weighted_metrics_with_sample_weight():
      decimal = decimal_precision[K.backend()]
  
      model = create_sequential_model()
      model.compile(loss=loss, optimizer='rmsprop', metrics=[loss], weighted_metrics=[loss])
  
      (x_train, y_train), (x_test, y_test), (sample_weight, class_weight, test_ids) = _get_test_data()
  
      history = model.fit(x_train, y_train, batch_size=batch_size,
                          epochs=epochs // 3, verbose=0,
                          sample_weight=sample_weight)
  
      h = history.history
      assert_array_almost_equal(h['loss'], h['weighted_' + loss_full_name], decimal=decimal)
  
      history = model.fit(x_train, y_train, batch_size=batch_size,
                          epochs=epochs // 3, verbose=0,
                          sample_weight=sample_weight,
                          validation_split=0.1)
  
      h = history.history
      assert_almost_equal(h['val_loss'], h['val_weighted_' + loss_full_name], decimal=decimal)
  
      model.train_on_batch(x_train[:32], y_train[:32],
                           sample_weight=sample_weight[:32])
      model.test_on_batch(x_train[:32], y_train[:32],
                          sample_weight=sample_weight[:32])
  
      test_sample_weight = np.ones((y_test.shape[0])) * standard_weight
      test_sample_weight[test_ids] = high_weight
  
      scores = model.evaluate(x_test, y_test, verbose=0, sample_weight=test_sample_weight)
      loss_score, metric_score, weighted_metric_score = scores
  
      assert loss_score < standard_score_sequential
      assert loss_score != metric_score
      assert_almost_equal(loss_score, weighted_metric_score, decimal=decimal)
  
  
  @keras_test
  def test_weighted_metrics_with_no_sample_weight():
      decimal = decimal_precision[K.backend()]
  
      model = create_sequential_model()
      model.compile(loss=loss, optimizer='rmsprop', metrics=[loss], weighted_metrics=[loss])
  
      (x_train, y_train), (x_test, y_test), _ = _get_test_data()
  
      history = model.fit(x_train, y_train, batch_size=batch_size,
                          epochs=epochs // 3, verbose=0)
  
      h = history.history
      assert_array_almost_equal(h['loss'], h[loss_full_name], decimal=decimal)
      assert_array_almost_equal(h['loss'], h['weighted_' + loss_full_name], decimal=decimal)
  
      history = model.fit(x_train, y_train, batch_size=batch_size,
                          epochs=epochs // 3, verbose=0, validation_split=0.1)
  
      h = history.history
      assert_array_almost_equal(h['val_loss'], h['val_' + loss_full_name], decimal=decimal)
      assert_array_almost_equal(h['val_loss'], h['val_weighted_' + loss_full_name], decimal=decimal)
  
      model.train_on_batch(x_train[:32], y_train[:32])
      model.test_on_batch(x_train[:32], y_train[:32])
  
      scores = model.evaluate(x_test, y_test, verbose=0)
      loss_score, metric_score, weighted_metric_score = scores
  
      assert_almost_equal(loss_score, metric_score, decimal=decimal)
      assert_almost_equal(loss_score, weighted_metric_score, decimal=decimal)
  
  
  @keras_test
  def test_weighted_metrics_with_weighted_accuracy_metric():
      model = create_sequential_model()
      model.compile(loss=loss, optimizer='rmsprop', metrics=['acc'], weighted_metrics=['acc'])
  
      (x_train, y_train), _, (sample_weight, _, _) = _get_test_data()
  
      history = model.fit(x_train, y_train, batch_size=batch_size,
                          epochs=epochs // 3, verbose=0,
                          sample_weight=sample_weight)
  
      assert history.history['acc'] != history.history['weighted_acc']
  
  
  @keras_test
  def test_weighted_metrics_with_multiple_outputs():
      decimal = decimal_precision[K.backend()]
  
      inputs = Input(shape=(5,))
      x = Dense(5)(inputs)
      output1 = Dense(1, name='output1')(x)
      output2 = Dense(1, name='output2')(x)
  
      model = Model(inputs=inputs, outputs=[output1, output2])
  
      metrics = {'output1': [loss], 'output2': [loss]}
      weighted_metrics = {'output2': [loss]}
      loss_map = {'output1': loss, 'output2': loss}
  
      model.compile(loss=loss_map, optimizer='sgd', metrics=metrics, weighted_metrics=weighted_metrics)
  
      x = np.array([[1, 1, 1, 1, 1]])
      y = {'output1': np.array([0]), 'output2': np.array([1])}
      weight = 5
  
      history = model.fit(x, y, sample_weight={'output2': np.array([weight])})
  
      unweighted_metric = history.history['output2_' + loss_full_name][0]
      weighted_metric = history.history['output2_weighted_' + loss_full_name][0]
  
      assert_almost_equal(unweighted_metric * weight, weighted_metric, decimal=decimal)
  
  
  @keras_test
  def test_class_weight_wrong_classes():
      model = create_sequential_model()
      model.compile(loss=loss, optimizer='rmsprop')
  
      (x_train, y_train), (x_test, y_test), (sample_weight, class_weight, test_ids) = _get_test_data()
  
      del class_weight[1]
      with pytest.raises(ValueError):
          model.fit(x_train, y_train,
                    epochs=0, verbose=0, class_weight=class_weight)
  
  
  if __name__ == '__main__':
      pytest.main([__file__])
  