Typed dataclasses for the metrics files that Riker, fgumi, Picard, and fgbio write.
Open a Riker, fgumi, Picard, or fgbio metrics file, plain or compressed, and get a list of frozen dataclasses with typed fields.
Integer columns are int, decimal columns are float, Y and N are bool, categorical columns are enumerations, Picard's ? and fgbio's NaN are NaN, and fgbio's Infinity and -Infinity are infinite.
The package can be installed with pip:
pip install captainslog-ngsRead two Riker metrics files from the same run, and combine their typed fields without casting:
from captainslog.riker import AlignmentMetrics
from captainslog.riker import IsizeMetrics
from captainslog.riker import ReadCategory
from captainslog.riker import RikerReader
with (
RikerReader.from_path[AlignmentMetrics]("riker/s.alignment-metrics.txt") as alignments,
RikerReader.from_path[IsizeMetrics]("riker/s.isize-metrics.txt") as isizes,
):
alignment = {row.category: row for row in alignments}
[isize_metrics] = isizes
frac_fr_reads = 2 * isize_metrics.read_pairs / alignment[ReadCategory.PAIR].total_reads
assert round(frac_fr_reads, 3) == 0.972A RikerReader reads a Riker metrics file into dataclasses, one per row.
from captainslog import riker
from captainslog.riker import HybcapMetrics
from captainslog.riker import IsizeHistogram
with RikerReader.from_path[HybcapMetrics]("riker/s.hybcap-metrics.txt") as reader:
[hybcap] = reader
assert hybcap.sample == "NA12891"
assert hybcap.mean_target_coverage == 70.54
with RikerReader.from_path[IsizeHistogram]("riker/s.isize-histogram.txt") as reader:
isize_histogram = riker.columns(list(reader))
assert isize_histogram.insert_size[:3] == (112, 128, 134)
assert isize_histogram.fr_count[:3] == (1, 1, 1)A PicardReader reads a Picard metrics file into dataclasses, one per row.
from captainslog.picard import HsMetrics
from captainslog.picard import PicardReader
with PicardReader.from_path[HsMetrics]("picard/hs.hs_metrics.txt") as reader:
[hs_metrics] = reader
assert hs_metrics.mean_target_coverage == 70.54
assert hs_metrics.hs_library_size == 14903
assert reader.header.command_line.startswith("CollectHsMetrics BAIT_INTERVALS=")
assert reader.header.started_on == "Sat Oct 24 21:42:22 HST 2026"Tip
A Picard column named for a percent that holds a fraction from 0 to 1 is a frac field, such as frac_pf_reads_aligned for PCT_PF_READS_ALIGNED.
Writing the metrics back out keeps Picard's label for each frac field.
An FgumiReader or FgbioReader reads a metrics file of fgumi or fgbio into dataclasses, one per row.
Both tools write a file under whatever path they are given, so metrics_class names the dataclass of a file by its header row:
from captainslog import fgbio
from captainslog import fgumi
from captainslog.fgbio import DuplexYieldMetric
from captainslog.fgbio import FgbioReader
from captainslog.fgumi import FgumiReader
assert fgbio.metrics_class("fgbio/duplex.duplex_yield_metrics.txt") is DuplexYieldMetric
with FgbioReader.from_path[DuplexYieldMetric]("fgbio/duplex.duplex_yield_metrics.txt") as reader:
*_, library = reader
assert library.fraction == 1.0
assert library.ds_duplexes == 294
assert round(library.ds_fraction_duplexes, 6) == 0.807692fgumi writes eleven of these files with fgbio's columns, such as the duplex yield metrics, so the dataclass of either tool reads a file of the other:
kind = fgumi.metrics_class("fgumi/duplex.duplex_yield_metrics.txt")
with FgumiReader.from_path[kind]("fgumi/duplex.duplex_yield_metrics.txt") as reader:
*_, other = reader
assert (other.ss_families, other.ds_duplexes) == (library.ss_families, library.ds_duplexes)Each reader has a writer, which makes realistic files for unit tests:
RikerWriterFgumiWriterandFgbioWriterPicardWriter, withwrite_histogramPicardHistogramWriterPicardCoverageWriter
from pathlib import Path
from tempfile import TemporaryDirectory
from captainslog.picard import PicardWriter
temp_dir = TemporaryDirectory()
path = Path(temp_dir.name) / "s.hs_metrics.txt"
with PicardWriter.from_path[HsMetrics](path) as writer:
writer.write(hs_metrics)
with PicardReader.from_path[HsMetrics](path) as reader:
assert list(reader) == [hs_metrics]The metrics_class function returns the metric type in a data file, and Picard's histogram_class returns the type of the histogram inside one.
from captainslog import picard
from captainslog.picard import HsHistogram
assert riker.metrics_class("riker/s.hybcap-metrics.txt") is HybcapMetrics
assert riker.metrics_class("riker/s.isize-histogram.txt") is IsizeHistogram
assert picard.metrics_class("picard/hs.hs_metrics.txt") is HsMetrics
assert picard.histogram_class("picard/hs.hs_metrics.txt") is HsHistogramA PicardHistogramReader reads the histogram of a file into one dataclass per bin, and columns transposes the bins into one tuple per column.
For the histograms whose columns Picard names after samples, libraries, and read groups, these functions turn the columns into typed series and the series back into bins:
insert_size_seriesandinsert_size_bins, for anInsertSizeHistogramrna_coverage_seriesandrna_coverage_bins, for anRnaCoverageHistogramlibrary_complexity_seriesandlibrary_complexity_bins, for aLibraryComplexityHistogram
from captainslog.picard import InsertSizeHistogram
from captainslog.picard import PairOrientation
from captainslog.picard import PicardHistogramReader
with PicardHistogramReader.from_path[InsertSizeHistogram]("picard/cmm.insert_size_metrics.txt") as reader:
insert_size_histogram = picard.columns(list(reader))
assert insert_size_histogram.insert_size[:3] == (112, 128, 134)
series = picard.insert_size_series(insert_size_histogram)
all_reads = next(s for s in series if s.name is None and s.orientation is PairOrientation.FR)
assert all_reads.counts[:3] == (1, 1, 1)
assert sum(all_reads.counts) == 478
names = [s.name for s in series]
assert names == [None, "NA12891", "Pond-492101", "NormPond-544932", "NA12891.r1", "NA12891.r2"]Histogram-only files read the same way:
from captainslog.picard import QualityScoreDistribution
with PicardHistogramReader.from_path[QualityScoreDistribution]("picard/cmm.quality_distribution_metrics.txt") as reader:
qualities = list(reader)
assert qualities[0].quality == 1
assert qualities[0].count_of_q == 2Picard writes a row for each accumulation level: all reads, then each sample, library, and read group.
from captainslog.picard import AccumulationLevel
from captainslog.picard import InsertSizeMetrics
with PicardReader.from_path[InsertSizeMetrics]("picard/cmm.insert_size_metrics.txt") as reader:
rows = list(reader)
assert [picard.accumulation_level(row) for row in rows] == [
AccumulationLevel.ALL_READS,
AccumulationLevel.SAMPLE,
AccumulationLevel.LIBRARY,
AccumulationLevel.LIBRARY,
AccumulationLevel.READ_GROUP,
AccumulationLevel.READ_GROUP,
]
[all_reads] = [row for row in rows if picard.accumulation_level(row) is AccumulationLevel.ALL_READS]
assert all_reads.read_pairs == 478
assert all_reads.pair_orientation is PairOrientation.FRCollectHsMetrics and CollectTargetedPcrMetrics can also write plain tables of coverage per target and per base.
from captainslog.picard import PerTargetCoverage
from captainslog.picard import PicardCoverageReader
with PicardCoverageReader.from_path[PerTargetCoverage]("picard/hs.per_target_coverage.txt") as reader:
targets = list(reader)
assert targets[0].name == "rs7555566"
assert targets[0].frac_gc == 0.41To derive metrics, subclass a dataclass, add properties or methods, and read into the subclass.
class MyHsMetrics(HsMetrics):
@property
def frac_bases_on_target(self) -> float:
return self.on_target_bases / self.pf_bases_aligned
with PicardReader.from_path[MyHsMetrics]("picard/hs.hs_metrics.txt") as reader:
[hs_metrics] = reader
assert round(hs_metrics.frac_bases_on_target, 4) == 0.2398With bedspec, find captainslog's targets, bases, SNPs, and CpGs by overlap:
from bedspec import Bed3
from bedspec import register_locatable
from bedspec.ext.captainslog import as_interval
from bedspec.overlap import TreeDetector
gene = Bed3("chr1", start=500, end=2_000)
targets = [
PerTargetCoverage(chrom="chr1", start=1_001, end=1_100, name="exon2"),
PerTargetCoverage(chrom="chr1", start=1_501, end=1_600, name="exon3"),
PerTargetCoverage(chrom="chr2", start=501, end=600, name="exon7"),
]
with register_locatable(PerTargetCoverage, as_interval):
exons = TreeDetector(targets).overlapping(gene)
assert {exon.name for exon in exons} == {"exon2", "exon3"}See the contributing guide for more information.