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

extern crate scoped_threadpool;
extern crate getopts;
extern crate bio;
extern crate time;

use scoped_threadpool::Pool;
use getopts::Options;
use std::process::Command;
use std::path::Path;
use bio::io::fasta;
use time::now;

type ThrdNm = u32;

pub fn rn_riblast_in_prll(args: &Vec<String>) {
  let prgrm = args[0].clone();
  let mut opts = Options::new();
  opts.reqopt("i", "", "RNA seqs in single file of FASTA format", "STR");
  opts.reqopt("o", "", "Path to output dir.", "STR");
  opts.reqopt("d", "", "Path to database", "STR");
  opts.optopt("l", "", "Max size of seed length (Default: 20)", "UINT");
  opts.optopt("e", "", "Interaction energy threshold for seed search (Default: -6.5)", "FLOAT");
  opts.optopt("f", "", "Hybridization energy threshold for removal of interaction candidates before gapped extension (Default: -3.05)", "FLOAT");
  opts.optopt("x", "", "Dropout length in gapped extension (Default: 18)", "UINT");
  opts.optopt("y", "", "Dropout length in gapless extension (Default: 5)", "UINT");
  opts.optopt("t", "", "# of threads in multithreading (Default: system val.)", "UINT");
  opts.optopt("b", "", "Path to RIblast binary (Default: system val.)", "STR");
  opts.optflag("h", "help", "Print help menu");
  let mtchs = match opts.parse(&args[1 ..]) {
    Ok(mtch) => {mtch}
    Err(fl) => {prnt_usg(&prgrm, &opts); panic!(fl.to_string())}
  };
  if mtchs.opt_present("h") {
    prnt_usg(&prgrm, &opts);
    return;
  }
  let mut thrd_nm = 0;
  if mtchs.opt_present("t") {
    thrd_nm = mtchs.opt_str("t").expect("Retrieving num. of threads from command args failed. ").parse().expect("Parsing num. of threads failed. ");
  } else {
    gt_sys_thrd_nm(&mut thrd_nm);
  };
  let riblast_pth = if mtchs.opt_present("b") {
    mtchs.opt_str("b").expect("Retrieving path to RIblast from command args failed. ")
  } else {
    String::from("RIblast")
  };
  let tm_stmp = now();
  let tm_stmp = (tm_stmp.tm_year + 1900).to_string() + "-" + &(tm_stmp.tm_mon + 1).to_string() + "-" + &tm_stmp.tm_mday.to_string() + "-" + &tm_stmp.tm_hour.to_string() + ":" + &tm_stmp.tm_min.to_string() + ":" + &tm_stmp.tm_sec.to_string();
  let tmp = String::from("/tmp/prll_ri_") + &tm_stmp;
  let tmp = Path::new(&tmp);
  if !tmp.exists() {
    Command::new("mkdir").arg(tmp.to_str().expect("Parsing temporary dir. path failed. ")).output().expect("Creating temporary dir. failed. ");
  }
  let inpt = mtchs.opt_str("i").expect("Retrieving input file path from command args failed. ");
  let otpt_dr = mtchs.opt_str("o").expect("Retrieving output dir. path from command args failed. ");
  let otpt_dr = Path::new(&otpt_dr);
  if !otpt_dr.exists() {
    Command::new("mkdir").arg(otpt_dr.to_str().expect("Parsing output dir. path failed. ")).output().expect("Creating output dir. failed. ");
  }
  let mut args = Vec::new();
  let opts = vec!["d", "l", "e", "f", "x", "y"];
  let opt_dflt_vls = vec!["", "20", "-6.5", "-3.05", "18", "5"];
  for itr_pr in opts.iter().zip(opt_dflt_vls.iter()) {
    let (opt, opt_dflt_vl) = itr_pr;
    let mut arg = String::from("-");
    arg.push_str(opt);
    args.push(arg);
    if mtchs.opt_present(opt) {
      args.push(String::from(mtchs.opt_str(opt).expect("Retrieving option failed. ")));
    } else {
      args.push(String::from(*opt_dflt_vl));
    }
  }
  let mut thrd_pl = Pool::new(thrd_nm);
  let _grd = thrd_pl.scoped(|scoped| {
    let fasta_rdr = fasta::Reader::from_file(Path::new(&inpt)).ok().expect("Reading FASTA file failed. ");
    for (_i, rc) in fasta_rdr.records().enumerate() {
      scoped.execute(|| {
        let dt = rc.ok().expect("Reading FASTA record failed. ");
        let dt_id = dt.id().expect("Record ID is missing. ");
        let tmp_otpt = tmp.join(String::from(dt_id) + ".fa");
        let otpt = otpt_dr.join(dt_id);
        let mut fasta_wrtr = fasta::Writer::to_file(Path::new(&tmp_otpt)).ok().expect("Writing into FASTA file is blocked. ");
        let _rslt = fasta_wrtr.write_record(&dt);
        let _rslt = fasta_wrtr.flush();
        let mut riblast_args = args.clone();
        riblast_args.insert(0, String::from("ris"));
        riblast_args.push(String::from("-i"));
        riblast_args.push(String::from(tmp_otpt.to_str().expect("Parsing temporary file path failed. ")));
        riblast_args.push(String::from("-o"));
        riblast_args.push(String::from(otpt.to_str().expect("Parsing output dir. path failed. ")));
        let mut riblast_arg_slcs = Vec::new();
        for riblast_arg in &riblast_args {
          riblast_arg_slcs.push(riblast_arg.as_str());
        }
        let otpt = Command::new(&riblast_pth).args(&riblast_arg_slcs).output().expect("RIblast failed. ");
        println!("Status: {}", otpt.status);
        println!("stdout: {}", String::from_utf8_lossy(&otpt.stdout));
        println!("stderr: {}", String::from_utf8_lossy(&otpt.stderr));
        assert!(otpt.status.success());
      });
    }
  });
  Command::new("rm").args(&["-rf", tmp.to_str().expect("Parsing temporary dir. path failed. ")]).output().expect("Removing temporary dir. failed. ");
}

pub fn prnt_usg(prgrm: &str, opts: &Options) {
  let brf = format!("Usage: {} [options]", prgrm);
  print!("{}", opts.usage(&brf));
}

fn gt_sys_thrd_nm(thrd_nm: &mut ThrdNm) {
  let otpt = Command::new("nproc").output().expect("Counting system threads failed. ");
  *thrd_nm = String::from_utf8_lossy(&otpt.stdout).trim().parse().expect("Parsing num. of threads failed. ");
}