Skip to content

Solver

solver_mixupvi_module(model, signature_latent_space, adata, layer, device='cpu')

Placeholder for MixupVI module solver.

Source code in src/pydeconv/model/solver/nn.py 
37
38
39
40
41
42
43
44
45
46
47
def solver_mixupvi_module(
    model: torch.nn.Module, signature_latent_space: SignatureMatrix, adata: AnnData, layer: str, device: str = "cpu"
):
    """Placeholder for MixupVI module solver."""
    x = adata.to_df(layer=layer)
    x = torch.tensor(x.values, dtype=torch.float32).to(device)
    model.eval()
    latent_features = model.extract_latent(x)
    anndata_latent = AnnData(latent_features, layers={"latent_features": latent_features})
    cell_prop_ = solver_nnls(anndata_latent, signature_latent_space, layer="latent_features")
    return cell_prop_

solver_nnls(adata, signature_matrix, layer)

Solve the linear regression using non-negative least squares.

Parameters:

Name Type Description Default
adata AnnData

Anndata object containing the data.

 required
signature_matrix SignatureMatrix

Signature matrix object.

 required
layer str

Layer to use for the regressions.

 required

Returns:

Name Type Description
coef_ ndarray

Coefficients of the linear regression.
Source code in src/pydeconv/model/solver/linear.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def solver_nnls(adata: AnnData, signature_matrix: SignatureMatrix, layer: str):
    """
    Solve the linear regression using non-negative least squares.

    Parameters
    ----------
    adata : AnnData
        Anndata object containing the data.
    signature_matrix : SignatureMatrix
        Signature matrix object.
    layer : str
        Layer to use for the regressions.

    Returns
    -------
    coef_ : np.ndarray
        Coefficients of the linear regression.
    """
    X = signature_matrix.values
    Y = adata.to_df(layer=layer).T

    coef_ = np.vstack([optimize.nnls(X, yi)[0] for yi in Y.T.values])
    return coef_

solver_nusvr(adata, signature_matrix, layer, norm=True, scale=False)

Solve the linear regression using ordinary least squares

Parameters:

Name Type Description Default
adata AnnData

Anndata object containing the data

 required
signature_matrix SignatureMatrix

Signature matrice object

 required
layer str

Layer to use for the regression

 required
norm bool

If True, the data will be normalized to the range [-1, 1]

 True
scale bool

If True, the data will be scaled to have zero mean and unit variance.

 False

Returns:

Name Type Description
coef_ ndarray

Coefficients of the linear regression
Source code in src/pydeconv/model/solver/svm.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def solver_nusvr(
    adata: AnnData,
    signature_matrix: SignatureMatrix,
    layer: str,
    norm: bool = True,
    scale=False,
) -> np.ndarray:
    """Solve the linear regression using ordinary least squares

    Parameters
    ----------
    adata : AnnData
        Anndata object containing the data
    signature_matrix : SignatureMatrix
        Signature matrice object
    layer : str
        Layer to use for the regression
    norm : bool
        If True, the data will be normalized to the range [-1, 1]
    scale : bool
        If True, the data will be scaled to have zero mean and unit variance.

    Returns
    -------

    coef_ : np.ndarray
        Coefficients of the linear regression
    """
    model = svm.NuSVR(nu=0.5, gamma="auto", kernel="linear", C=1.0, cache_size=200)

    X = signature_matrix.values  # row: gene, column: cell type
    y = adata.to_df(layer=layer).T  # row: gene, column: samples

    # Normalize X and y to the range [-1, 1]
    if norm:
        # errited from R code
        X_max = np.max(X)
        X_min = np.min(X)
        upper_bounds = [np.max([X_max, y_max]) for y_max in y.max(axis=0).values]
        lower_bounds = [np.min([X_min, y_min]) for y_min in y.min(axis=0).values]

        X_norm = [
            ((X - lower_bound) / upper_bound) * 2 - 1
            for lower_bound, upper_bound in zip(lower_bounds, upper_bounds, strict=True)
        ]
        y_norm = [
            ((y[col] - lower_bound) / upper_bound) * 2 - 1
            for lower_bound, upper_bound, col in zip(lower_bounds, upper_bounds, y.columns, strict=True)
        ]

        if scale:
            X_norm_scale = [
                pd.DataFrame(StandardScaler().transform(X_norm_col), columns=X.columns, index=X.index)
                for X_norm_col in X_norm
            ]
            y_norm_scale = [StandardScaler().transform(y_norm_col.to_frame()) for y_norm_col in y_norm]
            coeffs = np.array(
                [
                    model.fit(X_norm_col, y_norm_col).coef_
                    for X_norm_col, y_norm_col in zip(X_norm_scale, y_norm_scale, strict=True)
                ]
            ).squeeze()
        else:
            coeffs = np.array(
                [model.fit(X_norm_col, y_norm_col).coef_ for X_norm_col, y_norm_col in zip(X_norm, y_norm, strict=True)]
            ).squeeze()

    else:
        if scale:
            X = pd.DataFrame(StandardScaler().transform(X), columns=X.columns, index=X.index)
            y = pd.DataFrame(StandardScaler().transform(y), columns=y.columns, index=y.index)

        coeffs = np.array([model.fit(X, y[col]).coef_ for col in y.columns]).squeeze()

    return coeffs

