-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
127 lines (118 loc) · 6.2 KB
/
main.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
82
83
84
85
86
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
115
116
117
118
119
120
121
122
123
124
125
126
127
from fastapi import FastAPI, Request
from utils import utils
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.encoders import jsonable_encoder
import copy
# Note: dir_name2dd_df itself is not returned by the API, but is used for filtering by genes and cell types
# c.f. /data/{project}/{viz} below
(dir_name2project_data, dir_name2file_name2df) = utils.get_projects()
api = FastAPI()
# List projects
@api.get("/list")
def list_projects(request: Request):
case_examples = ['CaseExample3_specificityScoring', 'CaseExample2_spatialNiches', 'CaseExample1_differentiation']
projectName2Title = dict(zip(dir_name2project_data.keys(),
[x['title'] for x in dir_name2project_data.values()]))
#If request url is our main cpdbviz site, only list exemplar data. Otherwise list everything available.
if "www.cellphonedb.org" in str(request.url):
projectName2Title = {case_example: projectName2Title[case_example] for case_example in case_examples if case_example in projectName2Title}
return dict(sorted(projectName2Title.items(), key=lambda item: item[1]))
@api.get("/data/{project}/{viz}")
def get_viz_data(project: str,
viz: str,
genes: str = None,
interacting_pairs: str = None,
classes: str = None,
modalities: str = None,
min_score: str = 0,
cell_types: str = None,
cell_type_pairs: str = None,
microenvironments: str = None,
refresh_plot: bool = False,
values_to_show: str = 'means',
interacting_pairs_selection_logic: str = None,
sort_interacting_pairs_alphabetically: bool = False
):
if viz == 'single_gene_expression':
selected_genes = get_jsonable(genes)
selected_cell_types = get_jsonable(cell_types)
ret = copy.deepcopy(dir_name2project_data[project])
utils.populate_deconvoluted_data(ret, dir_name2file_name2df[project]['deconvoluted_result'], \
selected_genes = selected_genes, selected_cell_types = selected_cell_types,
refresh_plot = refresh_plot, percents = False)
dict_sge = ret['single_gene_expression']
if 'deconvoluted_percents' in dir_name2file_name2df[project]:
utils.populate_deconvoluted_data(ret, dir_name2file_name2df[project]['deconvoluted_percents'], \
# The following ensures that the same genes and cell types are used to
# filter deconvoluted_percents as were used to filter deconvoluted_result
selected_genes=dict_sge['genes'], \
selected_cell_types=dict_sge['cell_types'], \
refresh_plot=refresh_plot, percents = True)
if refresh_plot:
# Autocompletes are initialised on first load only - hence on refresh_plot we avoid
# bulking-up the API output unnecessarily
dict_sge.pop('all_genes')
ret = dict_sge
elif viz == 'cell_cell_interaction_search':
selected_genes = get_jsonable(genes)
selected_interacting_pairs = get_jsonable(interacting_pairs)
selected_cell_types = get_jsonable(cell_types)
selected_cell_type_pairs = get_jsonable(cell_type_pairs)
selected_microenvironments = get_jsonable(microenvironments)
selected_classes = get_jsonable(classes)
selected_modalities = get_jsonable(modalities)
ret = copy.deepcopy(dir_name2project_data[project][viz])
if refresh_plot:
# Autocompletes are initialised on first load only - hence on refresh_plot
# we avoid bulking-up the API output unnecessarily
ret.pop('all_genes')
ret.pop('all_interacting_pairs')
utils.filter_interactions_for_cci_search(ret, dir_name2file_name2df[project],
selected_genes, selected_interacting_pairs, selected_classes, selected_cell_types,
selected_cell_type_pairs, selected_microenvironments, refresh_plot, values_to_show,
interacting_pairs_selection_logic, sort_interacting_pairs_alphabetically)
# 'analysis_means' is used for pre-selecting interacting pairs - it is not needed by the front end
ret.pop('analysis_means')
# cellphonedb is needed for retrieving properties of interacting pairs but it is not used by the front end directly
if 'cellphonedb' in ret:
ret.pop('cellphonedb')
elif viz == 'cell_cell_interaction_summary':
selected_classes = get_jsonable(classes)
selected_modalities = get_jsonable(modalities)
ret = copy.deepcopy(dir_name2project_data[project][viz])
significant_interactions_only = True
utils.filter_interactions_for_cci_summary(
ret, dir_name2file_name2df[project], selected_classes,
selected_modalities, int(min_score), significant_interactions_only)
else:
ret = dir_name2project_data[project][viz]
return ret
@api.get("/generate/hash")
def generate_random_hash() -> str:
return utils.generate_random_hash()
@api.get("/validate/{project_id}")
def validate_projectid(project_id: str) -> bool:
ret = False
if project_id in dir_name2project_data:
ret = True
return ret
@api.get("/validate/auth/{project_id}")
def validate_auth(project_id: str, auth: str) -> bool:
ret = False
if 'hash' in dir_name2project_data[project_id]:
if hash is not None and auth == dir_name2project_data[project_id]['hash']:
ret = True
else:
ret = True
return ret
def get_jsonable(val: str) -> list:
ret = []
if jsonable_encoder(val):
ret = jsonable_encoder(val).split(",")
return ret
app = FastAPI()
# See: https://fastapi.tiangolo.com/tutorial/cors/
app.add_middleware(CORSMiddleware, allow_origins=["*"])
app.mount("/api", api)
app.mount("/", StaticFiles(directory="public", html = True), name="public")