def test_differential_expression(save_path):
    dataset = CortexDataset(save_path=save_path)
    n_cells = len(dataset)
    all_indices = np.arange(n_cells)
    vae = VAE(dataset.nb_genes, dataset.n_batches)
    trainer = UnsupervisedTrainer(vae, dataset, train_size=0.5, use_cuda=use_cuda)
    trainer.train(n_epochs=2)
    post = trainer.create_posterior(vae, dataset, shuffle=False, indices=all_indices)
    with tempfile.TemporaryDirectory() as temp_dir:
        posterior_save_path = os.path.join(temp_dir, 'posterior_data')
        post = post.sequential(batch_size=3)
        post.save_posterior(posterior_save_path)
        new_vae = VAE(dataset.nb_genes, dataset.n_batches)
        new_post = load_posterior(posterior_save_path, model=new_vae, use_cuda=False)
    assert (new_post.data_loader.batch_size == 3)
    assert np.array_equal(new_post.indices, post.indices)
    assert np.array_equal(new_post.gene_dataset.X, post.gene_dataset.X)
    px_scales = post.scale_sampler(n_samples_per_cell=4, n_samples=None, selection=all_indices)['scale']
    assert (px_scales.shape[1] == dataset.nb_genes), 'posterior scales should have shape (n_samples, n_genes)'
    idx_1 = [1, 2, 3]
    idx_2 = [4, 5, 6, 7]
    de_dataframe = post.differential_expression_score(idx1=idx_1, idx2=idx_2, n_samples=10, mode='vanilla', use_permutation=True, M_permutation=100)
    de_dataframe = post.differential_expression_score(idx1=idx_1, idx2=idx_2, n_samples=10, mode='change', use_permutation=True, M_permutation=100, cred_interval_lvls=[0.5, 0.95])
    print(de_dataframe.keys())
    assert (de_dataframe['lfc_confidence_interval_0.5_min'] <= de_dataframe['lfc_confidence_interval_0.5_max']).all()
    assert (de_dataframe['lfc_confidence_interval_0.95_min'] <= de_dataframe['lfc_confidence_interval_0.95_max']).all()
    de_probabilities = de_dataframe.loc[:, 'proba_de']
    assert ((0.0 <= de_probabilities) & (de_probabilities <= 1.0)).all()
    sp = os.path.join(save_path, '10X')
    dataset = Dataset10X(dataset_name='pbmc_10k_protein_v3', save_path=sp)
    n_cells = len(dataset)
    all_indices = np.arange(n_cells)
    vae = TOTALVI(dataset.nb_genes, len(dataset.protein_names), n_batch=dataset.n_batches)
    trainer = TotalTrainer(vae, dataset, train_size=0.5, use_cuda=use_cuda, early_stopping_kwargs=None)
    trainer.train(n_epochs=2)
    post = trainer.create_posterior(vae, dataset, shuffle=False, indices=all_indices, type_class=TotalPosterior)
    idx_1 = [1, 2, 3]
    idx_2 = [4, 5, 6, 7]
    de_dataframe = post.differential_expression_score(idx1=idx_1, idx2=idx_2, n_samples=10, mode='vanilla', use_permutation=True, M_permutation=100)
    de_dataframe = post.differential_expression_score(idx1=idx_1, idx2=idx_2, n_samples=10, mode='change', use_permutation=True, M_permutation=100)
def test_differential_expression(save_path):
    dataset = CortexDataset(save_path=save_path)
    n_cells = len(dataset)
    all_indices = np.arange(n_cells)
    vae = VAE(dataset.nb_genes, dataset.n_batches)
    trainer = UnsupervisedTrainer(vae, dataset, train_size=0.5, use_cuda=use_cuda)
    trainer.train(n_epochs=2)
    post = trainer.create_posterior(vae, dataset, shuffle=False, indices=all_indices)
    with tempfile.TemporaryDirectory() as temp_dir:
        posterior_save_path = os.path.join(temp_dir, 'posterior_data')
        post = post.sequential(batch_size=3)
        post.save_posterior(posterior_save_path)
        new_vae = VAE(dataset.nb_genes, dataset.n_batches)
        new_post = load_posterior(posterior_save_path, model=new_vae, use_cuda=False)
    assert (new_post.data_loader.batch_size == 3)
    assert np.array_equal(new_post.indices, post.indices)
    assert np.array_equal(new_post.gene_dataset.X, post.gene_dataset.X)
    px_scales = post.scale_sampler(n_samples_per_cell=4, n_samples=None, selection=all_indices)['scale']
    assert (px_scales.shape[1] == dataset.nb_genes), 'posterior scales should have shape (n_samples, n_genes)'
    idx_1 = [1, 2, 3]
    idx_2 = [4, 5, 6, 7]
    de_dataframe = post.differential_expression_score(idx1=idx_1, idx2=idx_2, n_samples=10, mode='vanilla', use_permutation=True, M_permutation=100)
    de_dataframe = post.differential_expression_score(idx1=idx_1, idx2=idx_2, n_samples=10, mode='change', use_permutation=True, M_permutation=100, cred_interval_lvls=[0.5, 0.95])
    print(de_dataframe.keys())
    assert (de_dataframe['lfc_confidence_interval_0.5_min'] <= de_dataframe['lfc_confidence_interval_0.5_max']).all()
    assert (de_dataframe['lfc_confidence_interval_0.95_min'] <= de_dataframe['lfc_confidence_interval_0.95_max']).all()
    de_probabilities = de_dataframe.loc[:, 'proba_de']
    assert ((0.0 <= de_probabilities) & (de_probabilities <= 1.0)).all()
    sp = os.path.join(save_path, '10X')
    dataset = Dataset10X(dataset_name='pbmc_10k_protein_v3', save_path=sp)
    n_cells = len(dataset)
    all_indices = np.arange(n_cells)
    vae = TOTALVI(dataset.nb_genes, len(dataset.protein_names), n_batch=dataset.n_batches)
    trainer = TotalTrainer(vae, dataset, train_size=0.5, use_cuda=use_cuda, early_stopping_kwargs=None)
    trainer.train(n_epochs=2)
    post = trainer.create_posterior(vae, dataset, shuffle=False, indices=all_indices, type_class=TotalPosterior)
    idx_1 = [1, 2, 3]
    idx_2 = [4, 5, 6, 7]
    de_dataframe = post.differential_expression_score(idx1=idx_1, idx2=idx_2, n_samples=10, mode='vanilla', use_permutation=True, M_permutation=100)
    de_dataframe = post.differential_expression_score(idx1=idx_1, idx2=idx_2, n_samples=10, mode='change', use_permutation=True, M_permutation=100)