-
Notifications
You must be signed in to change notification settings - Fork 0
/
TopACT.py
81 lines (59 loc) · 2.16 KB
/
TopACT.py
1
2
3
4
5
6
7
8
9
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
#!/usr/bin/env python
def main():
import argparse
from pathlib import Path
parser = argparse.ArgumentParser(description="")
parser.add_argument("training", help="Path to h5ad file for training.", type=Path)
parser.add_argument("spatial", help="Path of spatial data (tsv).", type=Path)
parser.add_argument("out", help="Path of output directory", type=Path)
parser.add_argument(
"--n_processes",
help="Number of processes for classification",
type=int,
default=1,
required=False,
)
args = parser.parse_args()
import shutil
from tempfile import TemporaryDirectory
import pandas as pd
from skimage.morphology import convex_hull_image
from topact.spatial import CountGrid
df_spatial = pd.read_table(args.spatial, comment="#", dtype={"geneID": "category"})
clf, genes = train(args.training, df_spatial["geneID"].cat.categories)
sd = CountGrid.from_coord_table(
df_spatial,
genes=genes,
count_col="MIDCounts",
gene_col="geneID",
)
mask = convex_hull_image(sd.count_matrix())
with TemporaryDirectory(suffix="_topact") as dir:
tmp_file = Path(dir) / "confidence.npy"
sd.classify_parallel(
clf,
min_scale=3,
max_scale=9,
num_proc=args.n_processes,
outfile=str(tmp_file),
mask=mask,
verbose=False,
)
shutil.move(tmp_file, args.out / "confidence.npy")
def train(h5ad_path, genes, label_col="subclass_label"):
import anndata as ad
from sklearn.preprocessing import normalize
from topact.classifier import SVCClassifier, train_from_countmatrix
from topact.countdata import CountMatrix
adata = ad.read_h5ad(h5ad_path)
adata = adata[:, adata.var_names.isin(genes)]
mtx = normalize(adata.X, norm="l1")
genes = adata.var_names.tolist()
celltypes = adata.obs[label_col]
sc = CountMatrix(mtx, genes=genes)
sc.add_metadata("celltype", celltypes)
clf = SVCClassifier()
train_from_countmatrix(clf, sc, "celltype")
return clf, adata.var_names.tolist()
if __name__ == "__main__":
main()