15455217694

Committed 05 Jun 2025 12:02AM UTC coverage: 83.594% (-0.01%) from 83.605%

Build # 15455217694

Build Type

push

github

Committed by

web-flow

Commit Message

feat: Allow for non-diploid genotypes (#476)

* feat: allow for non-diploid genotypes

Run Details

16 of 25 new or added lines in 3 files covered. (64.0%)

61 existing lines in 11 files now uncovered.

2721 of 3255 relevant lines covered (83.59%)

25554.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.19

/src/bcf/buffer.rs

// Copyright 2017 Johannes Köster.
// Licensed under the MIT license (http://opensource.org/licenses/MIT)
// This file may not be copied, modified, or distributed
// except according to those terms.

use std::cmp::Ordering;
use std::collections::{vec_deque, VecDeque};
use std::mem;

use crate::bcf::{self, Read};
use crate::errors::Result;

/// A buffer for BCF records. This allows access regions in a sorted BCF file while iterating
/// over it in a single pass.
/// The buffer is implemented as a ringbuffer, such that extension or movement to the right has
/// linear complexity. The buffer does not use any indexed random access. Hence, for getting a
/// region at the very end of the BCF, you will have to wait until all records before have
/// been read.
#[derive(Debug)]
pub struct RecordBuffer {
    reader: bcf::Reader,
    ringbuffer: VecDeque<bcf::Record>,
    ringbuffer2: VecDeque<bcf::Record>,
    overflow: Option<bcf::Record>,
}

unsafe impl Sync for RecordBuffer {}
unsafe impl Send for RecordBuffer {}

impl RecordBuffer {
    /// Create new buffer.
    pub fn new(reader: bcf::Reader) -> Self {
        RecordBuffer {
            reader,
            ringbuffer: VecDeque::new(),
            ringbuffer2: VecDeque::new(),
            overflow: None,
        }
    }

    fn last_rid(&self) -> Option<u32> {
        self.ringbuffer.back().map(|rec| rec.rid().unwrap())
    }

    fn next_rid(&self) -> Option<u32> {
        self.ringbuffer2.back().map(|rec| rec.rid().unwrap())
    }

    fn swap_buffers(&mut self) {
        // swap with buffer for next rid
        mem::swap(&mut self.ringbuffer2, &mut self.ringbuffer);
        // clear second buffer
        self.ringbuffer2.clear();
    }

    fn drain_left(&mut self, rid: u32, window_start: u64) -> usize {
        // remove records too far left or from wrong rid
        // rec.rid() will always yield Some(), because otherwise we won't put the rec into the
        // buffer.
        let to_remove = self
            .ringbuffer
            .iter()
            .take_while(|rec| (rec.pos() as u64) < window_start || rec.rid().unwrap() != rid)
            .count();
        self.ringbuffer.drain(..to_remove);
        to_remove
    }

    /// Fill the buffer with variants in the given window. The start coordinate has to be right of
    /// the start coordinate of any previous `fill` operation.
    /// Coordinates are 0-based, and end is exclusive.
    /// Returns tuple with numbers of added and deleted records compared to previous fetch.
    pub fn fetch(&mut self, chrom: &[u8], start: u64, end: u64) -> Result<(usize, usize)> {
        // TODO panic if start is left of previous start or we have moved past the given chrom
        // before.
        let rid = self.reader.header.name2rid(chrom)?;
        let mut added = 0;
        let mut deleted = 0;

        // shrink and swap
        match (self.last_rid(), self.next_rid()) {
            (Some(last_rid), _) => {
                if last_rid != rid {
                    deleted = self.ringbuffer.len();
                    self.swap_buffers();
                    added = self.ringbuffer.len();
                // TODO drain left?
                } else {
                    deleted = self.drain_left(rid, start);
                }
            }
            (_, Some(_)) => {
                // TODO is this really necessary? If there was no fetch before, there is nothing
                // to delete.
                deleted = self.ringbuffer.len();
                self.swap_buffers();
                deleted += self.drain_left(rid, start);
                added = self.ringbuffer.len();
            }
            _ => (),
        }

        if !self.ringbuffer2.is_empty() {
            // We have already read beyond the current rid. Hence we can't extend to the right for
            // this rid.
            return Ok((added, deleted));
        }

        // move overflow from last fill into ringbuffer
        if self.overflow.is_some() {
            let pos = self.overflow.as_ref().unwrap().pos() as u64;
            if pos >= start {
                if pos <= end {
                    self.ringbuffer.push_back(self.overflow.take().unwrap());
                    added += 1;
                } else {
                    return Ok((added, deleted));
                }
            } else {
                // discard overflow
                self.overflow.take();
            }
        }

        // extend to the right
        loop {
            let mut rec = self.reader.empty_record();

            if self.reader.read(&mut rec).is_none() {
                // EOF
                break;
            }
            let pos = rec.pos() as u64;
            if let Some(rec_rid) = rec.rid() {
                match rec_rid.cmp(&rid) {
                    Ordering::Equal => {
                        if pos >= end {
                            // Record is beyond our window. Store it anyways but stop.
                            self.overflow = Some(rec);
                            break;
                        } else if pos >= start {
                            // Record is within our window.
                            self.ringbuffer.push_back(rec);
                            added += 1;
                        } else {
                            // Record is upstream of our window, ignore it
                            continue;
                        }
                    }
                    Ordering::Greater => {
                        // record comes from next rid. Store it in second buffer but stop filling.
                        self.ringbuffer2.push_back(rec);
                        break;
                    }
                    _ => {
                        // Record comes from previous rid. Ignore it.
                        continue;
                    }
                }
            } else {
                // skip records without proper rid
                continue;
            }
        }

        Ok((added, deleted))
    }

    /// Iterate over records that have been fetched with `fetch`.
    pub fn iter(&self) -> vec_deque::Iter<'_, bcf::Record> {
        self.ringbuffer.iter()
    }

    /// Iterate over mutable references to records that have been fetched with `fetch`.
    pub fn iter_mut(&mut self) -> vec_deque::IterMut<'_, bcf::Record> {
        self.ringbuffer.iter_mut()
    }

    pub fn len(&self) -> usize {
        self.ringbuffer.len()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::bcf;

    #[test]
    fn test_buffer() {
        let reader = bcf::Reader::from_path("test/test.bcf").unwrap();
        let mut buffer = RecordBuffer::new(reader);

        buffer.fetch(b"1", 100, 10023).unwrap();
        {
            let records: Vec<_> = buffer.iter().collect();
            assert_eq!(records.len(), 2);
            assert_eq!(records[0].pos(), 10021);
            assert_eq!(records[1].pos(), 10022);
        }

        buffer.fetch(b"1", 10023, 10024).unwrap();
        {
            let records: Vec<_> = buffer.iter().collect();
            assert_eq!(records.len(), 1);
            assert_eq!(records[0].pos(), 10023);
        }
    }
}