solver_ols(adata, signature_matrix, layer)

Solve the linear regression using ordinary least squares.

Parameters:

Name Type Description Default
adata AnnData

Anndata object containing the data.

 required
signature_matrix SignatureMatrix

Signature matrix object.

 required
layer str

Layer to use for the regression.

 required

Returns:

Name Type Description
coef_ ndarray

Coefficients of the linear regression.
Source code in src/pydeconv/model/solver/linear.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def solver_ols(adata: AnnData, signature_matrix: SignatureMatrix, layer: str):
    """
    Solve the linear regression using ordinary least squares.

    Parameters
    ----------
    adata : AnnData
        Anndata object containing the data.
    signature_matrix : SignatureMatrix
        Signature matrix object.
    layer : str
        Layer to use for the regression.

    Returns
    -------
    coef_ : np.ndarray
        Coefficients of the linear regression.
    """
    X = signature_matrix.values
    Y = adata.to_df(layer=layer).T

    coef_, _, _, _ = linalg.lstsq(X, Y)
    coef_ = coef_.T

    return coef_

solver_rlr(adata, signature_matrix, layer)

Solve the linear regression using robust linear regression. and huber loss function.

Parameters:

Name Type Description Default
adata AnnData

Anndata object containing the data.

 required
signature_matrix SignatureMatrix

Signature matrix object.

 required
layer str

Layer to use for the regression.

 required

Returns:

Name Type Description
coef_ ndarray

Coefficients of the linear regression.
Source code in src/pydeconv/model/solver/linear.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def solver_rlr(adata: AnnData, signature_matrix: SignatureMatrix, layer: str):
    """
    Solve the linear regression using robust linear regression. and huber loss function.

    Parameters
    ----------
    adata : AnnData
        Anndata object containing the data.
    signature_matrix : SignatureMatrix
        Signature matrix object.
    layer : str
        Layer to use for the regression.

    Returns
    -------
    coef_ : np.ndarray
        Coefficients of the linear regression.
    """
    X = signature_matrix.values
    Y = adata.to_df(layer=layer).T

    coef_ = np.vstack([sm.RLM(y, X, M=sm.robust.norms.HuberT()).fit().params for y in Y.T.values])
    return coef_

solver_svr(adata, signature_matrix, layer)

Solve the linear regression using ordinary least squares

Parameters:

Name Type Description Default
adata AnnData

Anndata object containing the data

 required
signature_matrix SignatureMatrix

Signature matrice object

 required
layer str

Layer to use for the regression

 required

Returns:

Name Type Description
coef_ ndarray

Coefficients of the linear regression
Source code in src/pydeconv/model/solver/svm.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def solver_svr(adata: AnnData, signature_matrix: SignatureMatrix, layer: str) -> np.ndarray:
    """Solve the linear regression using ordinary least squares

    Parameters
    ----------

    adata : AnnData
        Anndata object containing the data
    signature_matrix : SignatureMatrix
        Signature matrice object
    layer : str
        Layer to use for the regression

    Returns
    -------

    coef_ : np.ndarray
        Coefficients of the linear regression
    """
    # model = make_pipeline(StandardScaler(), svm.NuSVR(nu=0.5, gamma="auto", kernel="linear", C=1.0, cache_size=40))
    model = svm.SVR(kernel="linear")

    X = signature_matrix.values  # row: gene, column: cell type
    y = adata.to_df(layer=layer).T  # row: gene, column: samples

    coeffs = np.array([model.fit(X, y[col]).coef_ for col in y.columns]).squeeze()

    return coeffs

solver_torch_module(model, adata, layer, device='cpu')

Solve the deconvolution problem using a torch model.

Parameters:

Name Type Description Default
model Module

