diff --git a/src/Cargo.lock b/src/Cargo.lock index 904168b..583e5a0 100644 --- a/src/Cargo.lock +++ b/src/Cargo.lock @@ -1734,6 +1734,7 @@ dependencies = [ "obiread", "obiskbuilder", "obiskio", + "obisys", "ph", "pprof", "rayon", diff --git a/src/obikmer/Cargo.toml b/src/obikmer/Cargo.toml index 49a5a65..d8a642d 100644 --- a/src/obikmer/Cargo.toml +++ b/src/obikmer/Cargo.toml @@ -17,6 +17,7 @@ obidebruinj = { path = "../obidebruinj" } clap = { version = "4", features = ["derive"] } obikrope = { path = "../obikrope" } obikpartitionner = { path = "../obikpartitionner" } +obisys = { path = "../obisys" } obiskio = { path = "../obiskio" } niffler = "3" rayon = "1" diff --git a/src/obikmer/src/cmd/partition.rs b/src/obikmer/src/cmd/partition.rs index 03cf3cd..073c235 100644 --- a/src/obikmer/src/cmd/partition.rs +++ b/src/obikmer/src/cmd/partition.rs @@ -2,6 +2,7 @@ use std::path::PathBuf; use clap::Args; use obikpartitionner::KmerPartition; +use obisys::{Reporter, Stage}; use obikseq::{RoutableSuperKmer, set_k, set_m}; use tracing::info; @@ -39,14 +40,15 @@ pub fn run(args: PartitionArgs) { }); let path_source = args.common.seqfile_paths(); + let mut rep = Reporter::new(); + let t = Stage::start("scatter"); let pipe = obipipeline::make_pipe! { PipelineData : PathBuf => Vec, ||? { |path| open_chunks(path) } : Path => RawChunk, |? { move |rope| obiread::normalize_sequence_chunk(rope, k) } : RawChunk => NormChunk, | { move |rope| obiskbuilder::build_superkmers(rope, k, level_max, theta) }: NormChunk => Batch, }; - for batch in pipe.apply(path_source, n_workers, 1) { kp.write_batch(batch).unwrap_or_else(|e| { eprintln!("error: {e}"); @@ -54,8 +56,16 @@ pub fn run(args: PartitionArgs) { }); } kp.close().expect("close error"); + rep.push(t.stop()); info!("dereplicating..."); + let t = Stage::start("dereplicate"); kp.dereplicate().expect("dereplicate error"); + rep.push(t.stop()); + + let t = Stage::start("count_kmer"); kp.count_kmer().expect("count kmer error"); + rep.push(t.stop()); + + rep.print(); }