1	// Copyright 2017 Johannes Köster.
2	// Licensed under the MIT license (http://opensource.org/licenses/MIT)
3	// This file may not be copied, modified, or distributed
4	// except according to those terms.
5
6	use std::cmp::Ordering;
7	use std::collections::{vec_deque, VecDeque};
8	use std::mem;
9
10	use crate::bcf::{self, Read};
11	use crate::errors::Result;
12
13	/// A buffer for BCF records. This allows access regions in a sorted BCF file while iterating
14	/// over it in a single pass.
15	/// The buffer is implemented as a ringbuffer, such that extension or movement to the right has
16	/// linear complexity. The buffer does not use any indexed random access. Hence, for getting a
17	/// region at the very end of the BCF, you will have to wait until all records before have
18	/// been read.
19	#[derive(Debug)]
20	pub struct RecordBuffer {
21	reader: bcf::Reader,
22	ringbuffer: VecDeque<bcf::Record>,
23	ringbuffer2: VecDeque<bcf::Record>,
24	overflow: Option<bcf::Record>,
25	}
26
27	unsafe impl Sync for RecordBuffer {}
28	unsafe impl Send for RecordBuffer {}
29
30	impl RecordBuffer {
31	/// Create new buffer.
32	pub fn new(reader: bcf::Reader) -> Self {	2✔
33	RecordBuffer {
34	reader,
35	ringbuffer: VecDeque::new(),	3✔
36	ringbuffer2: VecDeque::new(),	2✔
37	overflow: None,
38	}
39	}
40
41	fn last_rid(&self) -> Option<u32> {	3✔
42	self.ringbuffer.back().map(\|rec\| rec.rid().unwrap())	12✔
43	}
44
45	fn next_rid(&self) -> Option<u32> {	3✔
46	self.ringbuffer2.back().map(\|rec\| rec.rid().unwrap())	7✔
47	}
48
49	fn swap_buffers(&mut self) {	×
50	// swap with buffer for next rid
51	mem::swap(&mut self.ringbuffer2, &mut self.ringbuffer);	×
52	// clear second buffer
53	self.ringbuffer2.clear();	×
54	}
55
56	fn drain_left(&mut self, rid: u32, window_start: u64) -> usize {	2✔
57	// remove records too far left or from wrong rid
58	// rec.rid() will always yield Some(), because otherwise we won't put the rec into the
59	// buffer.
60	let to_remove = self	2✔
61	.ringbuffer	1✔
62	.iter()
63	.take_while(\|rec\| (rec.pos() as u64) < window_start \|\| rec.rid().unwrap() != rid)	6✔
64	.count();
65	self.ringbuffer.drain(..to_remove);	4✔
66	to_remove	1✔
67	}
68
69	/// Fill the buffer with variants in the given window. The start coordinate has to be right of
70	/// the start coordinate of any previous `fill` operation.
71	/// Coordinates are 0-based, and end is exclusive.
72	/// Returns tuple with numbers of added and deleted records compared to previous fetch.
73	pub fn fetch(&mut self, chrom: &[u8], start: u64, end: u64) -> Result<(usize, usize)> {	3✔
74	// TODO panic if start is left of previous start or we have moved past the given chrom
75	// before.
76	let rid = self.reader.header.name2rid(chrom)?;	7✔
77	let mut added = 0;	1✔
78	let mut deleted = 0;	1✔
79
80	// shrink and swap
81	match (self.last_rid(), self.next_rid()) {	2✔
82	(Some(last_rid), _) => {	2✔
UNCOV 83	if last_rid != rid {	2✔
84	deleted = self.ringbuffer.len();	×
85	self.swap_buffers();	×
86	added = self.ringbuffer.len();	×
87	// TODO drain left?
88	} else {
89	deleted = self.drain_left(rid, start);	2✔
90	}
91	}
92	(_, Some(_)) => {	×
93	// TODO is this really necessary? If there was no fetch before, there is nothing
94	// to delete.
95	deleted = self.ringbuffer.len();	×
96	self.swap_buffers();	×
97	deleted += self.drain_left(rid, start);	×
98	added = self.ringbuffer.len();	×
99	}
100	_ => (),	1✔
101	}
102
103	if !self.ringbuffer2.is_empty() {	1✔
104	// We have already read beyond the current rid. Hence we can't extend to the right for
105	// this rid.
106	return Ok((added, deleted));	×
107	}
108
109	// move overflow from last fill into ringbuffer
110	if self.overflow.is_some() {	5✔
111	let pos = self.overflow.as_ref().unwrap().pos() as u64;	4✔
112	if pos >= start {	3✔
113	if pos <= end {	3✔
114	self.ringbuffer.push_back(self.overflow.take().unwrap());	6✔
115	added += 1;	2✔
116	} else {
117	return Ok((added, deleted));	×
118	}
119	} else {
120	// discard overflow
121	self.overflow.take();	×
122	}
123	}
124
125	// extend to the right
126	loop {	1✔
127	let mut rec = self.reader.empty_record();	13✔
128
129	if self.reader.read(&mut rec).is_none() {	14✔
130	// EOF
131	break;
132	}
133	let pos = rec.pos() as u64;	1✔
134	if let Some(rec_rid) = rec.rid() {	5✔
135	match rec_rid.cmp(&rid) {	2✔
136	Ordering::Equal => {
137	if pos >= end {	5✔
138	// Record is beyond our window. Store it anyways but stop.
139	self.overflow = Some(rec);	3✔
140	break;
141	} else if pos >= start {	5✔
142	// Record is within our window.
143	self.ringbuffer.push_back(rec);	3✔
144	added += 1;	3✔
145	} else {
146	// Record is upstream of our window, ignore it
147	continue;
148	}
149	}
150	Ordering::Greater => {
151	// record comes from next rid. Store it in second buffer but stop filling.
152	self.ringbuffer2.push_back(rec);	×
153	break;
154	}
155	_ => {
156	// Record comes from previous rid. Ignore it.
157	continue;
158	}
159	}
160	} else {
161	// skip records without proper rid
162	continue;
163	}
164	}
165
166	Ok((added, deleted))	3✔
167	}
168
169	/// Iterate over records that have been fetched with `fetch`.
170	pub fn iter(&self) -> vec_deque::Iter<'_, bcf::Record> {	3✔
171	self.ringbuffer.iter()	5✔
172	}
173
174	/// Iterate over mutable references to records that have been fetched with `fetch`.
175	pub fn iter_mut(&mut self) -> vec_deque::IterMut<'_, bcf::Record> {	×
176	self.ringbuffer.iter_mut()	×
177	}
178
179	pub fn len(&self) -> usize {	×
180	self.ringbuffer.len()	×
181	}
182
183	pub fn is_empty(&self) -> bool {	×
184	self.len() == 0	×
185	}
186	}
187
188	#[cfg(test)]
189	mod tests {
190	use super::*;
191	use crate::bcf;
192
193	#[test]
194	fn test_buffer() {
195	let reader = bcf::Reader::from_path("test/test.bcf").unwrap();
196	let mut buffer = RecordBuffer::new(reader);
197
198	buffer.fetch(b"1", 100, 10023).unwrap();
199	{
200	let records: Vec<_> = buffer.iter().collect();
201	assert_eq!(records.len(), 2);
202	assert_eq!(records[0].pos(), 10021);
203	assert_eq!(records[1].pos(), 10022);
204	}
205
206	buffer.fetch(b"1", 10023, 10024).unwrap();
207	{
208	let records: Vec<_> = buffer.iter().collect();
209	assert_eq!(records.len(), 1);
210	assert_eq!(records[0].pos(), 10023);
211	}
212	}
213	}

rust-bio / rust-htslib / 15455217694

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous