tb-to-fhir-full/main.nf at main · oucru-id/tb-to-fhir-full · GitHub

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
#!/usr/bin/env nextflow

nextflow.enable.dsl = 2

include { ILLUMINA }              from './workflows/illumina.nf'
include { NANOPORE }              from './workflows/nanopore.nf'
include { VCF_PROCESSING }        from './workflows/vcf.nf'
include { LINEAGE }               from './workflows/lineage.nf'
include { GENERATE_REPORT }       from './workflows/report.nf'
include { GENERATE_SAMPLE_REPORTS } from './workflows/report.nf'
include { FHIR }                  from './workflows/fhir.nf'
include { VALIDATE }              from './workflows/validate_fhir.nf'
include { MERGE_CLINICAL_DATA }   from './workflows/merge_clinical_data.nf'
include { UPLOAD_FHIR }           from './workflows/upload_fhir.nf'
include { VERSIONS }              from './workflows/utils.nf'

workflow {

log.info """
    Mycobacterium tuberculosis Mutation Analysis Pipeline (v${params.version})
    Developed by SPHERES OUCRU-ID
    Documentation: https://tb-pipeline-docs.readthedocs.io/
"""

    illumina_reads_ch = Channel
        .fromFilePairs("${params.reads_dir}/*_{1,2}_illumina.fastq.gz")
        .map { id, files -> tuple(id, files) }

    nanopore_reads_ch = Channel
        .fromPath("${params.reads_dir}/*_ont.fastq.gz", checkIfExists: false)
        .map { file -> tuple(file.baseName.replaceFirst(/_ont$/, ''), file) }

    vcf_ch = Channel
        .fromPath("${params.vcf_dir}/*.vcf{,.gz}", checkIfExists: false)
        .map { file -> tuple(file.baseName.replaceFirst(/\.vcf(\.gz)?$/, ''), file) }

    illumina_out = ILLUMINA(illumina_reads_ch)
    nanopore_out = NANOPORE(nanopore_reads_ch)
    vcf_out = VCF_PROCESSING(vcf_ch)

    all_filtered = illumina_out.filtered
        .mix(nanopore_out.filtered)
        .mix(vcf_out.filtered)

    lineage_out = LINEAGE(all_filtered)

    all_qc = illumina_out.qc_report
        .mix(nanopore_out.qc_report)
        .map { it -> it[1] }
        .flatten()
        .filter { it.toString().endsWith('_fastqc.zip') }
        .collect()

    all_annotated = illumina_out.annotated
        .mix(nanopore_out.annotated)
        .mix(vcf_out.annotated)
        .map { it -> it instanceof List ? it[1] : it }

    GENERATE_REPORT(all_qc)

    lineage_files = lineage_out.lineage_results
        .map { sample_id, file_path -> file_path }
        .collect()

    sample_reports = GENERATE_SAMPLE_REPORTS(all_annotated, lineage_files)

    patient_metadata_ch = Channel.fromPath(params.patient_metadata, checkIfExists: false)
        .first()

    org_metadata_ch = Channel.fromPath(params.organization_metadata, checkIfExists: false)
        .first()

    practitioner_metadata_ch = Channel.fromPath(params.practitioner_metadata, checkIfExists: false)
        .first()

    fhir_out = FHIR(all_annotated, lineage_files, org_metadata_ch)

    merged_clinical_out = MERGE_CLINICAL_DATA(
        fhir_out.fhir_output,
        patient_metadata_ch,
        org_metadata_ch,
        practitioner_metadata_ch
    )

    validation_out = VALIDATE(merged_clinical_out.merged_fhir)

    upload_out = UPLOAD_FHIR(validation_out.validated_fhir)

    VERSIONS()
}