use std::io::{BufRead, Read as _};
use std::path::Path;
use indexmap::IndexMap;
use crate::bbi::header::{BbiKind, BED_FIELD_NAMES};
use crate::bbi::writer::{BbiWriter, BbiWriterOptions};
use crate::error::{Error, Result};
use crate::genomic::ChrMap;
use crate::progress::{CancelFlag, ProgressFn, ProgressTracker};
const VALUE_BATCH: usize = 65536;
const PROGRESS_INTERVAL: u64 = 65536;
const MAX_LINE_SIZE: usize = 16 << 20;
const ORDER_HINT: &str = "input must be pooled by chromosome and sorted by start, \
eg with `sort -k1,1 -k2,2n`";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TextFormat {
BedGraph,
Wig,
Bed,
}
impl TextFormat {
pub fn as_str(self) -> &'static str {
match self {
TextFormat::BedGraph => "bedgraph",
TextFormat::Wig => "wig",
TextFormat::Bed => "bed",
}
}
}
#[derive(Debug, Clone)]
pub struct ConvertResult {
pub format: TextFormat,
pub line_count: u64,
pub item_count: u64,
pub skipped_count: u64,
pub clipped_count: u64,
pub chr_sizes: Vec<(String, i64)>,
}
struct LineReader {
input: crate::source::TextInput,
path: String,
size: u64,
consumed: u64,
line_number: u64,
buffer: Vec<u8>,
}
impl LineReader {
fn open(path: &Path) -> Result<Self> {
let (input, size) = crate::source::open_text(path)?;
Ok(Self {
input,
path: path.to_string_lossy().into_owned(),
size,
consumed: 0,
line_number: 0,
buffer: Vec::with_capacity(4096),
})
}
fn read_line_into(&mut self, out: &mut String) -> Result<bool> {
let Some(line) = self.next_line()? else {
return Ok(false);
};
out.clear();
out.push_str(line);
Ok(true)
}
fn next_line(&mut self) -> Result<Option<&str>> {
self.buffer.clear();
let read = self
.input
.by_ref()
.take(MAX_LINE_SIZE as u64 + 1)
.read_until(b'\n', &mut self.buffer)
.map_err(|e| Error::io(&self.path, e))?;
if read == 0 {
return Ok(None);
}
self.consumed += read as u64;
self.line_number += 1;
if self.buffer.len() > MAX_LINE_SIZE {
return Err(Error::format(
&self.path,
format!(
"line {} is longer than {MAX_LINE_SIZE} bytes",
self.line_number
),
));
}
while matches!(self.buffer.last(), Some(b'\n' | b'\r')) {
self.buffer.pop();
}
if std::str::from_utf8(&self.buffer).is_err() {
self.buffer = String::from_utf8_lossy(&self.buffer)
.into_owned()
.into_bytes();
}
Ok(Some(
std::str::from_utf8(&self.buffer).expect("the buffer above is valid utf-8 or replaced"),
))
}
fn fail(&self, message: impl std::fmt::Display) -> Error {
Error::format(&self.path, format!("line {}: {message}", self.line_number))
}
fn guard(&self, message: impl std::fmt::Display) -> Error {
Error::format(
&self.path,
format!("line {}: {message}\n{ORDER_HINT}", self.line_number),
)
}
}
fn is_blank(c: char) -> bool {
c == ' ' || c == '\t'
}
fn split_blanks(line: &str) -> Vec<&str> {
line.split(is_blank).filter(|f| !f.is_empty()).collect()
}
fn split_tabs(line: &str) -> Vec<&str> {
line.split('\t').collect()
}
fn trim_trailing_tab(fields: &mut Vec<&str>, expected: Option<usize>) {
if fields.len() < 2 || !fields.last().is_some_and(|f| f.is_empty()) {
return;
}
match expected {
None => {
fields.pop();
}
Some(width) if fields.len() == width + 1 => {
fields.pop();
}
Some(_) => {}
}
}
fn starts_with_token(line: &str, token: &str) -> bool {
let bytes = line.as_bytes();
let token = token.as_bytes();
if bytes.len() < token.len() {
return false;
}
if !bytes[..token.len()].eq_ignore_ascii_case(token) {
return false;
}
bytes.len() == token.len() || bytes[token.len()] == b' ' || bytes[token.len()] == b'\t'
}
fn is_skipped_line(line: &str) -> bool {
let trimmed = line.trim_start_matches(is_blank);
trimmed.is_empty()
|| trimmed.starts_with('#')
|| starts_with_token(trimmed, "track")
|| starts_with_token(trimmed, "browser")
}
fn is_declaration(field: &str) -> bool {
starts_with_token(field, "fixedstep") || starts_with_token(field, "variablestep")
}
fn is_bedgraph_record(fields: &[&str]) -> bool {
fields[1].parse::<i64>().is_ok()
&& fields[2].parse::<i64>().is_ok()
&& fields[3].parse::<f64>().is_ok()
}
fn is_orphan_wig_data(fields: &[&str]) -> bool {
!fields.is_empty() && fields.len() <= 2 && fields.iter().all(|f| f.parse::<f64>().is_ok())
}
pub fn sniff_format(first_data_line: &str) -> Result<TextFormat> {
let fields = split_blanks(first_data_line);
if fields.first().is_some_and(|f| is_declaration(f)) {
return Ok(TextFormat::Wig);
}
if fields.len() == 4 && is_bedgraph_record(&fields) {
return Ok(TextFormat::BedGraph);
}
if is_orphan_wig_data(&fields) {
return Err(Error::invalid(
"wig data before any fixedStep or variableStep declaration",
));
}
Err(Error::invalid(format!(
"\"{first_data_line}\" is neither a bedgraph record (chr, start, end, value) \
nor a wig declaration (fixedStep or variableStep), so the format of the input \
cannot be told"
)))
}
#[derive(Debug, Clone, Default)]
pub struct WigDeclaration {
pub fixed_step: bool,
pub chr: String,
pub start: i64,
pub step: i64,
pub span: i64,
}
pub fn parse_wig_declaration(line: &str) -> Result<WigDeclaration> {
let fields = split_blanks(line);
let mut declaration = WigDeclaration {
fixed_step: fields
.first()
.is_some_and(|f| starts_with_token(f, "fixedstep")),
step: 1,
span: 1,
..Default::default()
};
let (mut has_chr, mut has_start) = (false, false);
for field in &fields[1..] {
let Some((key, value)) = field.split_once('=') else {
return Err(Error::invalid(format!(
"\"{field}\" is not a key=value of a wig declaration"
)));
};
let number = |what: &str| -> Result<i64> {
value
.parse::<i64>()
.map_err(|_| Error::invalid(format!("could not read \"{value}\" as a {what}")))
};
match key.to_ascii_lowercase().as_str() {
"chrom" => {
declaration.chr = value.to_string();
has_chr = true;
}
"start" => {
let start = number("start")?;
if start < 1 {
return Err(Error::invalid(format!(
"start {start} is not a 1-based coordinate"
)));
}
declaration.start = start - 1;
has_start = true;
}
"step" => declaration.step = number("step")?,
"span" => declaration.span = number("span")?,
other => {
return Err(Error::invalid(format!(
"{other} is not a wig declaration key (chrom, start, step, span)"
)))
}
}
}
if !has_chr {
return Err(Error::invalid("wig declaration has no chrom"));
}
if declaration.fixed_step && !has_start {
return Err(Error::invalid("fixedStep declaration has no start"));
}
if declaration.step <= 0 {
return Err(Error::invalid(format!(
"step {} must be positive",
declaration.step
)));
}
if declaration.span <= 0 {
return Err(Error::invalid(format!(
"span {} must be positive",
declaration.span
)));
}
Ok(declaration)
}
struct ValueSink {
bin_size: i64,
declared: Option<ChrMap>,
chr: String,
bin: Option<i64>,
bin_sum: f64,
bin_covered: i64,
run_start_bin: Option<i64>,
run_values: Vec<f32>,
item_count: u64,
clipped_count: u64,
last_end: i64,
chr_size: Option<i64>,
}
impl ValueSink {
fn new(bin_size: i64, declared: Option<ChrMap>) -> Self {
Self {
bin_size,
declared,
chr: String::new(),
bin: None,
bin_sum: 0.0,
bin_covered: 0,
run_start_bin: None,
run_values: Vec::new(),
item_count: 0,
clipped_count: 0,
last_end: 0,
chr_size: None,
}
}
fn binning(&self) -> bool {
self.bin_size > 0
}
fn set_chr(&mut self, writer: &mut BbiWriter, chr: &str) -> Result<()> {
if self.chr == chr {
return Ok(());
}
self.finish(writer)?;
self.chr.clear();
self.chr.push_str(chr);
self.last_end = 0;
self.chr_size = None;
if let Some(declared) = &self.declared {
self.chr_size = Some(declared.resolve(chr)?.size);
}
Ok(())
}
fn add(
&mut self,
writer: &mut BbiWriter,
chr: &str,
start: i64,
end: i64,
value: f32,
) -> Result<()> {
self.set_chr(writer, chr)?;
self.item_count += 1;
if !self.binning() {
let chr = std::mem::take(&mut self.chr);
let result = writer.write_value(&chr, start, end, value);
self.chr = chr;
return result;
}
self.accumulate(writer, start, end, value)
}
fn add_run(
&mut self,
writer: &mut BbiWriter,
chr: &str,
start: i64,
span: i64,
values: &[f32],
) -> Result<()> {
if values.is_empty() {
return Ok(());
}
self.set_chr(writer, chr)?;
self.item_count += values.len() as u64;
if !self.binning() {
let chr = std::mem::take(&mut self.chr);
let result = writer.write_values(&chr, start, span, values);
self.chr = chr;
return result;
}
for (i, value) in values.iter().enumerate() {
let i = i as i64;
self.accumulate(writer, start + span * i, start + span * (i + 1), *value)?;
}
Ok(())
}
fn accumulate(
&mut self,
writer: &mut BbiWriter,
start: i64,
mut end: i64,
value: f32,
) -> Result<()> {
if start < 0 {
return Err(Error::invalid(format!("start {start} is negative")));
}
if end <= start {
return Err(Error::invalid(format!(
"end {end} is not past the start {start}"
)));
}
if start < self.last_end {
return Err(Error::invalid(format!(
"{}:{start}-{end} starts before the end {} of the previous value, values \
must be added in order and without overlap",
self.chr, self.last_end
)));
}
if let Some(size) = self.chr_size {
if end > size {
if start >= size {
return Err(Error::invalid(format!(
"{}:{start}-{end} starts past the end of {}, which is {size} bases long",
self.chr, self.chr
)));
}
self.clipped_count += 1;
end = size;
}
}
self.last_end = end;
let mut index = start / self.bin_size;
while index * self.bin_size < end {
if Some(index) != self.bin {
self.close_bin(writer)?;
self.bin = Some(index);
self.bin_sum = 0.0;
self.bin_covered = 0;
}
let overlap = end.min((index + 1) * self.bin_size) - start.max(index * self.bin_size);
self.bin_sum += value as f64 * overlap as f64;
self.bin_covered += overlap;
index += 1;
}
Ok(())
}
fn close_bin(&mut self, writer: &mut BbiWriter) -> Result<()> {
let Some(index) = self.bin.take() else {
return Ok(());
};
let (covered, sum) = (self.bin_covered, self.bin_sum);
self.bin_sum = 0.0;
self.bin_covered = 0;
if covered <= 0 {
return Ok(());
}
let value = (sum / covered as f64) as f32;
if self
.run_start_bin
.is_some_and(|first| index != first + self.run_values.len() as i64)
{
self.flush_run(writer)?;
}
if self.run_start_bin.is_none() {
self.run_start_bin = Some(index);
}
self.run_values.push(value);
if self.run_values.len() >= VALUE_BATCH {
self.flush_run(writer)?;
}
Ok(())
}
fn flush_run(&mut self, writer: &mut BbiWriter) -> Result<()> {
let Some(first) = self.run_start_bin.take() else {
return Ok(());
};
if self.run_values.is_empty() {
return Ok(());
}
let start = first * self.bin_size;
let values = std::mem::take(&mut self.run_values);
let chr = std::mem::take(&mut self.chr);
let result = writer.write_values(&chr, start, self.bin_size, &values);
self.chr = chr;
self.run_values = values;
self.run_values.clear();
result
}
fn finish(&mut self, writer: &mut BbiWriter) -> Result<()> {
self.close_bin(writer)?;
self.flush_run(writer)
}
}
pub fn convert_to_bigwig(
input: &Path,
output: &Path,
bin_size: Option<i64>,
mut options: BbiWriterOptions,
progress: Option<ProgressFn>,
cancel: Option<CancelFlag>,
) -> Result<ConvertResult> {
let bin_size = bin_size.unwrap_or(0);
if bin_size < 0 {
return Err(Error::invalid(format!(
"bin_size {bin_size} must not be negative"
)));
}
options.kind = BbiKind::BigWig;
let declared = options.chr_sizes.clone();
let mut reader = LineReader::open(input)?;
let mut writer = BbiWriter::create(&output.to_string_lossy(), options)?;
let mut sink = ValueSink::new(bin_size, declared);
let tracker = ProgressTracker::with_callback(reader.size, progress);
let mut format: Option<TextFormat> = None;
let mut line_count = 0u64;
let mut declaration = WigDeclaration::default();
let mut declared_yet = false;
let mut run: Vec<f32> = Vec::new();
let mut run_start = 0i64;
let mut reported = 0u64;
let mut line = String::new();
let result = (|| -> Result<()> {
while reader.read_line_into(&mut line)? {
line_count += 1;
if line_count % PROGRESS_INTERVAL == 0 {
tracker.add(reader.consumed - reported);
reported = reader.consumed;
if cancel.as_ref().is_some_and(CancelFlag::is_cancelled) {
return Err(Error::invalid("conversion cancelled"));
}
}
if is_skipped_line(&line) {
continue;
}
let fields = split_blanks(&line);
if fields.is_empty() {
continue;
}
let declaration_line = is_declaration(fields[0]);
if format.is_none() {
format = Some(sniff_format(&line).map_err(|e| reader.fail(e))?);
}
if format == Some(TextFormat::Wig) {
if declaration_line {
flush_run(&mut sink, &mut writer, &declaration, run_start, &mut run)
.map_err(|e| reader.guard(e))?;
declaration = parse_wig_declaration(&line).map_err(|e| reader.fail(e))?;
declared_yet = true;
continue;
}
if !declared_yet {
return Err(
reader.fail("wig data before any fixedStep or variableStep declaration")
);
}
if declaration.fixed_step {
if fields.len() != 1 {
return Err(reader.fail(format!(
"fixedStep data has {} columns, not 1",
fields.len()
)));
}
let value = parse_f32(fields[0]).map_err(|e| reader.fail(e))?;
if declaration.step == declaration.span {
if run.is_empty() {
run_start = declaration.start;
}
run.push(value);
if run.len() >= VALUE_BATCH {
flush_run(&mut sink, &mut writer, &declaration, run_start, &mut run)
.map_err(|e| reader.guard(e))?;
}
} else {
sink.add(
&mut writer,
&declaration.chr,
declaration.start,
declaration.start + declaration.span,
value,
)
.map_err(|e| reader.guard(e))?;
}
declaration.start += declaration.step;
} else {
if fields.len() != 2 {
return Err(reader.fail(format!(
"variableStep data has {} columns, not 2",
fields.len()
)));
}
let start = parse_i64(fields[0], "position").map_err(|e| reader.fail(e))?;
let value = parse_f32(fields[1]).map_err(|e| reader.fail(e))?;
if start < 1 {
return Err(reader.fail(format!("position {start} is not 1-based")));
}
sink.add(
&mut writer,
&declaration.chr,
start - 1,
start - 1 + declaration.span,
value,
)
.map_err(|e| reader.guard(e))?;
}
continue;
}
if declaration_line {
return Err(
reader.fail("wig declaration in what has been read as a bedgraph so far")
);
}
if fields.len() != 4 {
return Err(reader.fail(format!(
"bedgraph record has {} columns, not 4",
fields.len()
)));
}
let start = parse_i64(fields[1], "start").map_err(|e| reader.fail(e))?;
let end = parse_i64(fields[2], "end").map_err(|e| reader.fail(e))?;
let value = parse_f32(fields[3]).map_err(|e| reader.fail(e))?;
sink.add(&mut writer, fields[0], start, end, value)
.map_err(|e| reader.guard(e))?;
}
flush_run(&mut sink, &mut writer, &declaration, run_start, &mut run)
.map_err(|e| reader.guard(e))?;
sink.finish(&mut writer).map_err(|e| reader.guard(e))?;
writer.close()
})();
if let Err(error) = result {
writer.abandon();
return Err(error);
}
tracker.done_report();
Ok(ConvertResult {
format: format.unwrap_or(TextFormat::BedGraph),
line_count,
item_count: sink.item_count,
skipped_count: writer.skipped_count(),
clipped_count: writer.clipped_count() + sink.clipped_count,
chr_sizes: writer.chr_sizes(),
})
}
fn flush_run(
sink: &mut ValueSink,
writer: &mut BbiWriter,
declaration: &WigDeclaration,
run_start: i64,
run: &mut Vec<f32>,
) -> Result<()> {
if run.is_empty() {
return Ok(());
}
let result = sink.add_run(writer, &declaration.chr, run_start, declaration.span, run);
run.clear();
result
}
fn parse_i64(text: &str, what: &str) -> Result<i64> {
text.parse()
.map_err(|_| Error::invalid(format!("could not read \"{text}\" as a {what}")))
}
fn parse_f32(text: &str) -> Result<f32> {
text.parse::<f64>()
.map(|v| v as f32)
.map_err(|_| Error::invalid(format!("could not read \"{text}\" as a number")))
}
const BED_FIELD_STANDARD_TYPES: &[&str] = &[
"string", "uint", "uint", "string", "uint", "string", "uint", "uint", "string", "uint",
"string", "string",
];
pub fn default_bed_fields(col_count: usize) -> IndexMap<String, String> {
(0..col_count)
.map(|index| {
if index < BED_FIELD_NAMES.len() {
(
BED_FIELD_NAMES[index].to_string(),
BED_FIELD_STANDARD_TYPES[index].to_string(),
)
} else {
(format!("field{}", index + 1), "string".to_string())
}
})
.collect()
}
pub fn convert_to_bigbed(
input: &Path,
output: &Path,
mut options: BbiWriterOptions,
progress: Option<ProgressFn>,
cancel: Option<CancelFlag>,
) -> Result<ConvertResult> {
options.kind = BbiKind::BigBed;
let mut reader = LineReader::open(input)?;
let tracker = ProgressTracker::with_callback(reader.size, progress);
let mut writer: Option<BbiWriter> = None;
let mut declared_fields = std::mem::take(&mut options.fields);
let mut col_count = 0usize;
let mut line_count = 0u64;
let mut reported = 0u64;
let mut values: IndexMap<String, String> = IndexMap::new();
let mut line = String::new();
let result = (|| -> Result<()> {
while reader.read_line_into(&mut line)? {
line_count += 1;
if line_count % PROGRESS_INTERVAL == 0 {
tracker.add(reader.consumed - reported);
reported = reader.consumed;
if cancel.as_ref().is_some_and(CancelFlag::is_cancelled) {
return Err(Error::invalid("conversion cancelled"));
}
}
if is_skipped_line(&line) {
continue;
}
let mut fields = split_tabs(&line);
trim_trailing_tab(
&mut fields,
if writer.is_none() {
(!declared_fields.is_empty()).then(|| declared_fields.len())
} else {
Some(col_count)
},
);
if fields.len() < 3 {
return Err(reader.fail(format!(
"bed record has {} tab-separated columns, and needs at least 3 \
(chrom, chromStart, chromEnd)",
fields.len()
)));
}
if writer.is_none() {
col_count = fields.len();
if declared_fields.is_empty() {
declared_fields = default_bed_fields(col_count);
} else if declared_fields.len() != col_count {
return Err(reader.fail(format!(
"fields declares {} columns and the first record has {col_count}",
declared_fields.len()
)));
}
let mut opened = BbiWriterOptions {
fields: declared_fields.clone(),
..clone_options(&options)
};
opened.kind = BbiKind::BigBed;
writer = Some(
BbiWriter::create(&output.to_string_lossy(), opened)
.map_err(|e| reader.fail(e))?,
);
for name in declared_fields.keys().skip(3) {
values.insert(name.clone(), String::new());
}
} else if fields.len() != col_count {
return Err(reader.fail(format!(
"bed record has {} columns and the first one had {col_count}; a bigbed \
stores one shape of record",
fields.len()
)));
}
let start = parse_i64(fields[1], "chromStart").map_err(|e| reader.fail(e))?;
let end = parse_i64(fields[2], "chromEnd").map_err(|e| reader.fail(e))?;
for (index, slot) in values.values_mut().enumerate() {
slot.clear();
slot.push_str(fields[index + 3]);
}
writer
.as_mut()
.expect("opened above")
.write_entry(fields[0], start, end, &values)
.map_err(|e| reader.guard(e))?;
}
Ok(())
})();
if let Err(error) = result {
if let Some(writer) = &mut writer {
writer.abandon();
}
return Err(error);
}
let mut writer = match writer {
Some(writer) => writer,
None => {
if declared_fields.is_empty() {
declared_fields = default_bed_fields(3);
}
let mut opened = BbiWriterOptions {
fields: declared_fields,
..clone_options(&options)
};
opened.kind = BbiKind::BigBed;
BbiWriter::create(&output.to_string_lossy(), opened)?
}
};
writer.close()?;
tracker.done_report();
Ok(ConvertResult {
format: TextFormat::Bed,
line_count,
item_count: writer.entry_count(),
skipped_count: writer.skipped_count(),
clipped_count: writer.clipped_count(),
chr_sizes: writer.chr_sizes(),
})
}
fn clone_options(options: &BbiWriterOptions) -> BbiWriterOptions {
BbiWriterOptions {
kind: options.kind,
chr_sizes: options.chr_sizes.clone(),
fields: options.fields.clone(),
items_per_slot: options.items_per_slot,
block_size: options.block_size,
compression_level: options.compression_level,
parallel: options.parallel,
section_policy: options.section_policy,
cost_model: options.cost_model,
}
}