dag9_tenx_single_cell_immune_profiling.py 46 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092
  1. from datetime import timedelta
  2. import os,json,logging,subprocess
  3. from airflow.models import DAG,Variable
  4. from airflow.utils.dates import days_ago
  5. from airflow.operators.python_operator import PythonOperator
  6. from airflow.operators.python_operator import BranchPythonOperator
  7. from airflow.operators.dummy_operator import DummyOperator
  8. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import fetch_analysis_info_and_branch_func
  9. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import configure_cellranger_run_func
  10. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import run_sc_read_trimmming_func
  11. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import run_cellranger_tool
  12. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import decide_analysis_branch_func
  13. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import load_cellranger_result_to_db_func
  14. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import ftp_files_upload_for_analysis
  15. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import irods_files_upload_for_analysis
  16. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import run_singlecell_notebook_wrapper_func
  17. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import load_analysis_files_func
  18. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import task_branch_function
  19. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import upload_analysis_file_to_box
  20. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import convert_bam_to_cram_func
  21. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import run_picard_for_cellranger
  22. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import run_samtools_for_cellranger
  23. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import run_multiqc_for_cellranger
  24. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import index_and_copy_bam_for_parallel_analysis
  25. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import change_pipeline_status
  26. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import clean_up_files
  27. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import create_and_update_qc_pages
  28. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import load_cellranger_metrices_to_collection
  29. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import generate_cell_sorted_bam_func
  30. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import run_velocyto_func
  31. from igf_airflow.utils.dag9_tenx_single_cell_immune_profiling_utils import run_scvelo_for_sc_5p_func
  32. ## ARGS
  33. default_args = {
  34. 'owner': 'airflow',
  35. 'depends_on_past': False,
  36. 'start_date': days_ago(2),
  37. 'email_on_failure': False,
  38. 'email_on_retry': False,
  39. 'retries': 4,
  40. 'max_active_runs':10,
  41. 'catchup':True,
  42. 'retry_delay': timedelta(minutes=5),
  43. 'provide_context': True,
  44. }
  45. FEATURE_TYPE_LIST = \
  46. Variable.get('tenx_single_cell_immune_profiling_feature_types',default_var={})#.split(',')
  47. ## DAG
  48. dag = \
  49. DAG(
  50. dag_id='dag9_tenx_single_cell_immune_profiling',
  51. schedule_interval=None,
  52. tags=['hpc','analysis','tenx','sc'],
  53. default_args=default_args,
  54. concurrency=100,
  55. max_active_runs=20,
  56. orientation='LR')
  57. with dag:
  58. ## TASK
  59. fetch_analysis_info_and_branch = \
  60. BranchPythonOperator(
  61. task_id='fetch_analysis_info',
  62. dag=dag,
  63. queue='hpc_4G',
  64. params={'no_analysis_task':'no_analysis',
  65. 'analysis_description_xcom_key':'analysis_description',
  66. 'analysis_info_xcom_key':'analysis_info'},
  67. python_callable=fetch_analysis_info_and_branch_func)
  68. ## TASK
  69. configure_cellranger_run = \
  70. PythonOperator(
  71. task_id='configure_cellranger_run',
  72. dag=dag,
  73. queue='hpc_4G',
  74. trigger_rule='none_failed_or_skipped',
  75. params={'xcom_pull_task_id':'fetch_analysis_info',
  76. 'analysis_description_xcom_key':'analysis_description',
  77. 'analysis_info_xcom_key':'analysis_info',
  78. 'library_csv_xcom_key':'cellranger_library_csv'},
  79. python_callable=configure_cellranger_run_func)
  80. for analysis_name in FEATURE_TYPE_LIST.keys():
  81. ## TASK
  82. task_branch = \
  83. BranchPythonOperator(
  84. task_id=analysis_name,
  85. dag=dag,
  86. queue='hpc_4G',
  87. params={'xcom_pull_task_id':'fetch_analysis_info',
  88. 'analysis_info_xcom_key':'analysis_info',
  89. 'analysis_name':analysis_name,
  90. 'task_prefix':'run_trim'},
  91. python_callable=task_branch_function)
  92. run_trim_list = list()
  93. for run_id in range(0,10):
  94. ## TASK
  95. t = \
  96. PythonOperator(
  97. task_id='run_trim_{0}_{1}'.format(analysis_name,run_id),
  98. dag=dag,
  99. queue='hpc_4G',
  100. params={'xcom_pull_task_id':'fetch_analysis_info',
  101. 'analysis_info_xcom_key':'analysis_info',
  102. 'analysis_description_xcom_key':'analysis_description',
  103. 'analysis_name':analysis_name,
  104. 'run_id':run_id,
  105. 'r1-length':0,
  106. 'r2-length':0,
  107. 'fastq_input_dir_tag':'fastq_dir',
  108. 'use_ephemeral_space':True,
  109. 'fastq_output_dir_tag':'output_path'},
  110. python_callable=run_sc_read_trimmming_func)
  111. run_trim_list.append(t)
  112. ## TASK
  113. collect_trimmed_files = \
  114. DummyOperator(
  115. task_id='collect_trimmed_files_{0}'.format(analysis_name),
  116. trigger_rule='none_failed_or_skipped',
  117. dag=dag)
  118. ## PIPELINE
  119. fetch_analysis_info_and_branch >> task_branch
  120. task_branch >> run_trim_list
  121. run_trim_list >> collect_trimmed_files
  122. collect_trimmed_files >> configure_cellranger_run
  123. ## TASK
  124. no_analysis = \
  125. DummyOperator(
  126. task_id='no_analysis',
  127. dag=dag)
  128. ## PIPELINE
  129. fetch_analysis_info_and_branch >> no_analysis
  130. ## TASK
  131. run_cellranger = \
  132. PythonOperator(
  133. task_id='run_cellranger',
  134. dag=dag,
  135. queue='hpc_64G16t24hr',
  136. params={'analysis_description_xcom_pull_task':'fetch_analysis_info',
  137. 'analysis_description_xcom_key':'analysis_description',
  138. 'library_csv_xcom_key':'cellranger_library_csv',
  139. 'library_csv_xcom_pull_task':'configure_cellranger_run',
  140. 'cellranger_xcom_key':'cellranger_output',
  141. 'cellranger_options':['--localcores 16','--localmem 64']},
  142. python_callable=run_cellranger_tool)
  143. ## PIPELINE
  144. configure_cellranger_run >> run_cellranger
  145. ## TASK
  146. decide_analysis_branch = \
  147. BranchPythonOperator(
  148. task_id='decide_analysis_branch',
  149. dag=dag,
  150. queue='hpc_4G',
  151. python_callable=decide_analysis_branch_func,
  152. params={'load_cellranger_result_to_db_task':'load_cellranger_result_to_db',
  153. 'run_scanpy_for_sc_5p_task':'run_scanpy_for_sc_5p',
  154. 'run_scirpy_for_vdj_task':'run_scirpy_for_vdj',
  155. 'run_scirpy_for_vdj_b_task':'run_scirpy_for_vdj_b',
  156. 'run_scirpy_vdj_t_task':'run_scirpy_for_vdj_t',
  157. 'run_seurat_for_sc_5p_task':'run_seurat_for_sc_5p',
  158. 'convert_cellranger_bam_to_cram_task':'convert_cellranger_bam_to_cram',
  159. 'library_csv_xcom_key':'cellranger_library_csv',
  160. 'library_csv_xcom_pull_task':'configure_cellranger_run'})
  161. ## PIPELINE
  162. run_cellranger >> decide_analysis_branch
  163. ## TASK
  164. load_cellranger_result_to_db = \
  165. PythonOperator(
  166. task_id='load_cellranger_result_to_db',
  167. dag=dag,
  168. queue='hpc_4G',
  169. python_callable=load_cellranger_result_to_db_func,
  170. params={'analysis_description_xcom_pull_task':'fetch_analysis_info',
  171. 'analysis_description_xcom_key':'analysis_description',
  172. 'cellranger_xcom_key':'cellranger_output',
  173. 'cellranger_xcom_pull_task':'run_cellranger',
  174. 'collection_type':'CELLRANGER_MULTI',
  175. 'html_collection_type':'CELLRANGER_HTML',
  176. 'collection_table':'sample',
  177. 'xcom_collection_name_key':'sample_igf_id',
  178. 'genome_column':'genome_build',
  179. 'analysis_name':'cellranger_multi',
  180. 'output_xcom_key':'loaded_output_files',
  181. 'html_xcom_key':'html_report_file',
  182. 'html_report_file_name':'web_summary.html'})
  183. upload_cellranger_report_to_ftp = \
  184. PythonOperator(
  185. task_id='upload_cellranger_report_to_ftp',
  186. dag=dag,
  187. queue='hpc_4G',
  188. python_callable=ftp_files_upload_for_analysis,
  189. params={'xcom_pull_task':'load_cellranger_result_to_db',
  190. 'xcom_pull_files_key':'html_report_file',
  191. 'collection_name_task':'load_cellranger_result_to_db',
  192. 'collection_name_key':'sample_igf_id',
  193. 'collection_type':'FTP_CELLRANGER_HTML',
  194. 'collection_table':'sample',
  195. 'collect_remote_file':True})
  196. upload_cellranger_report_to_box = \
  197. PythonOperator(
  198. task_id='upload_cellranger_report_to_box',
  199. dag=dag,
  200. queue='hpc_4G',
  201. python_callable=upload_analysis_file_to_box,
  202. params={'xcom_pull_task':'load_cellranger_result_to_db',
  203. 'xcom_pull_files_key':'html_report_file',
  204. 'analysis_tag':'cellranger_multi'})
  205. upload_cellranger_results_to_irods = \
  206. PythonOperator(
  207. task_id='upload_cellranger_results_to_irods',
  208. dag=dag,
  209. queue='hpc_4G',
  210. python_callable=irods_files_upload_for_analysis,
  211. params={'xcom_pull_task':'load_cellranger_result_to_db',
  212. 'xcom_pull_files_key':'loaded_output_files',
  213. 'collection_name_key':'sample_igf_id',
  214. 'collection_name_task':'load_cellranger_result_to_db',
  215. 'analysis_name':'cellranger_multi'})
  216. ## PIPELINE
  217. decide_analysis_branch >> load_cellranger_result_to_db
  218. load_cellranger_result_to_db >> upload_cellranger_report_to_ftp
  219. load_cellranger_result_to_db >> upload_cellranger_report_to_box
  220. load_cellranger_result_to_db >> upload_cellranger_results_to_irods
  221. ## TASK
  222. run_scanpy_for_sc_5p = \
  223. PythonOperator(
  224. task_id='run_scanpy_for_sc_5p',
  225. dag=dag,
  226. queue='hpc_4G',
  227. python_callable=run_singlecell_notebook_wrapper_func,
  228. params={'cellranger_xcom_key':'cellranger_output',
  229. 'cellranger_xcom_pull_task':'run_cellranger',
  230. 'scanpy_timeout':1200,
  231. 'allow_errors':False,
  232. 'kernel_name':'python3',
  233. 'count_dir':'count',
  234. 'analysis_name':'scanpy',
  235. 'output_notebook_key':'scanpy_notebook',
  236. 'output_cellbrowser_key':'cellbrowser_dirs',
  237. 'output_scanpy_h5ad_key':'scanpy_h5ad',
  238. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  239. 'analysis_description_xcom_key':'analysis_description'})
  240. load_cellranger_gex_matrics_to_db = \
  241. PythonOperator(
  242. task_id='load_cellranger_gex_matrics_to_db',
  243. dag=dag,
  244. queue='hpc_4G',
  245. python_callable=load_cellranger_metrices_to_collection,
  246. params={'cellranger_xcom_key':'cellranger_output',
  247. 'cellranger_xcom_pull_task':'run_cellranger',
  248. 'collection_type':'CELLRANGER_MULTI',
  249. 'collection_name_task':'load_cellranger_result_to_db',
  250. 'collection_name_key':'sample_igf_id',
  251. 'metrics_summary_file':'metrics_summary.csv',
  252. 'attribute_prefix':'CELLRANGER_COUNT'})
  253. load_scanpy_report_for_sc_5p_to_db = \
  254. PythonOperator(
  255. task_id='load_scanpy_report_for_sc_5p_to_db',
  256. dag=dag,
  257. queue='hpc_4G',
  258. python_callable=load_analysis_files_func,
  259. params={'collection_name_task':'load_cellranger_result_to_db',
  260. 'collection_name_key':'sample_igf_id',
  261. 'file_name_task':'run_scanpy_for_sc_5p',
  262. 'file_name_key':'scanpy_notebook',
  263. 'analysis_name':'scanpy_5p',
  264. 'collection_type':'SCANPY_HTML',
  265. 'collection_table':'sample',
  266. 'output_files_key':'output_db_files'})
  267. upload_scanpy_report_for_sc_5p_to_ftp = \
  268. PythonOperator(
  269. task_id='upload_scanpy_report_for_sc_5p_to_ftp',
  270. dag=dag,
  271. queue='hpc_4G',
  272. python_callable=ftp_files_upload_for_analysis,
  273. params={'xcom_pull_task':'load_scanpy_report_for_sc_5p_to_db',
  274. 'xcom_pull_files_key':'output_db_files',
  275. 'collection_name_task':'load_cellranger_result_to_db',
  276. 'collection_name_key':'sample_igf_id',
  277. 'collection_type':'FTP_SCANPY_HTML',
  278. 'collection_table':'sample',
  279. 'collect_remote_file':True})
  280. upload_scanpy_report_for_sc_5p_to_box = \
  281. PythonOperator(
  282. task_id='upload_scanpy_report_for_sc_5p_to_box',
  283. dag=dag,
  284. queue='hpc_4G',
  285. python_callable=upload_analysis_file_to_box,
  286. params={'xcom_pull_task':'load_scanpy_report_for_sc_5p_to_db',
  287. 'xcom_pull_files_key':'output_db_files',
  288. 'analysis_tag':'scanpy_single_sample_report'})
  289. upload_cellbrowser_for_sc_5p_to_ftp = \
  290. PythonOperator(
  291. task_id='upload_cellbrowser_for_sc_5p_to_ftp',
  292. dag=dag,
  293. queue='hpc_4G',
  294. python_callable=ftp_files_upload_for_analysis,
  295. params={'xcom_pull_task':'run_scanpy_for_sc_5p',
  296. 'xcom_pull_files_key':'cellbrowser_dirs',
  297. 'collection_name_task':'load_cellranger_result_to_db',
  298. 'collection_name_key':'sample_igf_id',
  299. 'collection_type':'FTP_CELLBROWSER',
  300. 'collection_table':'sample',
  301. 'collect_remote_file':True})
  302. ## PIPELINE
  303. decide_analysis_branch >> run_scanpy_for_sc_5p
  304. run_scanpy_for_sc_5p >> load_scanpy_report_for_sc_5p_to_db
  305. run_scanpy_for_sc_5p >> load_cellranger_gex_matrics_to_db
  306. load_scanpy_report_for_sc_5p_to_db >> upload_scanpy_report_for_sc_5p_to_ftp
  307. load_scanpy_report_for_sc_5p_to_db >> upload_scanpy_report_for_sc_5p_to_box
  308. run_scanpy_for_sc_5p >> upload_cellbrowser_for_sc_5p_to_ftp
  309. ## TASK
  310. run_scirpy_for_vdj_b = \
  311. PythonOperator(
  312. task_id='run_scirpy_for_vdj_b',
  313. dag=dag,
  314. queue='hpc_4G',
  315. python_callable=run_singlecell_notebook_wrapper_func,
  316. params={'cellranger_xcom_key':'cellranger_output',
  317. 'cellranger_xcom_pull_task':'run_cellranger',
  318. 'scanpy_timeout':1200,
  319. 'allow_errors':False,
  320. 'kernel_name':'python3',
  321. 'analysis_name':'scirpy',
  322. 'vdj_dir':'vdj_b',
  323. 'count_dir':'count',
  324. 'output_notebook_key':'scirpy_notebook',
  325. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  326. 'analysis_description_xcom_key':'analysis_description'})
  327. load_cellranger_vdjB_matrics_to_db = \
  328. PythonOperator(
  329. task_id='load_cellranger_vdjB_matrics_to_db',
  330. dag=dag,
  331. queue='hpc_4G',
  332. python_callable=load_cellranger_metrices_to_collection,
  333. params={'cellranger_xcom_key':'cellranger_output',
  334. 'cellranger_xcom_pull_task':'run_cellranger',
  335. 'collection_type':'CELLRANGER_MULTI',
  336. 'collection_name_task':'load_cellranger_result_to_db',
  337. 'collection_name_key':'sample_igf_id',
  338. 'metrics_summary_file':'metrics_summary.csv',
  339. 'attribute_prefix':'CELLRANGER_VDJB'})
  340. load_scirpy_report_for_vdj_b_to_db = \
  341. PythonOperator(
  342. task_id='load_scirpy_report_for_vdj_b_to_db',
  343. dag=dag,
  344. queue='hpc_4G',
  345. python_callable=load_analysis_files_func,
  346. params={'collection_name_task':'load_cellranger_result_to_db',
  347. 'collection_name_key':'sample_igf_id',
  348. 'file_name_task':'run_scirpy_for_vdj_b',
  349. 'file_name_key':'scirpy_notebook',
  350. 'analysis_name':'scirpy_vdj_b',
  351. 'collection_type':'SCIRPY_VDJ_B_HTML',
  352. 'collection_table':'sample',
  353. 'output_files_key':'output_db_files'})
  354. upload_scirpy_report_for_vdj_b_to_ftp = \
  355. PythonOperator(
  356. task_id='upload_scirpy_report_for_vdj_b_to_ftp',
  357. dag=dag,
  358. queue='hpc_4G',
  359. python_callable=ftp_files_upload_for_analysis,
  360. params={'xcom_pull_task':'load_scirpy_report_for_vdj_b_to_db',
  361. 'xcom_pull_files_key':'output_db_files',
  362. 'collection_name_task':'load_cellranger_result_to_db',
  363. 'collection_name_key':'sample_igf_id',
  364. 'collection_type':'FTP_SCIRPY_VDJ_B_HTML',
  365. 'collection_table':'sample',
  366. 'collect_remote_file':True})
  367. upload_scirpy_report_for_vdj_b_to_box = \
  368. PythonOperator(
  369. task_id='upload_scanpy_report_for_vdj_b_to_box',
  370. dag=dag,
  371. queue='hpc_4G',
  372. python_callable=upload_analysis_file_to_box,
  373. params={'xcom_pull_task':'load_scirpy_report_for_vdj_b_to_db',
  374. 'xcom_pull_files_key':'output_db_files',
  375. 'analysis_tag':'scirpy_vdj_b_single_sample_report'})
  376. ## PIPELINE
  377. decide_analysis_branch >> run_scirpy_for_vdj_b
  378. run_scirpy_for_vdj_b >> load_scirpy_report_for_vdj_b_to_db
  379. run_scirpy_for_vdj_b >> load_cellranger_vdjB_matrics_to_db
  380. load_scirpy_report_for_vdj_b_to_db >> upload_scirpy_report_for_vdj_b_to_ftp
  381. load_scirpy_report_for_vdj_b_to_db >> upload_scirpy_report_for_vdj_b_to_box
  382. ## TASK
  383. run_scirpy_for_vdj_t = \
  384. PythonOperator(
  385. task_id='run_scirpy_for_vdj_t',
  386. dag=dag,
  387. queue='hpc_4G',
  388. python_callable=run_singlecell_notebook_wrapper_func,
  389. params={'cellranger_xcom_key':'cellranger_output',
  390. 'cellranger_xcom_pull_task':'run_cellranger',
  391. 'scanpy_timeout':1200,
  392. 'allow_errors':False,
  393. 'kernel_name':'python3',
  394. 'analysis_name':'scirpy',
  395. 'vdj_dir':'vdj_t',
  396. 'count_dir':'count',
  397. 'output_notebook_key':'scirpy_notebook',
  398. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  399. 'analysis_description_xcom_key':'analysis_description'})
  400. load_cellranger_vdjT_matrics_to_db = \
  401. PythonOperator(
  402. task_id='load_cellranger_vdjT_matrics_to_db',
  403. dag=dag,
  404. queue='hpc_4G',
  405. python_callable=load_cellranger_metrices_to_collection,
  406. params={'cellranger_xcom_key':'cellranger_output',
  407. 'cellranger_xcom_pull_task':'run_cellranger',
  408. 'collection_type':'CELLRANGER_MULTI',
  409. 'collection_name_task':'load_cellranger_result_to_db',
  410. 'collection_name_key':'sample_igf_id',
  411. 'metrics_summary_file':'metrics_summary.csv',
  412. 'attribute_prefix':'CELLRANGER_VDJT'})
  413. load_scirpy_report_for_vdj_t_to_db = \
  414. PythonOperator(
  415. task_id='load_scirpy_report_for_vdj_t_to_db',
  416. dag=dag,
  417. queue='hpc_4G',
  418. python_callable=load_analysis_files_func,
  419. params={'collection_name_task':'load_cellranger_result_to_db',
  420. 'collection_name_key':'sample_igf_id',
  421. 'file_name_task':'run_scirpy_for_vdj_t',
  422. 'file_name_key':'scirpy_notebook',
  423. 'analysis_name':'scirpy_vdj_t',
  424. 'collection_type':'SCIRPY_VDJ_T_HTML',
  425. 'collection_table':'sample',
  426. 'output_files_key':'output_db_files'})
  427. upload_scirpy_report_for_vdj_t_to_ftp = \
  428. PythonOperator(
  429. task_id='upload_scirpy_report_for_vdj_t_to_ftp',
  430. dag=dag,
  431. queue='hpc_4G',
  432. python_callable=ftp_files_upload_for_analysis,
  433. params={'xcom_pull_task':'load_scirpy_report_for_vdj_t_to_db',
  434. 'xcom_pull_files_key':'output_db_files',
  435. 'collection_name_task':'load_cellranger_result_to_db',
  436. 'collection_name_key':'sample_igf_id',
  437. 'collection_type':'FTP_SCIRPY_VDJ_T_HTML',
  438. 'collection_table':'sample',
  439. 'collect_remote_file':True})
  440. upload_scirpy_report_for_vdj_t_to_box = \
  441. PythonOperator(
  442. task_id='upload_scirpy_report_for_vdj_t_to_box',
  443. dag=dag,
  444. queue='hpc_4G',
  445. python_callable=upload_analysis_file_to_box,
  446. params={'xcom_pull_task':'load_scirpy_report_for_vdj_t_to_db',
  447. 'xcom_pull_files_key':'output_db_files',
  448. 'analysis_tag':'scirpy_vdj_t_single_sample_report'})
  449. ## PIPELINE
  450. decide_analysis_branch >> run_scirpy_for_vdj_t
  451. run_scirpy_for_vdj_t >> load_scirpy_report_for_vdj_t_to_db
  452. run_scirpy_for_vdj_t >> load_cellranger_vdjT_matrics_to_db
  453. load_scirpy_report_for_vdj_t_to_db >> upload_scirpy_report_for_vdj_t_to_ftp
  454. load_scirpy_report_for_vdj_t_to_db >> upload_scirpy_report_for_vdj_t_to_box
  455. ## TASK
  456. run_seurat_for_sc_5p = \
  457. PythonOperator(
  458. task_id='run_seurat_for_sc_5p',
  459. dag=dag,
  460. queue='hpc_4G',
  461. python_callable=run_singlecell_notebook_wrapper_func,
  462. params={'cellranger_xcom_key':'cellranger_output',
  463. 'cellranger_xcom_pull_task':'run_cellranger',
  464. 'scanpy_timeout':1200,
  465. 'allow_errors':False,
  466. 'kernel_name':'ir',
  467. 'analysis_name':'seurat',
  468. 'vdj_dir':'vdj',
  469. 'count_dir':'count',
  470. 'output_notebook_key':'seurat_notebook',
  471. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  472. 'analysis_description_xcom_key':'analysis_description'})
  473. load_seurat_report_for_sc_5p_db = \
  474. PythonOperator(
  475. task_id='load_seurat_report_for_sc_5p_db',
  476. dag=dag,
  477. queue='hpc_4G',
  478. python_callable=load_analysis_files_func,
  479. params={'collection_name_task':'load_cellranger_result_to_db',
  480. 'collection_name_key':'sample_igf_id',
  481. 'file_name_task':'run_seurat_for_sc_5p',
  482. 'file_name_key':'seurat_notebook',
  483. 'analysis_name':'seurat_5p',
  484. 'collection_type':'SEURAT_HTML',
  485. 'collection_table':'sample',
  486. 'output_files_key':'output_db_files'})
  487. upload_seurat_report_for_sc_5p_ftp = \
  488. PythonOperator(
  489. task_id='upload_seurat_report_for_sc_5p_ftp',
  490. dag=dag,
  491. queue='hpc_4G',
  492. python_callable=ftp_files_upload_for_analysis,
  493. params={'xcom_pull_task':'load_seurat_report_for_sc_5p_db',
  494. 'xcom_pull_files_key':'output_db_files',
  495. 'collection_name_task':'load_cellranger_result_to_db',
  496. 'collection_name_key':'sample_igf_id',
  497. 'collection_type':'FTP_SEURAT_HTML',
  498. 'collection_table':'sample',
  499. 'collect_remote_file':True})
  500. upload_seurat_report_for_sc_5p_to_box = \
  501. PythonOperator(
  502. task_id='upload_seurat_report_for_sc_5p_to_box',
  503. dag=dag,
  504. queue='hpc_4G',
  505. python_callable=upload_analysis_file_to_box,
  506. params={'xcom_pull_task':'load_seurat_report_for_sc_5p_db',
  507. 'xcom_pull_files_key':'output_db_files',
  508. 'analysis_tag':'seurat_single_sample_report'})
  509. ## PIPELINE
  510. decide_analysis_branch >> run_seurat_for_sc_5p
  511. run_seurat_for_sc_5p >> load_seurat_report_for_sc_5p_db
  512. load_seurat_report_for_sc_5p_db >> upload_seurat_report_for_sc_5p_ftp
  513. load_seurat_report_for_sc_5p_db >> upload_seurat_report_for_sc_5p_to_box
  514. ## TASK
  515. convert_cellranger_bam_to_cram = \
  516. PythonOperator(
  517. task_id='convert_cellranger_bam_to_cram',
  518. dag=dag,
  519. queue='hpc_4G4t',
  520. python_callable=convert_bam_to_cram_func,
  521. params={
  522. 'xcom_pull_files_key': 'cellranger_output',
  523. 'xcom_pull_task': 'run_cellranger',
  524. 'analysis_description_xcom_pull_task': 'fetch_analysis_info',
  525. 'analysis_description_xcom_key': 'analysis_description',
  526. 'use_ephemeral_space': True,
  527. 'threads': 4,
  528. 'cellranger_bam_path': 'count/sample_alignments.bam',
  529. 'analysis_name': 'cellranger',
  530. 'collection_type': 'ANALYSIS_CRAM',
  531. 'collection_table': 'sample',
  532. 'cram_files_xcom_key': 'cram_files'})
  533. ## PIPELINE
  534. decide_analysis_branch >> convert_cellranger_bam_to_cram
  535. ## TASK
  536. generate_cell_sorted_bam = \
  537. PythonOperator(
  538. task_id='generate_cell_sorted_bam',
  539. dag=dag,
  540. queue='hpc_16G8t',
  541. python_callable=generate_cell_sorted_bam_func,
  542. params={'xcom_pull_task': 'run_cellranger',
  543. 'xcom_pull_files_key': 'cellranger_output',
  544. 'cellranger_bam_path': 'count/possorted_genome_bam.bam',
  545. 'cellsorted_bam_path': 'count/cellsorted_possorted_genome_bam.bam',
  546. 'samtools_mem': '2G',
  547. 'threads': 7})
  548. run_velocyto = \
  549. PythonOperator(
  550. task_id='run_velocyto',
  551. queue='hpc_16G_long',
  552. python_callable=run_velocyto_func,
  553. params={'xcom_pull_task': 'run_cellranger',
  554. 'xcom_pull_files_key': 'cellranger_output',
  555. 'analysis_description_xcom_pull_task': 'fetch_analysis_info',
  556. 'analysis_description_xcom_key': 'analysis_description' })
  557. run_scvelo_for_sc_5p = \
  558. PythonOperator(
  559. task_id='run_scvelo_for_sc_5p',
  560. dag=dag,
  561. queue='hpc_32G16t',
  562. python_callable=run_scvelo_for_sc_5p_func,
  563. params={'xcom_pull_task': 'run_cellranger',
  564. 'xcom_pull_files_key': 'cellranger_output',
  565. 'analysis_description_xcom_pull_task': 'fetch_analysis_info',
  566. 'analysis_description_xcom_key': 'analysis_description',
  567. 'loom_file_key': 'loom_output',
  568. 'loom_file_task': 'run_velocyto',
  569. 'scanpy_h5ad_task':'run_scanpy_for_sc_5p',
  570. 'scanpy_h5ad_key': 'scanpy_h5ad',
  571. 'timeout': 2400,
  572. 'allow_errors': False,
  573. 'cpu_threads': 14,
  574. 'output_notebook_key': 'scvelo_notebook'})
  575. load_loom_file_to_rds = \
  576. PythonOperator(
  577. task_id='load_loom_file_to_rds',
  578. dag=dag,
  579. queue='hpc_4G',
  580. python_callable=load_analysis_files_func,
  581. params={'collection_name_task':'load_cellranger_result_to_db',
  582. 'collection_name_key':'sample_igf_id',
  583. 'file_name_task':'run_velocyto',
  584. 'file_name_key':'loom_output',
  585. 'analysis_name':'velocyto_5p',
  586. 'collection_type':'VELOCYTO_LOOM',
  587. 'collection_table':'sample',
  588. 'output_files_key':'output_db_files'})
  589. upload_loom_file_to_irods = \
  590. PythonOperator(
  591. task_id='upload_loom_file_to_irods',
  592. dag=dag,
  593. queue='hpc_4G',
  594. python_callable=irods_files_upload_for_analysis,
  595. params={'xcom_pull_task':'load_loom_file_to_rds',
  596. 'xcom_pull_files_key':'output_db_files',
  597. 'collection_name_key':'sample_igf_id',
  598. 'collection_name_task':'load_cellranger_result_to_db',
  599. 'analysis_name':'velocyto_loom'})
  600. load_scvelo_report_to_rds = \
  601. PythonOperator(
  602. task_id='load_scvelo_report_to_rds',
  603. dag=dag,
  604. queue='hpc_4G',
  605. python_callable=load_analysis_files_func,
  606. params={'collection_name_task':'load_cellranger_result_to_db',
  607. 'collection_name_key':'sample_igf_id',
  608. 'file_name_task':'run_scvelo_for_sc_5p',
  609. 'file_name_key':'scvelo_notebook',
  610. 'analysis_name':'scvelo_5p',
  611. 'collection_type':'SCVELO_HTML',
  612. 'collection_table':'sample',
  613. 'output_files_key':'output_db_files'})
  614. upload_scvelo_report_to_ftp = \
  615. PythonOperator(
  616. task_id='upload_scvelo_report_to_ftp',
  617. dag=dag,
  618. queue='hpc_4G',
  619. python_callable=ftp_files_upload_for_analysis,
  620. params={'xcom_pull_task':'load_scvelo_report_to_rds',
  621. 'xcom_pull_files_key':'output_db_files',
  622. 'collection_name_task':'load_cellranger_result_to_db',
  623. 'collection_name_key':'sample_igf_id',
  624. 'collection_type':'FTP_SCVELO_HTML',
  625. 'collection_table':'sample',
  626. 'collect_remote_file':True})
  627. upload_scvelo_report_to_box = \
  628. PythonOperator(
  629. task_id='upload_scvelo_report_to_box',
  630. dag=dag,
  631. queue='hpc_4G',
  632. python_callable=upload_analysis_file_to_box,
  633. params={'xcom_pull_task':'load_scvelo_report_to_rds',
  634. 'xcom_pull_files_key':'output_db_files',
  635. 'analysis_tag':'scvelo_single_sample_report'})
  636. ## PIPELINE
  637. convert_cellranger_bam_to_cram >> generate_cell_sorted_bam
  638. generate_cell_sorted_bam >> run_velocyto
  639. run_velocyto >> run_scvelo_for_sc_5p
  640. run_scanpy_for_sc_5p >> run_scvelo_for_sc_5p
  641. run_velocyto >> load_loom_file_to_rds
  642. load_loom_file_to_rds >> upload_loom_file_to_irods
  643. run_scvelo_for_sc_5p >> load_scvelo_report_to_rds
  644. load_scvelo_report_to_rds >> upload_scvelo_report_to_ftp
  645. load_scvelo_report_to_rds >> upload_scvelo_report_to_box
  646. ## TASK
  647. copy_bam_for_parallel_runs = \
  648. BranchPythonOperator(
  649. task_id='copy_bam_for_parallel_runs',
  650. dag=dag,
  651. queue='hpc_4G',
  652. python_callable=index_and_copy_bam_for_parallel_analysis,
  653. params={'xcom_pull_files_key':'cellranger_output',
  654. 'xcom_pull_task':'run_cellranger',
  655. 'list_of_tasks':[
  656. 'run_picard_alignment_summary',
  657. 'run_picard_qual_summary',
  658. 'run_picard_rna_summary',
  659. 'run_picard_gc_summary',
  660. 'run_picard_base_dist_summary',
  661. 'run_samtools_stats']})
  662. ## TASK
  663. upload_cram_to_irods = \
  664. PythonOperator(
  665. task_id='upload_cram_to_irods',
  666. dag=dag,
  667. queue='hpc_4G',
  668. python_callable=irods_files_upload_for_analysis,
  669. params={'xcom_pull_task':'convert_cellranger_bam_to_cram',
  670. 'xcom_pull_files_key':'cram_files',
  671. 'collection_name_key':'sample_igf_id',
  672. 'collection_name_task':'load_cellranger_result_to_db',
  673. 'analysis_name':'cellranger_multi'})
  674. ## PIPELINE
  675. #convert_cellranger_bam_to_cram >> copy_bam_for_parallel_runs # we need to load metrics to cram
  676. generate_cell_sorted_bam >> copy_bam_for_parallel_runs
  677. convert_cellranger_bam_to_cram >> upload_cram_to_irods
  678. ## TASK
  679. run_picard_alignment_summary = \
  680. PythonOperator(
  681. task_id='run_picard_alignment_summary',
  682. dag=dag,
  683. queue='hpc_4G',
  684. python_callable=run_picard_for_cellranger,
  685. params={'xcom_pull_files_key':'run_picard_alignment_summary',
  686. 'xcom_pull_task':'copy_bam_for_parallel_runs',
  687. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  688. 'analysis_description_xcom_key':'analysis_description',
  689. 'use_ephemeral_space':True,
  690. 'load_metrics_to_cram':True,
  691. 'java_param':'-Xmx4g',
  692. 'picard_command':'CollectAlignmentSummaryMetrics',
  693. 'picard_option':{},
  694. 'analysis_files_xcom_key':'picard_alignment_summary',
  695. 'bam_files_xcom_key':None})
  696. ## PIPELINE
  697. copy_bam_for_parallel_runs >> run_picard_alignment_summary
  698. ## TASK
  699. cleanup_picard_alignment_summary_input = \
  700. PythonOperator(
  701. task_id='cleanup_picard_alignment_summary_input',
  702. dag=dag,
  703. queue='hpc_4G',
  704. python_callable=clean_up_files,
  705. params={'xcom_pull_files_key':'run_picard_alignment_summary',
  706. 'xcom_pull_task':'copy_bam_for_parallel_runs'})
  707. ## PIPELINE
  708. run_picard_alignment_summary >> cleanup_picard_alignment_summary_input
  709. ## TASK
  710. upload_picard_alignment_summary_to_box = \
  711. PythonOperator(
  712. task_id='upload_picard_alignment_summary_to_box',
  713. dag=dag,
  714. queue='hpc_4G',
  715. python_callable=upload_analysis_file_to_box,
  716. params={'xcom_pull_task':'run_picard_alignment_summary',
  717. 'xcom_pull_files_key':'picard_alignment_summary',
  718. 'analysis_tag':'Picard-CollectAlignmentSummaryMetrics'})
  719. ## PIPELINE
  720. run_picard_alignment_summary >> upload_picard_alignment_summary_to_box
  721. ## TASK
  722. run_picard_qual_summary = \
  723. PythonOperator(
  724. task_id='run_picard_qual_summary',
  725. dag=dag,
  726. queue='hpc_4G',
  727. python_callable=run_picard_for_cellranger,
  728. params={'xcom_pull_files_key':'run_picard_qual_summary',
  729. 'xcom_pull_task':'copy_bam_for_parallel_runs',
  730. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  731. 'analysis_description_xcom_key':'analysis_description',
  732. 'use_ephemeral_space':True,
  733. 'load_metrics_to_cram':True,
  734. 'java_param':'-Xmx4g',
  735. 'picard_command':'QualityScoreDistribution',
  736. 'picard_option':{},
  737. 'analysis_files_xcom_key':'picard_qual_summary',
  738. 'bam_files_xcom_key':None})
  739. ## PIPELINE
  740. copy_bam_for_parallel_runs >> run_picard_qual_summary
  741. ## TASK
  742. cleanup_picard_qual_summary_input = \
  743. PythonOperator(
  744. task_id='cleanup_picard_qual_summary_input',
  745. dag=dag,
  746. queue='hpc_4G',
  747. python_callable=clean_up_files,
  748. params={'xcom_pull_files_key':'run_picard_qual_summary',
  749. 'xcom_pull_task':'copy_bam_for_parallel_runs'})
  750. ## PIPELINE
  751. run_picard_qual_summary >> cleanup_picard_qual_summary_input
  752. ## TASK
  753. upload_picard_qual_summary_to_box = \
  754. PythonOperator(
  755. task_id='upload_picard_qual_summary_to_box',
  756. dag=dag,
  757. queue='hpc_4G',
  758. python_callable=upload_analysis_file_to_box,
  759. params={'xcom_pull_task':'run_picard_qual_summary',
  760. 'xcom_pull_files_key':'picard_qual_summary',
  761. 'analysis_tag':'Picard-QualityScoreDistribution'})
  762. ## PIPELINE
  763. run_picard_qual_summary >> upload_picard_qual_summary_to_box
  764. ## TASK
  765. run_picard_rna_summary = \
  766. PythonOperator(
  767. task_id='run_picard_rna_summary',
  768. dag=dag,
  769. queue='hpc_8G',
  770. python_callable=run_picard_for_cellranger,
  771. params={'xcom_pull_files_key':'run_picard_rna_summary',
  772. 'xcom_pull_task':'copy_bam_for_parallel_runs',
  773. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  774. 'analysis_description_xcom_key':'analysis_description',
  775. 'use_ephemeral_space':True,
  776. 'load_metrics_to_cram':True,
  777. 'java_param':'-Xmx7g',
  778. 'picard_command':'CollectRnaSeqMetrics',
  779. 'picard_option':{},
  780. 'analysis_files_xcom_key':'picard_rna_summary',
  781. 'bam_files_xcom_key':None})
  782. ## PIPELINE
  783. copy_bam_for_parallel_runs >> run_picard_rna_summary
  784. ## TASK
  785. cleanup_picard_rna_summary_input = \
  786. PythonOperator(
  787. task_id='cleanup_picard_rna_summary_input',
  788. dag=dag,
  789. queue='hpc_4G',
  790. python_callable=clean_up_files,
  791. params={'xcom_pull_files_key':'run_picard_rna_summary',
  792. 'xcom_pull_task':'copy_bam_for_parallel_runs'})
  793. ## PIPELINE
  794. run_picard_rna_summary >> cleanup_picard_rna_summary_input
  795. ## TASK
  796. upload_picard_rna_summary_to_box = \
  797. PythonOperator(
  798. task_id='upload_picard_rna_summary_to_box',
  799. dag=dag,
  800. queue='hpc_4G',
  801. python_callable=upload_analysis_file_to_box,
  802. params={'xcom_pull_task':'run_picard_rna_summary',
  803. 'xcom_pull_files_key':'picard_rna_summary',
  804. 'analysis_tag':'Picard-CollectRnaSeqMetrics'})
  805. ## PIPELINE
  806. run_picard_rna_summary >> upload_picard_rna_summary_to_box
  807. ## TASK
  808. run_picard_gc_summary = \
  809. PythonOperator(
  810. task_id='run_picard_gc_summary',
  811. dag=dag,
  812. queue='hpc_4G',
  813. python_callable=run_picard_for_cellranger,
  814. params={'xcom_pull_files_key':'run_picard_gc_summary',
  815. 'xcom_pull_task':'copy_bam_for_parallel_runs',
  816. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  817. 'analysis_description_xcom_key':'analysis_description',
  818. 'use_ephemeral_space':True,
  819. 'load_metrics_to_cram':True,
  820. 'java_param':'-Xmx4g',
  821. 'picard_command':'CollectGcBiasMetrics',
  822. 'picard_option':{},
  823. 'analysis_files_xcom_key':'picard_gc_summary',
  824. 'bam_files_xcom_key':None})
  825. ## PIPELINE
  826. copy_bam_for_parallel_runs >> run_picard_gc_summary
  827. ## TASK
  828. cleanup_picard_gc_summary_input = \
  829. PythonOperator(
  830. task_id='cleanup_picard_gc_summary_input',
  831. dag=dag,
  832. queue='hpc_4G',
  833. python_callable=clean_up_files,
  834. params={'xcom_pull_files_key':'run_picard_gc_summary',
  835. 'xcom_pull_task':'copy_bam_for_parallel_runs'})
  836. ## PIPELINE
  837. run_picard_gc_summary >> cleanup_picard_gc_summary_input
  838. ## TASK
  839. upload_picard_gc_summary_to_box = \
  840. PythonOperator(
  841. task_id='upload_picard_gc_summary_to_box',
  842. dag=dag,
  843. queue='hpc_4G',
  844. python_callable=upload_analysis_file_to_box,
  845. params={'xcom_pull_task':'run_picard_gc_summary',
  846. 'xcom_pull_files_key':'picard_gc_summary',
  847. 'analysis_tag':'Picard-CollectGcBiasMetrics'})
  848. ## PIPELINE
  849. run_picard_gc_summary >> upload_picard_gc_summary_to_box
  850. ## TASK
  851. run_picard_base_dist_summary = \
  852. PythonOperator(
  853. task_id='run_picard_base_dist_summary',
  854. dag=dag,
  855. queue='hpc_4G',
  856. python_callable=run_picard_for_cellranger,
  857. params={'xcom_pull_files_key':'run_picard_base_dist_summary',
  858. 'xcom_pull_task':'copy_bam_for_parallel_runs',
  859. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  860. 'analysis_description_xcom_key':'analysis_description',
  861. 'use_ephemeral_space':True,
  862. 'load_metrics_to_cram':True,
  863. 'java_param':'-Xmx4g',
  864. 'picard_command':'CollectBaseDistributionByCycle',
  865. 'picard_option':{},
  866. 'analysis_files_xcom_key':'picard_base_summary',
  867. 'bam_files_xcom_key':None})
  868. ## PIPELINE
  869. copy_bam_for_parallel_runs >> run_picard_base_dist_summary
  870. ## TASK
  871. cleanup_picard_base_dist_summary_input = \
  872. PythonOperator(
  873. task_id='cleanup_picard_base_dist_summary_input',
  874. dag=dag,
  875. queue='hpc_4G',
  876. python_callable=clean_up_files,
  877. params={'xcom_pull_files_key':'run_picard_base_dist_summary',
  878. 'xcom_pull_task':'copy_bam_for_parallel_runs'})
  879. ## PIPELINE
  880. run_picard_base_dist_summary >> cleanup_picard_base_dist_summary_input
  881. ## TASK
  882. upload_picard_base_dist_summary_to_box = \
  883. PythonOperator(
  884. task_id='upload_picard_base_dist_summary_to_box',
  885. dag=dag,
  886. queue='hpc_4G',
  887. python_callable=upload_analysis_file_to_box,
  888. params={'xcom_pull_task':'run_picard_base_dist_summary',
  889. 'xcom_pull_files_key':'picard_base_summary',
  890. 'analysis_tag':'Picard-CollectBaseDistributionByCycle'})
  891. ## PIPELINE
  892. run_picard_base_dist_summary >> upload_picard_base_dist_summary_to_box
  893. ## TASK
  894. run_samtools_stats = \
  895. PythonOperator(
  896. task_id='run_samtools_stats',
  897. dag=dag,
  898. queue='hpc_4G4t',
  899. python_callable=run_samtools_for_cellranger,
  900. params={'xcom_pull_files_key':'run_samtools_stats',
  901. 'xcom_pull_task':'copy_bam_for_parallel_runs',
  902. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  903. 'analysis_description_xcom_key':'analysis_description',
  904. 'use_ephemeral_space':True,
  905. 'load_metrics_to_cram':True,
  906. 'samtools_command':'stats',
  907. 'threads':4,
  908. 'analysis_files_xcom_key':'samtools_stats'})
  909. ## PIPELINE
  910. copy_bam_for_parallel_runs >> run_samtools_stats
  911. ## TASK
  912. upload_samtools_stats_to_box = \
  913. PythonOperator(
  914. task_id='upload_samtools_stats_to_box',
  915. dag=dag,
  916. queue='hpc_4G',
  917. python_callable=upload_analysis_file_to_box,
  918. params={'xcom_pull_task':'run_samtools_stats',
  919. 'xcom_pull_files_key':'samtools_stats',
  920. 'analysis_tag':'Samtools-stats'})
  921. ## PIPELINE
  922. run_samtools_stats >> upload_samtools_stats_to_box
  923. ## TASK
  924. run_samtools_idxstats = \
  925. PythonOperator(
  926. task_id='run_samtools_idxstats',
  927. dag=dag,
  928. queue='hpc_4G4t',
  929. python_callable=run_samtools_for_cellranger,
  930. params={'xcom_pull_files_key':'run_samtools_stats',
  931. 'xcom_pull_task':'copy_bam_for_parallel_runs',
  932. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  933. 'analysis_description_xcom_key':'analysis_description',
  934. 'use_ephemeral_space':True,
  935. 'load_metrics_to_cram':True,
  936. 'samtools_command':'idxstats',
  937. 'threads':4,
  938. 'analysis_files_xcom_key':'samtools_idxstats'})
  939. ## PIPELINE
  940. run_samtools_stats >> run_samtools_idxstats
  941. ## TASK
  942. cleanup_samtools_stats_input = \
  943. PythonOperator(
  944. task_id='cleanup_samtools_stats_input',
  945. dag=dag,
  946. queue='hpc_4G',
  947. python_callable=clean_up_files,
  948. params={'xcom_pull_files_key':'run_samtools_stats',
  949. 'xcom_pull_task':'copy_bam_for_parallel_runs'})
  950. ## PIPELINE
  951. run_samtools_idxstats >> cleanup_samtools_stats_input
  952. ## TASK
  953. upload_samtools_idxstats_to_box = \
  954. PythonOperator(
  955. task_id='upload_samtools_idxstats_to_box',
  956. dag=dag,
  957. queue='hpc_4G',
  958. python_callable=upload_analysis_file_to_box,
  959. params={'xcom_pull_task':'run_samtools_idxstats',
  960. 'xcom_pull_files_key':'samtools_idxstats',
  961. 'analysis_tag':'Samtools-idxstats'})
  962. ## PIPELINE
  963. run_samtools_idxstats >> upload_samtools_idxstats_to_box
  964. ## TASK
  965. run_multiqc = \
  966. PythonOperator(
  967. task_id='run_multiqc',
  968. dag=dag,
  969. queue='hpc_4G',
  970. trigger_rule='none_failed_or_skipped',
  971. python_callable=run_multiqc_for_cellranger,
  972. params={
  973. 'list_of_analysis_xcoms_and_tasks':{
  974. 'run_cellranger':'cellranger_output',
  975. 'run_picard_alignment_summary':'picard_alignment_summary',
  976. 'run_picard_qual_summary':'picard_qual_summary',
  977. 'run_picard_rna_summary':'picard_rna_summary',
  978. 'run_picard_gc_summary':'picard_gc_summary',
  979. 'run_picard_base_dist_summary':'picard_base_summary',
  980. 'run_samtools_stats':'samtools_stats',
  981. 'run_samtools_idxstats':'samtools_idxstats'},
  982. 'analysis_description_xcom_pull_task':'fetch_analysis_info',
  983. 'analysis_description_xcom_key':'analysis_description',
  984. 'use_ephemeral_space':True,
  985. 'multiqc_html_file_xcom_key':'multiqc_html',
  986. 'multiqc_data_file_xcom_key':'multiqc_data',
  987. 'tool_order_list':['picad','samtools']})
  988. ## PIPELINE
  989. run_picard_alignment_summary >> run_multiqc
  990. run_picard_qual_summary >> run_multiqc
  991. run_picard_rna_summary >> run_multiqc
  992. run_picard_gc_summary >> run_multiqc
  993. run_picard_base_dist_summary >> run_multiqc
  994. run_samtools_idxstats >> run_multiqc
  995. ## TASK
  996. load_multiqc_html = \
  997. PythonOperator(
  998. task_id='load_multiqc_html',
  999. dag=dag,
  1000. queue='hpc_4G',
  1001. python_callable=load_analysis_files_func,
  1002. params={'collection_name_task':'load_cellranger_result_to_db',
  1003. 'collection_name_key':'sample_igf_id',
  1004. 'file_name_task':'run_multiqc',
  1005. 'file_name_key':'multiqc_html',
  1006. 'analysis_name':'multiqc',
  1007. 'collection_type':'MULTIQC_HTML',
  1008. 'collection_table':'sample',
  1009. 'output_files_key':'output_db_files'})
  1010. ## PIPELINE
  1011. run_multiqc >> load_multiqc_html
  1012. ## TASK
  1013. upload_multiqc_to_ftp = \
  1014. PythonOperator(
  1015. task_id='upload_multiqc_to_ftp',
  1016. dag=dag,
  1017. queue='hpc_4G',
  1018. python_callable=ftp_files_upload_for_analysis,
  1019. params={'xcom_pull_task':'load_multiqc_html',
  1020. 'xcom_pull_files_key':'output_db_files',
  1021. 'collection_name_task':'load_cellranger_result_to_db',
  1022. 'collection_name_key':'sample_igf_id',
  1023. 'collection_type':'FTP_MULTIQC_HTML',
  1024. 'collection_table':'sample',
  1025. 'collect_remote_file':True})
  1026. ## PIPELINE
  1027. load_multiqc_html >> upload_multiqc_to_ftp
  1028. ## TASK
  1029. upload_multiqc_to_box = \
  1030. PythonOperator(
  1031. task_id='upload_multiqc_to_box',
  1032. dag=dag,
  1033. queue='hpc_4G',
  1034. python_callable=upload_analysis_file_to_box,
  1035. params={'xcom_pull_task':'load_multiqc_html',
  1036. 'xcom_pull_files_key':'output_db_files',
  1037. 'analysis_tag':'multiqc_report'})
  1038. ## PIPELINE
  1039. load_multiqc_html >> upload_multiqc_to_box
  1040. ## TASK
  1041. update_analysis_and_status = \
  1042. PythonOperator(
  1043. task_id='update_analysis_and_status',
  1044. dag=dag,
  1045. queue='hpc_4G',
  1046. python_callable=change_pipeline_status,
  1047. trigger_rule='none_failed_or_skipped',
  1048. params={'new_status':'FINISHED',
  1049. 'no_change_status':'SEEDED'})
  1050. ## PIPELINE
  1051. upload_multiqc_to_ftp >> update_analysis_and_status
  1052. upload_scanpy_report_for_sc_5p_to_ftp >> update_analysis_and_status
  1053. upload_scanpy_report_for_sc_5p_to_box >> update_analysis_and_status
  1054. upload_cellbrowser_for_sc_5p_to_ftp >> update_analysis_and_status
  1055. #upload_scirpy_report_for_vdj_to_ftp >> update_analysis_and_status # no more ambiguous VDJ
  1056. #upload_scirpy_report_for_vdj_to_box >> update_analysis_and_status
  1057. upload_scirpy_report_for_vdj_b_to_ftp >> update_analysis_and_status
  1058. upload_scirpy_report_for_vdj_b_to_box >> update_analysis_and_status
  1059. upload_scirpy_report_for_vdj_t_to_ftp >> update_analysis_and_status
  1060. upload_scirpy_report_for_vdj_t_to_box >> update_analysis_and_status
  1061. upload_seurat_report_for_sc_5p_ftp >> update_analysis_and_status
  1062. upload_seurat_report_for_sc_5p_to_box >> update_analysis_and_status
  1063. upload_cellranger_results_to_irods >> update_analysis_and_status
  1064. upload_cellranger_report_to_ftp >> update_analysis_and_status
  1065. upload_cellranger_report_to_box >> update_analysis_and_status
  1066. upload_cram_to_irods >> update_analysis_and_status
  1067. upload_scvelo_report_to_box >> update_analysis_and_status
  1068. upload_scvelo_report_to_ftp >> update_analysis_and_status
  1069. ## TASK
  1070. update_qc_pages = \
  1071. PythonOperator(
  1072. task_id='update_qc_pages',
  1073. dag=dag,
  1074. queue='hpc_4G',
  1075. python_callable=create_and_update_qc_pages,
  1076. params={'use_ephemeral_space':True,
  1077. 'collection_type_list':[
  1078. 'FTP_MULTIQC_HTML',
  1079. 'FTP_SEURAT_HTML',
  1080. 'FTP_SCIRPY_VDJ_T_HTML',
  1081. 'FTP_SCIRPY_VDJ_B_HTML',
  1082. 'FTP_SCIRPY_VDJ_HTML',
  1083. 'FTP_CELLBROWSER',
  1084. 'FTP_SCANPY_HTML',
  1085. 'FTP_SCVELO_HTML',
  1086. 'FTP_CELLRANGER_HTML']})
  1087. ## PIPELINE
  1088. update_analysis_and_status >> update_qc_pages