ngs 0.4.0

Command line tool for processing next-generation sequencing data.
Documentation 
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
128
129
130
131
132
133
134
135
136
137
138

//! Conversions from a CRAM file to other next-generation sequencing file formats.

use std::path::PathBuf;

use anyhow::Context;
use futures::TryStreamExt;
use noodles::bam;
use noodles::bgzf;
use noodles::bgzf::writer::CompressionLevel;
use noodles::fasta;
use noodles::sam;
use tokio::fs::File;

use crate::utils::args::CompressionStrategy;
use crate::utils::args::NumberOfRecords;
use crate::utils::display::RecordCounter;
use crate::utils::formats;
use crate::utils::formats::cram::ParsedAsyncCRAMFile;
use crate::utils::formats::utils::IndexCheck;

/// Converts a CRAM file to a SAM file in an asyncronous fashion.
pub async fn to_sam_async(
    from: PathBuf,
    to: PathBuf,
    fasta: PathBuf,
    max_records: NumberOfRecords,
) -> anyhow::Result<()> {
    // (1) Open and parse the CRAM file.
    let ParsedAsyncCRAMFile {
        mut reader, header, ..
    } = formats::cram::open_and_parse_async(from, IndexCheck::None)
        .await
        .with_context(|| {
            "opening and parsing CRAM file. Check that your reference FASTA matches \
            the FASTA used to generate the file."
        })?;

    // (2) Builds the FASTA repository and associated index.
    let repository = fasta::indexed_reader::Builder::default()
        .build_from_path(fasta)
        .map(fasta::repository::adapters::IndexedReader::new)
        .map(fasta::Repository::new)
        .with_context(|| "FASTA repository and associated index")?;

    // (3) Open the SAM file writer.
    let handle = File::create(to).await?;
    let mut writer = sam::AsyncWriter::new(handle);

    // (4) Write the header.
    writer.write_header(&header.parsed).await?;

    // (5) Write each record in the CRAM file to the SAM file.
    let mut counter = RecordCounter::new();
    let mut records = reader.records(&repository, &header.parsed);

    while let Some(record) = records.try_next().await? {
        let record = record.try_into_alignment_record(&header.parsed)?;
        writer
            .write_alignment_record(&header.parsed, &record)
            .await?;

        counter.inc();

        if counter.time_to_break(&max_records) {
            break;
        }
    }

    Ok(())
}

/// Converts a CRAM file to a BAM file in an asyncronous fashion.
pub async fn to_bam_async(
    from: PathBuf,
    to: PathBuf,
    fasta: PathBuf,
    max_records: NumberOfRecords,
    compression_strategy: CompressionStrategy,
) -> anyhow::Result<()> {
    // (1) Open and parse the CRAM file.
    let ParsedAsyncCRAMFile {
        mut reader, header, ..
    } = formats::cram::open_and_parse_async(from, IndexCheck::None)
        .await
        .with_context(|| {
            "opening and parsing CRAM file. Check that your reference FASTA matches \
            the FASTA used to generate the file."
        })?;

    // (2) Builds the FASTA repository and associated index.
    let repository = fasta::indexed_reader::Builder::default()
        .build_from_path(fasta)
        .map(fasta::repository::adapters::IndexedReader::new)
        .map(fasta::Repository::new)
        .with_context(|| "FASTA repository and associated index")?;

    // (3) Determine the compression level.
    let compression_level: CompressionLevel = compression_strategy.into();

    // (4) Open the BAM file writer.
    let mut writer = File::create(to)
        .await
        .map(|f| {
            bgzf::r#async::writer::Builder::default()
                .set_compression_level(compression_level)
                .build_with_writer(f)
        })
        .map(bam::AsyncWriter::from)
        .with_context(|| "opening output filestream")?;

    // (5) Write the header and the reference sequences.
    writer.write_header(&header.parsed).await?;
    writer
        .write_reference_sequences(header.parsed.reference_sequences())
        .await?;

    // (6) Write each record in the CRAM file to the BAM file.
    let mut counter = RecordCounter::new();
    let mut records = reader.records(&repository, &header.parsed);

    while let Some(record) = records.try_next().await? {
        let record = record.try_into_alignment_record(&header.parsed)?;
        writer
            .write_alignment_record(&header.parsed, &record)
            .await?;

        counter.inc();

        if counter.time_to_break(&max_records) {
            break;
        }
    }

    // (7) Shutdown the async writer.
    writer.shutdown().await?;

    Ok(())
}