The model to use for deconvolution.

 required
adata AnnData

Anndata object containing the data.

 required
layer str

Layer of the data to use for deconvolution.

 required
device str

Device to use for the computation, by default "cpu"

 'cpu'
Source code in src/pydeconv/model/solver/nn.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def solver_torch_module(model: torch.nn.Module, adata: AnnData, layer: str, device: str = "cpu"):
    """Solve the deconvolution problem using a torch model.

    Parameters
    ----------
    model : torch.nn.Module
        The model to use for deconvolution.
    adata : AnnData
        Anndata object containing the data.
    layer : str
        Layer of the data to use for deconvolution.
    device : str, optional
        Device to use for the computation, by default "cpu"
    """
    model.eval()
    x = adata.to_df(layer=layer)
    x = torch.tensor(x.values, dtype=torch.float32).to(device)
    model = model.to(device)
    # can't be use with torch no grad context, because of optional adaptative step
    cell_prop_ = model(x)
    cell_prop_ = cell_prop_.cpu().detach().numpy()
    return cell_prop_

solver_wls(adata, signature_matrix, layer, max_iter=1000, tol=0.01, dampened=None, solver_func=solver_nnls, parallel=False)

Solve the linear regression using Weighted Least Squares.

Parameters:

Name Type Description Default
adata AnnData

Anndata object containing the data.

 required
signature_matrix SignatureMatrix

Signature matrix object.

 required
layer str

Layer to use for the regression.

 required
max_iter int

Maximum number of iterations, by default 1000.

 1000
tol float

Tolerance for the convergence, by default 0.01.

 0.01
dampened Union[list, Literal['auto'], None]

Dampening constant for the weights. If None, no dampening is applied. If "auto", the dampening constant is calculated automatically, by default None.

 None
solver_func Callable

Function to solve the linear regression, by default solver_nnls.

 solver_nnls
parallel bool

Whether to run the solver in parallel, by default False.

 False

Returns:

Name Type Description
coef_ ndarray

Coefficients of the linear regression.
Source code in src/pydeconv/model/solver/linear.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def solver_wls(
    adata: AnnData,
    signature_matrix: SignatureMatrix,
    layer: str,
    max_iter: int = 1000,
    tol: float = 0.01,
    dampened: Union[list, Literal["auto"], None] = None,
    solver_func=solver_nnls,
    parallel: bool = False,
):
    """
    Solve the linear regression using Weighted Least Squares.

    Parameters
    ----------
    adata : AnnData
        Anndata object containing the data.
    signature_matrix : SignatureMatrix
        Signature matrix object.
    layer : str
        Layer to use for the regression.
    max_iter : int, optional
        Maximum number of iterations, by default 1000.
    tol : float, optional
        Tolerance for the convergence, by default 0.01.
    dampened : Union[list, Literal["auto"], None], optional
        Dampening constant for the weights. If None, no dampening is applied.
        If "auto", the dampening constant is calculated automatically, by default None.
    solver_func : Callable, optional
        Function to solve the linear regression, by default solver_nnls.
    parallel : bool, optional
        Whether to run the solver in parallel, by default False.

    Returns
    -------
    coef_ : np.ndarray
        Coefficients of the linear regression.
    """

    X = signature_matrix.values.copy()  # row: gene, column: cell type
    Y = adata.to_df(layer=layer).T.copy()  # row: genes, column: samples

    coef_init = solver_func(adata, signature_matrix, layer).T  # row: celltypes, column: samples

    # Define dampening constants
    if dampened == "auto":
        dampened = [find_dampening_constant(X, y, coef) for (_, y), coef in zip(Y.items(), coef_init.T, strict=True)]
    elif isinstance(dampened, float):
        dampened = np.array([dampened] * Y.shape[1])
    elif dampened is None:
        dampened = [None] * Y.shape[1]
    if len(dampened) != Y.shape[1]:
        raise ValueError("Dampening constant must be a float or a list of floats with the same length as the samples")

    # Solve the weighted linear regression
    if parallel:
        coef_ = solver_wls_parallel(X, Y, coef_init, max_iter=max_iter, tol=tol, dampened=dampened)
    else:
        coef_ = [
            solver_wls_per_patient(X, y, c, d, max_iter, tol)
            for (_, y), c, d in tqdm(
                zip(Y.items(), coef_init.T, dampened, strict=True), total=len(Y), position=3, leave=False
            )
        ]
    coef_ = np.array(coef_)

    return coef_  # output: row: samples, column: cell types