twitcher 0.1.9

Find template switch mutations in genomic data
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

use std::{
    fs::File,
    io::{BufWriter, Write},
};

use tokio::sync::{
    mpsc::{self, error::SendError},
    oneshot,
};
use tracing::{debug, error};

use crate::common::{
    aligner::InProgress,
    csv::{CSVAuxData, TwitcherCSVWriter},
};

#[derive(clap::Args, Debug)]
pub struct CliOutputArgs {
    /// Output CSV file; If not specified or "-", csv will be printed to stdout.
    #[arg(short, long, value_name = "FILE")]
    output: Option<String>,

    /// If enabled, write all clusters to the output. Otherwise (the default) only write clusters that are realigned with a template switch.
    #[arg(short, long)]
    all_clusters: bool,
}

pub struct ClusterResult {
    pub recv: InProgress,
    pub aux: CSVAuxData,
}

pub struct Writer {
    sender: mpsc::Sender<ClusterResult>,
    done_rx: oneshot::Receiver<()>,
}

impl Writer {
    pub fn spawn(args: &CliOutputArgs) -> anyhow::Result<Writer> {
        let (tx, mut rx) = mpsc::channel(1024);
        let (done_tx, done_rx) = oneshot::channel();
        let mut writer = InnerWriter::try_from(args)?;
        tokio::spawn(async move {
            while let Some(value) = rx.recv().await {
                if let Err(e) = writer.write(value).await {
                    error!("{e}");
                }
            }
            let _ = writer.inner.flush();
            // TODO this 'done' hack is not nice
            let _ = done_tx.send(());
        });
        Ok(Writer {
            sender: tx,
            done_rx,
        })
    }

    pub async fn write(&self, result: ClusterResult) -> Result<(), SendError<ClusterResult>> {
        self.sender.send(result).await
    }

    pub async fn wait_until_done(self) {
        std::mem::drop(self.sender);
        let _ = self.done_rx.await;
    }
}

struct InnerWriter {
    all_clusters: bool,
    inner: TwitcherCSVWriter,
}

impl InnerWriter {
    async fn write(&mut self, value: ClusterResult) -> anyhow::Result<()> {
        let ClusterResult { mut recv, aux } = value;
        let result = recv.recv().await?;
        let result = match result.as_ref() {
            Ok(result) => result,
            Err(sf @ crate::common::aligner::result::AlignmentFailure::SoftFailure { .. }) => {
                debug!("{sf}");
                return Ok(());
            }
            Err(crate::common::aligner::result::AlignmentFailure::Error { error }) => {
                return Err(anyhow::anyhow!(error.clone()));
            }
        };

        if !result.has_ts() && !self.all_clusters {
            return Ok(());
        }

        self.inner.write(result, aux)?;

        Ok(())
    }
}

impl TryFrom<&CliOutputArgs> for InnerWriter {
    type Error = anyhow::Error;

    fn try_from(args: &CliOutputArgs) -> Result<Self, Self::Error> {
        let write = match args.output.as_deref() {
            None | Some("-") => Box::new(std::io::stdout()) as Box<dyn Write + Send>,
            Some(path) => Box::new(BufWriter::new(File::create(path)?)) as Box<dyn Write + Send>,
        };
        let writer = TwitcherCSVWriter::new(write);
        Ok(InnerWriter {
            inner: writer,
            all_clusters: args.all_clusters,
        })
    }
}