feat: Add superkmer CLI subcommand to obikmer2
Introduces a new `superkmer` command that implements a multi-stage pipeline for reading nucleotide pages and constructing superkmers with configurable parameters. The implementation uses partition-aware scatter writing via `obifastwrite` to distribute output batches across workers, and adds the necessary local dependencies to the project manifest.
This commit is contained in:
Generated
+2
@@ -1654,12 +1654,14 @@ name = "obikmer2"
|
|||||||
version = "1.2.2"
|
version = "1.2.2"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"clap",
|
"clap",
|
||||||
|
"obifastwrite",
|
||||||
"obikalgorithm",
|
"obikalgorithm",
|
||||||
"obikindex",
|
"obikindex",
|
||||||
"obikindexer",
|
"obikindexer",
|
||||||
"obikseq",
|
"obikseq",
|
||||||
"obipipeline",
|
"obipipeline",
|
||||||
"obiread",
|
"obiread",
|
||||||
|
"obiskbuilder",
|
||||||
"obisys",
|
"obisys",
|
||||||
"tracing",
|
"tracing",
|
||||||
"tracing-subscriber",
|
"tracing-subscriber",
|
||||||
|
|||||||
@@ -15,6 +15,8 @@ obisys = { path = "../obisys" }
|
|||||||
obikindex = { path = "../obikindex", default-features = false }
|
obikindex = { path = "../obikindex", default-features = false }
|
||||||
obikindexer = { path = "../obikindexer" }
|
obikindexer = { path = "../obikindexer" }
|
||||||
obikalgorithm = { path = "../obikalgorithm" }
|
obikalgorithm = { path = "../obikalgorithm" }
|
||||||
|
obifastwrite = { path = "../obifastwrite" }
|
||||||
|
obiskbuilder = { path = "../obiskbuilder" }
|
||||||
clap = { version = "4", features = ["derive"] }
|
clap = { version = "4", features = ["derive"] }
|
||||||
tracing = "0.1.44"
|
tracing = "0.1.44"
|
||||||
tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
|
tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
|
||||||
|
|||||||
@@ -1 +1,2 @@
|
|||||||
pub mod index;
|
pub mod index;
|
||||||
|
pub mod superkmer;
|
||||||
|
|||||||
@@ -0,0 +1,72 @@
|
|||||||
|
use std::io::{self, BufWriter, Write};
|
||||||
|
use std::path::PathBuf;
|
||||||
|
|
||||||
|
use clap::Args;
|
||||||
|
use obifastwrite::write_scatter;
|
||||||
|
use obikseq::{RoutableSuperKmer, set_k, set_m};
|
||||||
|
|
||||||
|
use obipipeline::{Throttled, throttle};
|
||||||
|
|
||||||
|
use crate::cli::{CommonArgs, PipelineData, partitions_to_bits};
|
||||||
|
|
||||||
|
#[derive(Args)]
|
||||||
|
pub struct SuperkmerArgs {
|
||||||
|
#[command(flatten)]
|
||||||
|
pub common: CommonArgs,
|
||||||
|
}
|
||||||
|
|
||||||
|
// ── Stage functions ───────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
fn write_batch(
|
||||||
|
batch: Vec<RoutableSuperKmer>,
|
||||||
|
out: &mut BufWriter<io::Stdout>,
|
||||||
|
partition_bits: usize,
|
||||||
|
k: usize,
|
||||||
|
m: usize,
|
||||||
|
) -> io::Result<()> {
|
||||||
|
let partition_mask = (1u64 << partition_bits) - 1;
|
||||||
|
for rsk in batch {
|
||||||
|
let minimizer = *rsk.minimizer();
|
||||||
|
let partition = (minimizer.seq_hash() & partition_mask) as usize;
|
||||||
|
write_scatter(rsk.superkmer(), out, k, m, partition, minimizer)?;
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
// ── Entry point ───────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
pub fn run(args: SuperkmerArgs) {
|
||||||
|
args.common.validate();
|
||||||
|
|
||||||
|
let k = args.common.kmer_size;
|
||||||
|
let m = args.common.minimizer_size;
|
||||||
|
let theta = args.common.theta;
|
||||||
|
let level_max = args.common.level_max;
|
||||||
|
let partition_bits = partitions_to_bits(args.common.partitions);
|
||||||
|
let n_workers = args.common.threads.max(1);
|
||||||
|
let max_open = args.common.effective_max_open();
|
||||||
|
|
||||||
|
set_k(k);
|
||||||
|
set_m(m);
|
||||||
|
|
||||||
|
let path_source = throttle(args.common.seqfile_paths(), max_open);
|
||||||
|
|
||||||
|
let pipe = obipipeline::make_pipe! {
|
||||||
|
PipelineData : Throttled<PathBuf> => Vec<RoutableSuperKmer>,
|
||||||
|
||? {
|
||||||
|
let k = k;
|
||||||
|
move |pw: Throttled<PathBuf>| {
|
||||||
|
let path_str = pw.item.to_str().unwrap_or("").to_owned();
|
||||||
|
let _guard = pw.guard;
|
||||||
|
obiread::open_nuc_stream(&path_str, k)
|
||||||
|
}
|
||||||
|
} : Path => NucPage,
|
||||||
|
| { move |page| obiskbuilder::build_superkmers_page(page, k, level_max, theta) } : NucPage => Batch,
|
||||||
|
};
|
||||||
|
|
||||||
|
let mut out = BufWriter::new(io::stdout());
|
||||||
|
for batch in pipe.apply(path_source, n_workers, 1) {
|
||||||
|
write_batch(batch, &mut out, partition_bits, k, m).expect("write error");
|
||||||
|
}
|
||||||
|
out.flush().expect("flush error");
|
||||||
|
}
|
||||||
@@ -15,6 +15,8 @@ struct Cli {
|
|||||||
enum Commands {
|
enum Commands {
|
||||||
/// Build the complete genome index (scatter → dereplicate → count → layered MPHF)
|
/// Build the complete genome index (scatter → dereplicate → count → layered MPHF)
|
||||||
Index(cmd::index::IndexArgs),
|
Index(cmd::index::IndexArgs),
|
||||||
|
/// Extract super-k-mers from input sequences and scatter them by partition
|
||||||
|
Superkmer(cmd::superkmer::SuperkmerArgs),
|
||||||
}
|
}
|
||||||
|
|
||||||
fn main() {
|
fn main() {
|
||||||
@@ -28,5 +30,6 @@ fn main() {
|
|||||||
let cli = Cli::parse();
|
let cli = Cli::parse();
|
||||||
match cli.command {
|
match cli.command {
|
||||||
Commands::Index(args) => cmd::index::run(args),
|
Commands::Index(args) => cmd::index::run(args),
|
||||||
|
Commands::Superkmer(args) => cmd::superkmer::run(args),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user