adding faster handling for dosage calculations

cli/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "biosynth"
-version = "0.1.29"
+version = "0.1.30"
 edition = "2021"
 rust-version = "1.91"
 authors = ["Madhava Jay <madhava@openmined.org>"]
@@ -31,5 +31,6 @@ reqwest = { version = "0.12", default-features = false, features = ["blocking",
 noodles = { version = "0.104.0", features = ["vcf", "tabix", "bgzf", "core"] }
 glob = "0.3"
 dashmap = "6"
+flate2 = "1.1"
 
 [dev-dependencies]

cli/src/commands/cohort_bed.rs

@@ -9,7 +9,7 @@ use anyhow::{bail, Context, Result};
 use dashmap::DashMap;
 use rayon::prelude::*;
 
-use crate::genotype_reader::{detect_delimiter, RowOutcome, RowParser};
+use crate::genotype_reader::{detect_delimiter, open_text_reader, RowOutcome, RowParser};
 use crate::CohortBedArgs;
 
 const LOOKAHEAD_LINES: usize = 2048;
@@ -43,12 +43,20 @@ pub fn run_cohort_bed(args: CohortBedArgs) -> Result<()> {
     // id). After a worker sees the panel once, subsequent samples resolve rsids
     // from this local map with zero shared-map access — so the ~1e9 lookups stop
     // contending on the DashMap shards (which all samples hit identically).
+    // A bad file must not abort the cohort: log it and treat it as an empty
+    // sample (all-missing column) so the rest of the pipeline keeps running.
     let per_sample: Vec<Vec<(u32, u8, u8)>> = samples
         .par_iter()
-        .map_init(HashMap::<String, u32>::new, |local, (_sid, path)| {
-            parse_sample(path, &interner, local)
+        .map_init(HashMap::<String, u32>::new, |local, (sid, path)| {
+            parse_sample(path, &interner, local).unwrap_or_else(|e| {
+                eprintln!(
+                    "⚠️  cohort-bed: skipping unreadable sample {sid} ({}): {e:#}",
+                    path.display()
+                );
+                Vec::new()
+            })
         })
-        .collect::<Result<Vec<_>>>()?;
+        .collect::<Vec<_>>();
     eprintln!(
         "🧬 cohort-bed: parsed {} samples in {:.2}s",
         samples.len(),
@@ -323,9 +331,8 @@ fn parse_sample(
     interner: &Interner,
     local: &mut HashMap<String, u32>,
 ) -> Result<Vec<(u32, u8, u8)>> {
-    use std::io::{BufRead, BufReader};
-    let file = File::open(path).with_context(|| format!("Open {:?}", path))?;
-    let mut reader = BufReader::new(file);
+    use std::io::BufRead;
+    let mut reader = open_text_reader(path)?;
     let mut buffered: Vec<String> = Vec::new();
     let mut buf = String::new();
     while buffered.len() < LOOKAHEAD_LINES {
@@ -344,7 +351,12 @@ fn parse_sample(
     let mut out: Vec<(u32, u8, u8)> = Vec::new();
 
     let mut handle = |parser: &mut RowParser, line: &str| -> Result<()> {
-        if let RowOutcome::Parsed(row) = parser.consume_line(line)? {
+        // A bad row must not break the file: skip it and keep going.
+        let outcome = match parser.consume_line(line) {
+            Ok(o) => o,
+            Err(_) => return Ok(()),
+        };
+        if let RowOutcome::Parsed(row) = outcome {
             // variant_id = rsid, else chrom:pos (matches read_pipeline_genotypes)
             let rsid = if row.rsid.is_empty() {
                 format!("{}:{}", row.chrom, row.pos)
@@ -414,7 +426,7 @@ fn discover_samples(data_dir: &Path) -> Result<Vec<(String, PathBuf)>> {
             .with_context(|| format!("Read dir {:?}", dir))?
             .filter_map(|e| e.ok())
             .map(|e| e.path())
-            .filter(|p| p.is_file() && p.extension().and_then(|x| x.to_str()) == Some("txt"))
+            .filter(|p| p.is_file() && is_genotype_text_file(p))
             .collect();
         txts.sort();
         if let Some(first) = txts.into_iter().next() {
@@ -428,3 +440,15 @@ fn discover_samples(data_dir: &Path) -> Result<Vec<(String, PathBuf)>> {
     }
     Ok(samples)
 }
+
+fn is_genotype_text_file(path: &Path) -> bool {
+    match path.extension().and_then(|x| x.to_str()) {
+        Some(ext) if ext.eq_ignore_ascii_case("txt") => true,
+        Some(ext) if ext.eq_ignore_ascii_case("gz") => path
+            .file_stem()
+            .and_then(|stem| Path::new(stem).extension())
+            .and_then(|x| x.to_str())
+            .is_some_and(|inner| inner.eq_ignore_ascii_case("txt")),
+        _ => false,
+    }
+}

cli/src/commands/fast_allele_freq.rs

@@ -325,7 +325,7 @@ fn write_allele_freq(path: &Path, loci: &LociMap) -> Result<()> {
     Ok(())
 }
 
-fn collect_input_files(inputs: &[PathBuf]) -> Result<Vec<PathBuf>> {
+pub(crate) fn collect_input_files(inputs: &[PathBuf]) -> Result<Vec<PathBuf>> {
     let mut files = Vec::new();
     for input in inputs {
         if input.is_file() {

cli/src/commands/genotype_to_vcf.rs

@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 use std::fs::File;
-use std::io::{BufRead, BufReader, BufWriter, Write};
+use std::io::{BufRead, BufWriter, Write};
 use std::path::{Path, PathBuf};
 use std::process::Command;
 
@@ -9,10 +9,10 @@ use rayon::prelude::*;
 use rayon::ThreadPoolBuilder;
 
 use crate::download::ensure_reference_db;
-use crate::genotype_reader::{detect_delimiter, RowOutcome, RowParser};
+use crate::genotype_reader::{detect_delimiter, open_text_reader, RowOutcome, RowParser};
 use crate::rsid_cache::{default_cache_path, normalize_rsid, RsidCache};
 use crate::stats::{ReferenceVariant, StatsStore};
-use crate::util::collect_input_files;
+use crate::util::{collect_input_files, genotype_file_stem};
 use crate::GenotypeToVcfArgs;
 
 const LOOKAHEAD_LINES: usize = 2048;
@@ -133,8 +133,7 @@ fn convert_file(
     missing_log_path: &Path,
     include_metrics: bool,
 ) -> Result<ConversionStats> {
-    let input_file = File::open(input).with_context(|| format!("Open {:?}", input))?;
-    let mut reader = BufReader::new(input_file);
+    let mut reader = open_text_reader(input)?;
     let mut buffered_lines = Vec::new();
     let mut buffer = String::new();
 
@@ -566,10 +565,7 @@ fn resolve_output_paths(args: &GenotypeToVcfArgs, input: &Path) -> Result<Output
 }
 
 fn default_output_path(input: &Path, outdir: Option<&PathBuf>, gzip: bool) -> PathBuf {
-    let base_name = input
-        .file_stem()
-        .and_then(|stem| stem.to_str())
-        .unwrap_or("output");
+    let base_name = genotype_file_stem(input).unwrap_or_else(|| "output".to_string());
     let filename = if gzip {
         format!("{base_name}.vcf.gz")
     } else {
@@ -601,9 +597,7 @@ fn default_sample_name(input: &Path) -> String {
             return prefix.to_string();
         }
     }
-    input
-        .file_stem()
-        .and_then(|stem| stem.to_str())
+    genotype_file_stem(input)
         .map(|stem| stem.replace(' ', "_"))
         .filter(|stem| !stem.is_empty())
         .unwrap_or_else(|| "SAMPLE".to_string())