test_keras_backend.py 文件源码

def test_experiment_fit_async(self, get_model, get_loss_metric,
                                  get_custom_l, get_callback_fix):
        new_session()
        data, data_val = make_data(train_samples, test_samples)
        model, metrics, cust_objects = prepare_model(get_model(get_custom_l),
                                                     get_loss_metric,
                                                     get_custom_l)

        cust_objects['test_list'] = [1, 2]
        expe = Experiment(model)

        expected_value = 2
        for mod in [None, model]:
            for data_val_loc in [None, data_val]:
                _, thread = expe.fit_async([data], [data_val_loc],
                                           model=mod, nb_epoch=2,
                                           batch_size=batch_size,
                                           metrics=metrics,
                                           custom_objects=cust_objects,
                                           overwrite=True,
                                           verbose=2,
                                           callbacks=get_callback_fix)

                thread.join()

                for k in expe.full_res['metrics']:
                    if 'iter' not in k:
                        assert len(
                            expe.full_res['metrics'][k]) == expected_value

                if data_val_loc is not None:
                    for k in expe.full_res['metrics']:
                        if 'val' in k and 'iter' not in k:
                            assert None not in expe.full_res['metrics'][k]
                else:
                    for k in expe.full_res['metrics']:
                        if 'val' in k and 'iter' not in k:
                            assert all([np.isnan(v)
                                        for v in expe.full_res['metrics'][k]])

        if K.backend() == 'tensorflow':
            K.clear_session()

        print(self)