use anyhow::Result;
use rayon::prelude::*;
use rustc_hash::FxHashMap;
use std::fs::File;
use std::io::{BufWriter, Write};
use std::path::Path;
use std::sync::Mutex;
use crate::seqio::{FastaRecord, FastqFile};
#[derive(Clone)]
pub struct ExtenderConfig {
pub kmer_size: usize,
pub num_edge_kmers: usize,
pub min_coverage: usize,
pub branching_threshold: f64,
pub max_n_ratio: f64,
pub extension_step: usize,
pub max_consecutive_failures: usize,
pub max_extension_per_side: usize,
}
impl Default for ExtenderConfig {
fn default() -> Self {
Self {
kmer_size: 21,
num_edge_kmers: 5,
min_coverage: 2,
branching_threshold: 0.2,
max_n_ratio: 0.05,
extension_step: 200,
max_consecutive_failures: 2,
max_extension_per_side: 2000,
}
}
}
#[derive(Debug, Clone)]
pub struct ExtendedContig {
pub name: String,
pub extended_seq: String,
}
pub struct ContigExtender {
config: ExtenderConfig,
reads: std::sync::Arc<Vec<String>>,
}
impl ContigExtender {
pub fn new(config: ExtenderConfig) -> Self {
Self {
config,
reads: std::sync::Arc::new(Vec::new()),
}
}
pub fn load_reads_shared(r1_path: &Path, r2_path: &Path) -> Result<std::sync::Arc<Vec<String>>> {
eprintln!("Loading reads into memory...");
let r1_owned = r1_path.to_path_buf();
let r2_owned = r2_path.to_path_buf();
let load = |path: std::path::PathBuf| -> Result<Vec<String>> {
let mut reads = Vec::new();
let mut reader = FastqFile::open(&path)?;
while let Some(record) = reader.read_next()? {
reads.push(record.seq);
}
Ok(reads)
};
let handle_r1 = std::thread::spawn(move || load(r1_owned));
let handle_r2 = std::thread::spawn(move || load(r2_owned));
let mut reads = handle_r1.join().map_err(|_| anyhow::anyhow!("R1 load thread panicked"))??;
let reads_r2 = handle_r2.join().map_err(|_| anyhow::anyhow!("R2 load thread panicked"))??;
reads.extend(reads_r2);
eprintln!("Loaded {} reads into memory", reads.len());
Ok(std::sync::Arc::new(reads))
}
pub fn load_reads(&mut self, r1_path: &Path, r2_path: &Path) -> Result<()> {
self.reads = Self::load_reads_shared(r1_path, r2_path)?;
Ok(())
}
pub fn set_reads(&mut self, reads: std::sync::Arc<Vec<String>>) {
self.reads = reads;
}
pub fn extend_contigs(&self, contigs: &[FastaRecord]) -> Result<Vec<ExtendedContig>> {
let k = self.config.kmer_size;
let max_failures = self.config.max_consecutive_failures;
let states: Vec<Mutex<ContigState>> = contigs.iter().map(|c| {
Mutex::new(ContigState {
name: c.name.clone(),
current_seq: c.seq.clone(),
left_failures: 0,
right_failures: 0,
left_grown: 0,
right_grown: 0,
})
}).collect();
let t_start = std::time::Instant::now();
let mut rounds = 0usize;
loop {
rounds += 1;
let active_indices: Vec<usize> = states.iter().enumerate()
.filter(|(_, s)| {
let s = s.lock().unwrap();
s.left_failures < max_failures || s.right_failures < max_failures
})
.map(|(i, _)| i)
.collect();
if active_indices.is_empty() {
break;
}
let mut edge_kmers: FxHashMap<u64, Vec<(usize, bool, usize)>> = FxHashMap::default();
for &idx in &active_indices {
let state = states[idx].lock().unwrap();
let seq = &state.current_seq;
if seq.len() < k {
continue;
}
if state.left_failures < max_failures {
for offset in 0..self.config.num_edge_kmers.min(seq.len() - k + 1) {
if let Some(hash) = compute_kmer_hash(&seq[offset..offset+k]) {
edge_kmers.entry(hash).or_default().push((idx, true, offset));
}
}
}
if state.right_failures < max_failures {
let seq_len = seq.len();
for offset in 0..self.config.num_edge_kmers.min(seq.len() - k + 1) {
let start = seq_len - k - offset;
if let Some(hash) = compute_kmer_hash(&seq[start..start+k]) {
edge_kmers.entry(hash).or_default().push((idx, false, offset));
}
}
}
}
let left_candidates: Mutex<FxHashMap<usize, Vec<[u32; 4]>>> = Mutex::new(FxHashMap::default());
let right_candidates: Mutex<FxHashMap<usize, Vec<[u32; 4]>>> = Mutex::new(FxHashMap::default());
let max_len = self.config.extension_step;
self.reads.par_iter().for_each(|read_seq| {
scan_read(read_seq, &edge_kmers, &states, k, max_len,
&left_candidates, &right_candidates);
});
let left_candidates = left_candidates.into_inner().unwrap();
let right_candidates = right_candidates.into_inner().unwrap();
let any_extended = std::sync::atomic::AtomicBool::new(false);
active_indices.par_iter().for_each(|&idx| {
let mut state = states[idx].lock().unwrap();
if state.left_failures < max_failures {
if let Some(counts) = left_candidates.get(&idx) {
let consensus = build_consensus_from_counts(
counts,
self.config.min_coverage,
self.config.branching_threshold,
self.config.extension_step,
);
if !consensus.is_empty() {
let n_count = consensus.chars().filter(|&c| c == 'N').count();
let n_ratio = n_count as f64 / consensus.len() as f64;
if n_ratio <= self.config.max_n_ratio {
state.current_seq = format!("{}{}", consensus, state.current_seq);
state.left_failures = 0;
state.left_grown += consensus.len();
if state.left_grown >= self.config.max_extension_per_side {
state.left_failures = max_failures;
}
any_extended.store(true, std::sync::atomic::Ordering::Relaxed);
} else {
state.left_failures += 1;
}
} else {
state.left_failures += 1;
}
} else {
state.left_failures += 1;
}
}
if state.right_failures < max_failures {
if let Some(counts) = right_candidates.get(&idx) {
let consensus = build_consensus_from_counts(
counts,
self.config.min_coverage,
self.config.branching_threshold,
self.config.extension_step,
);
if !consensus.is_empty() {
let n_count = consensus.chars().filter(|&c| c == 'N').count();
let n_ratio = n_count as f64 / consensus.len() as f64;
if n_ratio <= self.config.max_n_ratio {
state.current_seq = format!("{}{}", state.current_seq, consensus);
state.right_failures = 0;
state.right_grown += consensus.len();
if state.right_grown >= self.config.max_extension_per_side {
state.right_failures = max_failures;
}
any_extended.store(true, std::sync::atomic::Ordering::Relaxed);
} else {
state.right_failures += 1;
}
} else {
state.right_failures += 1;
}
} else {
state.right_failures += 1;
}
}
});
if !any_extended.load(std::sync::atomic::Ordering::Relaxed) {
break;
}
}
eprintln!(
" [extender] {} contigs, {} rounds, {} reads, {:.1}s",
contigs.len(), rounds, self.reads.len(), t_start.elapsed().as_secs_f64()
);
let results = states.into_iter().map(|s| {
let s = s.into_inner().unwrap();
ExtendedContig {
name: s.name,
extended_seq: s.current_seq,
}
}).collect();
Ok(results)
}
#[inline]
pub fn extend_all_hybrid(&self, contigs: &[FastaRecord]) -> Result<Vec<ExtendedContig>> {
self.extend_contigs(contigs)
}
}
struct ContigState {
name: String,
current_seq: String,
left_failures: usize,
right_failures: usize,
left_grown: usize,
right_grown: usize,
}
fn compute_kmer_hash(kmer: &str) -> Option<u64> {
let bytes = kmer.as_bytes();
let mut forward = 0u64;
let mut reverse = 0u64;
for (i, &b) in bytes.iter().enumerate() {
let base = match b {
b'A' | b'a' => 0,
b'T' | b't' => 3,
b'G' | b'g' => 1,
b'C' | b'c' => 2,
_ => return None,
};
forward = (forward << 2) | base;
reverse |= (3 - base) << (2 * i);
}
Some(forward.min(reverse))
}
fn check_kmer_match(read_kmer: &str, contig_kmer: &str) -> (bool, bool) {
let is_forward = read_kmer == contig_kmer;
let is_revcomp = if is_forward {
false
} else {
reverse_complement(read_kmer) == contig_kmer
};
(is_forward, is_revcomp)
}
fn reverse_complement(seq: &str) -> String {
seq.chars()
.rev()
.map(|c| match c.to_ascii_uppercase() {
'A' => 'T',
'T' => 'A',
'G' => 'C',
'C' => 'G',
_ => 'N',
})
.collect()
}
#[allow(clippy::too_many_arguments)]
fn scan_read(
read_seq: &str,
edge_kmers: &FxHashMap<u64, Vec<(usize, bool, usize)>>,
states: &[Mutex<ContigState>],
k: usize,
max_len: usize,
left_candidates: &Mutex<FxHashMap<usize, Vec<[u32; 4]>>>,
right_candidates: &Mutex<FxHashMap<usize, Vec<[u32; 4]>>>,
) {
if read_seq.len() < k {
return;
}
let mut local_left: FxHashMap<usize, Vec<[u32; 4]>> = FxHashMap::default();
let mut local_right: FxHashMap<usize, Vec<[u32; 4]>> = FxHashMap::default();
for i in 0..=(read_seq.len() - k) {
let kmer_seq = &read_seq[i..i + k];
if let Some(hash) = compute_kmer_hash(kmer_seq) {
if let Some(matches) = edge_kmers.get(&hash) {
for &(contig_idx, is_left, edge_offset) in matches {
let state = states[contig_idx].lock().unwrap();
let contig_kmer = if is_left {
&state.current_seq[edge_offset..edge_offset + k]
} else {
let clen = state.current_seq.len();
&state.current_seq[clen - k - edge_offset..clen - edge_offset]
};
let (is_forward, is_revcomp) = check_kmer_match(kmer_seq, contig_kmer);
drop(state);
if is_left {
if is_forward && i > edge_offset {
let prefix = &read_seq[..i - edge_offset];
if !prefix.is_empty() {
let ext: String = prefix.chars().rev().collect();
accumulate_counts(local_left.entry(contig_idx).or_default(), &ext, max_len);
}
} else if is_revcomp && i + k + edge_offset < read_seq.len() {
let suffix = &read_seq[i + k + edge_offset..];
if !suffix.is_empty() {
let ext = reverse_complement(suffix);
accumulate_counts(local_left.entry(contig_idx).or_default(), &ext, max_len);
}
}
} else if is_forward && i + k + edge_offset < read_seq.len() {
let suffix = &read_seq[i + k + edge_offset..];
if !suffix.is_empty() {
accumulate_counts(local_right.entry(contig_idx).or_default(), suffix, max_len);
}
} else if is_revcomp && i > edge_offset {
let prefix = &read_seq[..i - edge_offset];
if !prefix.is_empty() {
let ext = reverse_complement(prefix);
accumulate_counts(local_right.entry(contig_idx).or_default(), &ext, max_len);
}
}
}
}
}
}
if !local_left.is_empty() {
let mut global = left_candidates.lock().unwrap();
for (idx, local_counts) in local_left {
merge_counts(global.entry(idx).or_default(), &local_counts);
}
}
if !local_right.is_empty() {
let mut global = right_candidates.lock().unwrap();
for (idx, local_counts) in local_right {
merge_counts(global.entry(idx).or_default(), &local_counts);
}
}
}
fn accumulate_counts(counts: &mut Vec<[u32; 4]>, overhang: &str, max_len: usize) {
for (i, c) in overhang.bytes().enumerate() {
if i >= max_len {
break;
}
let bi = match c.to_ascii_uppercase() {
b'A' => 0,
b'T' => 1,
b'G' => 2,
b'C' => 3,
_ => continue,
};
if i >= counts.len() {
counts.resize(i + 1, [0; 4]);
}
counts[i][bi] += 1;
}
}
fn merge_counts(dst: &mut Vec<[u32; 4]>, src: &[[u32; 4]]) {
if dst.len() < src.len() {
dst.resize(src.len(), [0; 4]);
}
for (d, s) in dst.iter_mut().zip(src.iter()) {
for b in 0..4 {
d[b] += s[b];
}
}
}
fn build_consensus_from_counts(
counts: &[[u32; 4]],
min_coverage: usize,
branching_threshold: f64,
max_len: usize,
) -> String {
let mut result = String::new();
for col in counts.iter().take(max_len) {
let total: u32 = col.iter().sum();
if (total as usize) < min_coverage {
break;
}
let max_idx = col.iter().enumerate().max_by_key(|&(_, &c)| c).map(|(i, _)| i).unwrap_or(0);
let mut sorted = *col;
sorted.sort_unstable_by(|a, b| b.cmp(a));
let minor_freq = sorted[1] as f64 / total as f64;
let base = if minor_freq >= branching_threshold {
'N'
} else {
match max_idx {
0 => 'A',
1 => 'T',
2 => 'G',
3 => 'C',
_ => 'N',
}
};
result.push(base);
}
result
}
#[cfg_attr(not(test), allow(dead_code))]
fn build_consensus_sequence(
sequences: &[String],
min_coverage: usize,
branching_threshold: f64,
max_len: usize,
) -> String {
if sequences.is_empty() {
return String::new();
}
let actual_max_len = sequences.iter().map(|s| s.len()).max().unwrap_or(0).min(max_len);
let mut result = String::new();
for i in 0..actual_max_len {
let bases: Vec<char> = sequences
.iter()
.filter_map(|s| s.chars().nth(i))
.filter(|&c| matches!(c.to_ascii_uppercase(), 'A' | 'T' | 'G' | 'C'))
.collect();
if bases.len() < min_coverage {
break;
}
let mut counts = [0usize; 4]; for &b in &bases {
match b.to_ascii_uppercase() {
'A' => counts[0] += 1,
'T' => counts[1] += 1,
'G' => counts[2] += 1,
'C' => counts[3] += 1,
_ => {}
}
}
let total = counts.iter().sum::<usize>();
let max_idx = counts.iter().enumerate()
.max_by_key(|&(_, &c)| c)
.map(|(i, _)| i)
.unwrap_or(0);
let mut sorted_counts = counts;
sorted_counts.sort_by(|a, b| b.cmp(a));
let second_count = sorted_counts[1];
let minor_freq = second_count as f64 / total as f64;
let base = if minor_freq >= branching_threshold {
'N' } else {
match max_idx {
0 => 'A',
1 => 'T',
2 => 'G',
3 => 'C',
_ => 'N',
}
};
result.push(base);
}
result
}
pub fn write_extended_contigs(results: &[ExtendedContig], path: &Path) -> Result<()> {
let mut writer = BufWriter::new(File::create(path)?);
for result in results {
writeln!(writer, ">{}", result.name)?;
writeln!(writer, "{}", result.extended_seq)?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_compute_kmer_hash() {
let h1 = compute_kmer_hash("ATGC").unwrap();
let h2 = compute_kmer_hash("ATGC").unwrap();
assert_eq!(h1, h2);
let h3 = compute_kmer_hash("GCAT").unwrap();
assert_eq!(h1, h3);
assert!(compute_kmer_hash("ATNG").is_none());
}
#[test]
fn test_reverse_complement() {
assert_eq!(reverse_complement("ATGC"), "GCAT");
assert_eq!(reverse_complement("AAAA"), "TTTT");
assert_eq!(reverse_complement(""), "");
}
#[test]
fn test_build_consensus() {
let seqs = vec![
"ATGC".to_string(),
"ATGC".to_string(),
"ATGC".to_string(),
];
let consensus = build_consensus_sequence(&seqs, 2, 0.2, 100);
assert_eq!(consensus, "ATGC");
}
#[test]
fn test_count_consensus_extension() {
let contig = "GTTCAGACCTAGGCATTACGGATCCGATTACGGCATTAGCCATTAGGCAT";
let ext_truth = "ACAGTGGTCATGCATGCTAGCTAGCATCGAT";
let read = format!("{}{}", &contig[contig.len() - 40..], ext_truth);
let contigs = vec![FastaRecord { name: "c1".into(), seq: contig.to_string() }];
let run = |n_reads: usize| -> String {
let mut cfg = ExtenderConfig::default();
cfg.max_consecutive_failures = 1;
let mut ext = ContigExtender::new(cfg);
ext.reads = std::sync::Arc::new(std::iter::repeat(read.clone()).take(n_reads).collect());
ext.extend_contigs(&contigs).unwrap()[0].extended_seq.clone()
};
let few = run(50);
let many = run(50_000);
assert_eq!(few, many, "extension depends on read depth (should not)");
assert!(few.len() > contig.len(), "no extension occurred");
assert!(few.contains(contig), "original contig not preserved");
}
#[test]
fn test_count_consensus_matches_string_version() {
let seqs = vec![
"ACGT".to_string(),
"ACGT".to_string(),
"ACTT".to_string(), ];
let mut counts: Vec<[u32; 4]> = Vec::new();
for s in &seqs {
accumulate_counts(&mut counts, s, 100);
}
let from_counts = build_consensus_from_counts(&counts, 2, 0.2, 100);
let from_strings = build_consensus_sequence(&seqs, 2, 0.2, 100);
assert_eq!(from_counts, from_strings);
}
fn peak_rss_kb() -> u64 {
std::fs::read_to_string("/proc/self/status").ok()
.and_then(|s| s.lines().find(|l| l.starts_with("VmHWM"))
.and_then(|l| l.split_whitespace().nth(1))
.and_then(|v| v.parse().ok()))
.unwrap_or(0)
}
#[test]
#[ignore]
fn bench_count_memory() {
let n_reads: usize = std::env::var("NREADS").ok().and_then(|v| v.parse().ok()).unwrap_or(2_000_000);
let contig = "GTTCAGACCTAGGCATTACGGATCCGATTACGGCATTAGCCATTAGGCAT";
let ext_truth = "ACAGTGGTCATGCATGCTAGCTAGCATCGAT";
let read = format!("{}{}", &contig[contig.len() - 40..], ext_truth);
let contigs = vec![FastaRecord { name: "c1".into(), seq: contig.to_string() }];
let mut cfg = ExtenderConfig::default();
cfg.max_consecutive_failures = 1;
let mut ext = ContigExtender::new(cfg);
ext.reads = std::sync::Arc::new(std::iter::repeat(read.clone()).take(n_reads).collect());
let reads_rss = peak_rss_kb();
let out = ext.extend_contigs(&contigs).unwrap();
let after = peak_rss_kb();
eprintln!("NREADS={} | reads_loaded={}MB after_extend={}MB | delta_extend={}MB | ext_len={}",
n_reads, reads_rss / 1024, after / 1024, (after.saturating_sub(reads_rss)) / 1024, out[0].extended_seq.len());
}
}