captainslog-ngs 0.1.1

Typed dataclasses for the metrics files that Riker and Picard write.

Open a Riker or Picard 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, and Picard's ? is NaN.

Installation

The package can be installed with pip:

pip install captainslog-ngs

At a Glance

Read 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.972

Quickstart

Reading Riker Metrics

A 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)

Reading Picard Metrics

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"

Writing Metrics

Each reader has a writer, which makes realistic files for unit tests:

• RikerWriter
• PicardWriter, with write_histogram
• PicardHistogramWriter
• PicardCoverageWriter

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]

Interacting with Metrics

Finding the Class of a File

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 HsHistogram

Histograms and Columns

A 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_series and insert_size_bins, for an InsertSizeHistogram
• rna_coverage_series and rna_coverage_bins, for an RnaCoverageHistogram
• library_complexity_series and library_complexity_bins, for a LibraryComplexityHistogram

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 == 2

Accumulation Levels

Picard 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.FR

Per-Target and Per-Base Coverage

CollectHsMetrics 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.41

Deriving Metrics

To 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.2398

Intervals with bedspec

With 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"}

Development and Testing

See the contributing guide for more information.