|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from itertools import product |
| 4 | + |
| 5 | +import numpy as np |
| 6 | +import pandas as pd |
| 7 | +import pytest |
| 8 | +from anndata import AnnData |
| 9 | +from scipy.sparse import csr_matrix # noqa: TID251 |
| 10 | + |
| 11 | +import scanpy as sc |
| 12 | +from scanpy.preprocessing._combat import _design_matrix |
| 13 | + |
| 14 | + |
| 15 | +def _categorical_example(n_covariates): |
| 16 | + combinations = list(product(["batch_a", "batch_b"], ["a", "b", "c"], ["x", "y"])) |
| 17 | + obs = pd.DataFrame(combinations * 4, columns=["batch", "condition", "donor"]) |
| 18 | + obs.index = obs.index.astype(str) |
| 19 | + for key in obs: |
| 20 | + obs[key] = obs[key].astype("category") |
| 21 | + rng = np.random.default_rng(4454) |
| 22 | + obs["age"] = rng.normal(size=len(obs)) |
| 23 | + x = rng.normal(size=(len(obs), 24)) |
| 24 | + x += obs["batch"].cat.codes.to_numpy()[:, None] * np.linspace(0.5, 2, 24) |
| 25 | + x += obs["condition"].cat.codes.to_numpy()[:, None] * np.linspace(0.3, 1, 24) |
| 26 | + categories = ["condition", "donor"][:n_covariates] |
| 27 | + numeric = pd.concat( |
| 28 | + [ |
| 29 | + obs[["batch", "age"]], |
| 30 | + pd.get_dummies(obs[categories], drop_first=True, dtype=float), |
| 31 | + ], |
| 32 | + axis=1, |
| 33 | + ) |
| 34 | + return AnnData(x, obs=obs), numeric, categories |
| 35 | + |
| 36 | + |
| 37 | +@pytest.mark.parametrize("n_covariates", [1, 2]) |
| 38 | +def test_categorical_design_is_full_rank(n_covariates): |
| 39 | + adata, numeric, categories = _categorical_example(n_covariates) |
| 40 | + model = adata.obs[["batch", *categories, "age"]] |
| 41 | + levels = adata.obs["batch"].cat.categories |
| 42 | + design = _design_matrix(model, "batch", levels) |
| 43 | + expected = _design_matrix(numeric, "batch", levels) |
| 44 | + assert design.shape == expected.shape |
| 45 | + assert np.linalg.matrix_rank(design) == design.shape[1] |
| 46 | + expected_columns = [*expected.columns[:2], *numeric.columns[2:], "age"] |
| 47 | + np.testing.assert_allclose(design.to_numpy(), expected[expected_columns].to_numpy()) |
| 48 | + |
| 49 | + |
| 50 | +@pytest.mark.parametrize("n_covariates", [1, 2]) |
| 51 | +@pytest.mark.parametrize("sparse", [False, True]) |
| 52 | +def test_categorical_combat_matches_reference_coding(n_covariates, sparse): |
| 53 | + adata, numeric, categories = _categorical_example(n_covariates) |
| 54 | + x = adata.X.copy() |
| 55 | + if sparse: |
| 56 | + adata.X = csr_matrix(adata.X) |
| 57 | + reference = AnnData(x.copy(), obs=numeric) |
| 58 | + expected = sc.pp.combat( |
| 59 | + reference, key="batch", covariates=list(numeric.columns[1:]), inplace=False |
| 60 | + ) |
| 61 | + actual = sc.pp.combat( |
| 62 | + adata, key="batch", covariates=[*categories, "age"], inplace=False |
| 63 | + ) |
| 64 | + assert np.isfinite(actual).all() |
| 65 | + np.testing.assert_allclose(actual, expected, rtol=1e-10, atol=1e-10) |
| 66 | + np.testing.assert_array_equal(adata.X.toarray() if sparse else adata.X, x) |
| 67 | + sc.pp.combat(adata, key="batch", covariates=[*categories, "age"]) |
| 68 | + np.testing.assert_allclose(adata.X, expected, rtol=1e-10, atol=1e-10) |
| 69 | + |
| 70 | + |
| 71 | +def test_single_level_categorical_covariate_matches_batch_only(): |
| 72 | + adata, _, _ = _categorical_example(1) |
| 73 | + adata.obs["constant"] = pd.Categorical(["same"] * adata.n_obs) |
| 74 | + expected = sc.pp.combat(adata, key="batch", inplace=False) |
| 75 | + actual = sc.pp.combat(adata, key="batch", covariates=["constant"], inplace=False) |
| 76 | + np.testing.assert_allclose(actual, expected, rtol=1e-10, atol=1e-10) |
0 commit comments