API reference

Generated from the package's own docstrings. The names here are everything ligase exports; anything not listed is an implementation detail and may change without notice — including ligase._core, the compiled module.

Loading data

A loader reads what it can from the file and takes the rest as declarations. What it inferred, and from where, is recorded on the result: dataset.dimension("reference").source says declared or names the header it came from.

ligase

Ligase: type-safe, reproducible computational biology pipelines.

Biological data types carry dimensions (organism, reference build, coordinate system, sample set), and the engine refuses to connect steps whose dimensions do not match -- like units in physics.

>>> import ligase as lg
>>> from ligase.types import Alignment, Reads, Reference
>>> hxb2 = lg.reference("refs/HXB2.fasta", organism="HIV-1", build="HXB2")
>>> samples = lg.samples("data/*.fastq.gz", paired=True)
>>> @lg.step(tool="minimap2")
... def align(reads: Reads, ref: Reference) -> Alignment: ...
>>> pipe = lg.Pipeline(samples.map(lambda s: align(s, hxb2)))
>>> pipe.check()   # type-checks every step; nothing runs
>>> pipe.run()     # parallel, cached, per-sample fan-out

reference

reference(path: str, *, organism: str | None = None, build: str | None = None) -> Reference

Load a reference FASTA, inferring its contigs.

Each contig's MD5 is computed here, over the uppercased sequence, as the SAM specification defines @SQ M5. Computing it ourselves is what lets us check an aligner's claim about which reference it used.

Parameters:

Name	Type	Description	Default
path	str	The FASTA, optionally gzipped.	required
organism	str | None	The organism, e.g. HIV-1. A FASTA cannot record this.	None
build	str | None	A human label such as HXB2 or hg38. Display only -- reference identity never rests on a label, so two files both labelled hg38 still fail to match if their contigs disagree.	None

Returns:

Type	Description
Reference	The loaded reference.

Raises:

Type	Description
LoadError	If the file cannot be read or is not a FASTA.

Source code in python/ligase/__init__.py

def reference(
    path: str,
    *,
    organism: str | None = None,
    build: str | None = None,
) -> Reference:
    """Load a reference FASTA, inferring its contigs.

    Each contig's MD5 is computed here, over the uppercased sequence, as the
    SAM specification defines `@SQ M5`. Computing it ourselves is what lets
    us check an aligner's claim about which reference it used.

    Args:
        path: The FASTA, optionally gzipped.
        organism: The organism, e.g. `HIV-1`. A FASTA cannot record this.
        build: A human label such as `HXB2` or `hg38`. Display only --
            reference identity never rests on a label, so two files both
            labelled `hg38` still fail to match if their contigs disagree.

    Returns:
        The loaded reference.

    Raises:
        LoadError: If the file cannot be read or is not a FASTA.
    """
    return Reference(_core.load_reference(path, organism, build))

reads

reads(r1: str, r2: str | None = None, *, organism: str | None = None, sample: str | None = None) -> Reads

Load sequencing reads, inferring what a FASTQ can tell us.

The quality encoding comes from the observed score range, and comes back unknown rather than guessed when the scores fall in the range Phred+33 and Phred+64 share. Guessing would shift every quality value by 31, which no downstream tool would notice.

Parameters:

Name	Type	Description	Default
r1	str	The reads, or the first mate of a pair. Optionally gzipped.	required
r2	str | None	The second mate, if the reads are paired.	None
organism	str | None	The organism. A FASTQ does not record it.	None
sample	str | None	The sample identifier, overriding the one taken from the filename.	None

Returns:

Type	Description
Reads	The loaded reads.

Raises:

Type	Description
LoadError	If a file cannot be read or is not a FASTQ.

Source code in python/ligase/__init__.py

def reads(
    r1: str,
    r2: str | None = None,
    *,
    organism: str | None = None,
    sample: str | None = None,
) -> Reads:
    """Load sequencing reads, inferring what a FASTQ can tell us.

    The quality encoding comes from the observed score range, and comes back
    unknown rather than guessed when the scores fall in the range Phred+33
    and Phred+64 share. Guessing would shift every quality value by 31,
    which no downstream tool would notice.

    Args:
        r1: The reads, or the first mate of a pair. Optionally gzipped.
        r2: The second mate, if the reads are paired.
        organism: The organism. A FASTQ does not record it.
        sample: The sample identifier, overriding the one taken from the
            filename.

    Returns:
        The loaded reads.

    Raises:
        LoadError: If a file cannot be read or is not a FASTQ.
    """
    samples = None if sample is None else [sample]
    return Reads(_core.load_reads(r1, r2, organism, samples))

alignment

alignment(path: str, *, organism: str | None = None, build: str | None = None, samples: list[str] | None = None) -> Alignment

Load a SAM or BAM, inferring dimensions from its header.

Both containers yield identical dimensions: a lossless samtools view -b changes nothing biological.

Parameters:

Name	Type	Description	Default
path	str	The alignment file.	required
organism	str | None	The organism. SAM and BAM headers do not record it.	None
build	str | None	A human label for the reference build. Display only.	None
samples	list[str] | None	The sample identifiers, overriding the @RG SM tags.	None

Returns:

Type	Description
Alignment	The loaded alignment.

Raises:

Type	Description
LoadError	If the file cannot be read or is not a SAM/BAM.

Source code in python/ligase/__init__.py

def alignment(
    path: str,
    *,
    organism: str | None = None,
    build: str | None = None,
    samples: list[str] | None = None,
) -> Alignment:
    """Load a SAM or BAM, inferring dimensions from its header.

    Both containers yield identical dimensions: a lossless
    ``samtools view -b`` changes nothing biological.

    Args:
        path: The alignment file.
        organism: The organism. SAM and BAM headers do not record it.
        build: A human label for the reference build. Display only.
        samples: The sample identifiers, overriding the `@RG SM` tags.

    Returns:
        The loaded alignment.

    Raises:
        LoadError: If the file cannot be read or is not a SAM/BAM.
    """
    return Alignment(_core.load_alignment(path, organism, build, samples))

variants

variants(path: str, *, organism: str | None = None, build: str | None = None, samples: list[str] | None = None) -> Variants

Load a VCF, inferring dimensions from its header.

Identity comes from ##contig alone. ##reference is deliberately not trusted: it holds a path, and two runs pointing at /refs/genome.fa can be pointing at different genomes.

Parameters:

Name	Type	Description	Default
path	str	The VCF, optionally gzipped or BGZF-compressed.	required
organism	str | None	The organism. A VCF does not record it.	None
build	str | None	A human label for the reference build. Display only.	None
samples	list[str] | None	The sample identifiers, overriding the header columns.	None

Returns:

Type	Description
Variants	The loaded variant calls.

Raises:

Type	Description
LoadError	If the file cannot be read or is not a VCF.

Source code in python/ligase/__init__.py

def variants(
    path: str,
    *,
    organism: str | None = None,
    build: str | None = None,
    samples: list[str] | None = None,
) -> Variants:
    """Load a VCF, inferring dimensions from its header.

    Identity comes from `##contig` alone. `##reference` is deliberately not
    trusted: it holds a path, and two runs pointing at `/refs/genome.fa` can
    be pointing at different genomes.

    Args:
        path: The VCF, optionally gzipped or BGZF-compressed.
        organism: The organism. A VCF does not record it.
        build: A human label for the reference build. Display only.
        samples: The sample identifiers, overriding the header columns.

    Returns:
        The loaded variant calls.

    Raises:
        LoadError: If the file cannot be read or is not a VCF.
    """
    return Variants(_core.load_variants(path, organism, build, samples))

intervals

intervals(path: str, *, format: str | None = None, organism: str | None = None, build: str | None = None) -> Intervals

Load genomic intervals from a BED or GFF, inferring what they say.

The coordinate convention comes from the format — BED is 0-based half-open, GFF and GTF are 1-based closed — and that is the point: a tool handed the wrong one produces results that are off by one base and look fine.

Contigs come from the records, so they carry no length and no checksum. They are still compared against a reference by name, because comparing what is present is what makes these files checkable at all.

Parameters:

Name	Type	Description	Default
path	str	The BED, GFF3 or GTF, optionally gzipped.	required
format	str | None	bed, gff, gff3 or gtf, when the filename does not say. A format is never guessed: guessing it wrong shifts every interval by a base.	None
organism	str | None	The organism. An interval file does not record it.	None
build	str | None	A human label for the reference build. Display only.	None

Returns:

Type	Description
Intervals	The loaded intervals.

Raises:

Type	Description
LoadError	If the file cannot be read, is not in the format given, or its format cannot be told from its name.

Source code in python/ligase/__init__.py

def intervals(
    path: str,
    *,
    format: str | None = None,  # noqa: A002 - the file's format, not a builtin
    organism: str | None = None,
    build: str | None = None,
) -> Intervals:
    """Load genomic intervals from a BED or GFF, inferring what they say.

    The coordinate convention comes from the format — BED is 0-based
    half-open, GFF and GTF are 1-based closed — and that is the point: a
    tool handed the wrong one produces results that are off by one base and
    look fine.

    Contigs come from the records, so they carry no length and no checksum.
    They are still compared against a reference by name, because comparing
    what is present is what makes these files checkable at all.

    Args:
        path: The BED, GFF3 or GTF, optionally gzipped.
        format: `bed`, `gff`, `gff3` or `gtf`, when the filename does not
            say. A format is never guessed: guessing it wrong shifts every
            interval by a base.
        organism: The organism. An interval file does not record it.
        build: A human label for the reference build. Display only.

    Returns:
        The loaded intervals.

    Raises:
        LoadError: If the file cannot be read, is not in the format given,
            or its format cannot be told from its name.
    """
    return Intervals(_core.load_intervals(path, format, organism, build))

Declaring and running steps

ligase

Ligase: type-safe, reproducible computational biology pipelines.

Biological data types carry dimensions (organism, reference build, coordinate system, sample set), and the engine refuses to connect steps whose dimensions do not match -- like units in physics.

>>> import ligase as lg
>>> from ligase.types import Alignment, Reads, Reference
>>> hxb2 = lg.reference("refs/HXB2.fasta", organism="HIV-1", build="HXB2")
>>> samples = lg.samples("data/*.fastq.gz", paired=True)
>>> @lg.step(tool="minimap2")
... def align(reads: Reads, ref: Reference) -> Alignment: ...
>>> pipe = lg.Pipeline(samples.map(lambda s: align(s, hxb2)))
>>> pipe.check()   # type-checks every step; nothing runs
>>> pipe.run()     # parallel, cached, per-sample fan-out

Pipeline

A graph of steps, built from its outputs and checked before it runs.

Construction walks back from the outputs through the handles each step was given, so a file passed to two steps becomes one node rather than two. Nothing is executed and nothing is read: the graph is built from signatures and from the dimensions the inputs were loaded with.

Source code in python/ligase/graph.py

class Pipeline:
    """A graph of steps, built from its outputs and checked before it runs.

    Construction walks back from the outputs through the handles each step
    was given, so a file passed to two steps becomes one node rather than
    two. Nothing is executed and nothing is read: the graph is built from
    signatures and from the dimensions the inputs were loaded with.
    """

    __slots__ = ("_graph", "_handles", "_outputs", "_roots")

    def __init__(self, outputs: Dataset | Iterable[Dataset]) -> None:
        """Build the graph that produces these outputs.

        Args:
            outputs: One handle, or an iterable of them — what
                `samples.map(...)` returns. A type checker sees handles as
                the data types their steps return, which is how a mis-wired
                pipeline is caught before `check()` ever runs.

        Raises:
            ValueError: If no outputs were given.
            TypeError: If a step was passed something that is neither
                loaded data nor another step's output.
        """
        self._graph = _core.Graph()
        self._handles: dict[int, StepOutput] = {}
        self._outputs: list[object] = list(_as_outputs(outputs))
        if not self._outputs:
            raise ValueError(
                "a Pipeline needs at least one output; pass the handle a step "
                "returned, or the list `samples.map(...)` gives you"
            )
        # Keyed by identity: two datasets loaded from the same file are two
        # nodes, because the user wrote them as two.
        placed: dict[int, int] = {}
        pinned: list[object] = []
        self._roots = [self._place(out, placed, pinned) for out in self._outputs]

    def _place(
        self, value: object, placed: dict[int, int], pinned: list[object]
    ) -> int:
        existing = placed.get(id(value))
        if existing is not None:
            return existing

        if isinstance(value, StepOutput):
            arguments = [
                (parameter.name, parameter.kind, self._place(given, placed, pinned))
                for parameter, given in _checked_inputs(value)
            ]
            node = self._graph.add_step(
                value.step.name,
                value.step.output,
                arguments,
                value.step.coordinates,
            )
            value._attach(self._graph, node)
            # A run asks each handle how its output is produced, so the
            # graph's nodes have to lead back to the handles that made them.
            self._handles[node] = value
        elif isinstance(value, Dataset):
            node = self._graph.add_input(value.kind.lower(), value._core)
        else:
            raise TypeError(
                f"a step was given {value!r}, which is neither loaded data nor "
                "another step's output"
            )

        placed[id(value)] = node
        # `id` is only unique while the object lives, so keep them all alive.
        pinned.append(value)
        return node

    @property
    def steps(self) -> tuple[str, ...]:
        """The steps this pipeline runs, in dependency order."""
        return tuple(self._graph.step_names(self._roots))

    def report(self) -> Report:
        """Check every step and return everything found, without raising."""
        raw, summary, text = self._graph.check(self._roots)
        return Report(build(raw), summary, text)

    def errors(self) -> list[CheckError]:
        """Return every problem in this pipeline, without raising.

        Returns:
            One error per problem, in the order the steps run. Empty when
            the pipeline is consistent. Nothing is computed.
        """
        return self.report().errors

    def check(self) -> None:
        """Type-check the whole pipeline. Nothing is computed.

        Every problem is found before anything is raised, so the raised
        error's `errors` attribute holds all of them.

        Raises:
            CheckError: If anything is wrong. The subclass says what:
                `DimensionError` and its own subclasses for data that
                disagrees, `StepTypeError` for a step wired to the wrong
                kind of data.
        """
        errors = self.errors()
        if errors:
            raise errors[0]

    def run(
        self,
        *,
        jobs: int | None = None,
        force: bool = False,
        cache: str | Path | None = None,
    ) -> RunReport:
        """Check the pipeline, then compute whatever is not already known.

        The pipeline is checked first, every time: running an unchecked
        pipeline is how an overnight run ends in a dimension error.

        A step runs when something that could change its result has changed
        — its code, a parameter, an input's contents, or the version of a
        tool it calls — and not otherwise. Independent steps run at once,
        and a step whose rerun produced the same output as before leaves
        everything below it alone.

        Args:
            jobs: How many steps may run at once. Defaults to the number of
                cores.
            force: Run every step, cache or no cache. For the one case the
                cache cannot see: a file rewritten in place, keeping its
                length and its timestamp.
            cache: Where to keep the cache. Defaults to `.ligase` under the
                working directory.

        Returns:
            What every step did, and where its output is.

        Raises:
            CheckError: If the pipeline does not type-check. Nothing runs.
            RunError: If any step failed. Everything that did succeed is on
                the error's `report`, which for a cohort is usually most of
                it.
        """
        self.check()
        tasks, callbacks = self._plan()
        directory = Path(cache) if cache is not None else Path.cwd() / CACHE_DIR

        finished, failed, summary, text, counted = self._graph.run(
            self._roots,
            tasks,
            str(directory),
            callbacks,
            jobs,
            force,
        )

        steps = []
        for (
            node,
            name,
            outcome,
            key,
            digest,
            canonical,
            output,
            duration,
            notes,
        ) in finished:
            # Tell the handle where its output landed, so a later
            # `lg.handle(x).path` and the next run's bodies can read it.
            produced = self._handles.get(node)
            if produced is not None:
                produced._produced = output
            steps.append(
                StepRun(
                    step=name,
                    sample=self._sample(node),
                    outcome=outcome,
                    output=output,
                    key=key,
                    hash=digest,
                    canonical=canonical,
                    duration_ms=duration,
                    node=node,
                    notes=tuple(notes),
                )
            )

        hashed, prefiltered, canonicalised = counted
        report = RunReport(
            steps=steps,
            failures=[
                StepFailure(
                    step=name,
                    sample=self._sample(node),
                    message=message,
                    node=node,
                )
                for node, name, message in failed
            ],
            summary=summary,
            text=text,
            hashed=hashed,
            prefiltered=prefiltered,
            canonicalised=canonicalised,
        )
        if report.failures:
            raise RunError(report)
        return report

    def audit(
        self,
        step: str,
        *,
        sample: str | None = None,
        jobs: int | None = None,
        cache: str | Path | None = None,
    ) -> AuditReport:
        """Run one step twice and report whether it is reproducible.

        The pipeline is run first, normally and through the cache, so that
        the step being audited has its inputs. Then each node of that step
        is run twice more, into scratch files that are thrown away, and the
        two outputs are compared both as bytes and as meaning.

        Three verdicts, because two would not do. A step whose bytes move
        while its result does not — which is every tool that records its own
        command line — is `reproducible`, not a failure. Only a step whose
        *result* moves is `not reproducible`.

        Args:
            step: The step's name, as written. Under per-sample fan-out
                there is one node of it per sample, and all of them are
                audited.
            sample: Audit only the node for this sample.
            jobs: How many steps may run at once in the preliminary run.
                The audited steps are taken in turn whatever this says, so
                that the durations reported mean something.
            cache: Where to keep the cache. Defaults to `.ligase` under the
                working directory.

        Returns:
            What two runs of each node showed.

        Raises:
            CheckError: If the pipeline does not type-check. Nothing runs.
            ValueError: If no step of that name is in the pipeline, or no
                node of it is for that sample.
        """
        self.check()
        tasks, callbacks = self._plan()
        directory = Path(cache) if cache is not None else Path.cwd() / CACHE_DIR

        audited, failed, verdict, summary, text = self._graph.audit(
            self._roots,
            tasks,
            str(directory),
            callbacks,
            (step, sample),
            jobs,
        )

        return AuditReport(
            steps=[
                StepAudit(
                    step=name,
                    sample=self._sample(node),
                    scope=scope,
                    verdict=step_verdict,
                    first=first,
                    second=second,
                    first_canonical=first_canonical,
                    second_canonical=second_canonical,
                    first_duration_ms=first_ms,
                    second_duration_ms=second_ms,
                    node=node,
                )
                for (
                    node,
                    name,
                    scope,
                    step_verdict,
                    first,
                    second,
                    first_canonical,
                    second_canonical,
                    first_ms,
                    second_ms,
                ) in audited
            ],
            failures=[
                StepFailure(
                    step=name,
                    sample=self._sample(node),
                    message=message,
                    node=node,
                )
                for node, name, message in failed
            ],
            verdict=verdict,
            summary=summary,
            text=text,
        )

    def methods(
        self,
        *,
        bundle: str | Path | None = None,
        verify: str | Path | None = None,
        include: Sequence[str | Path] = (),
        jobs: int | None = None,
        cache: str | Path | None = None,
    ) -> Methods:
        """Say how this pipeline's results were produced.

        The pipeline is run first, normally and through the cache, because a
        provenance record is filed under a step's key and computing keys is
        what a run does. A pipeline already computed costs a cache lookup
        per step, not a recomputation.

        What comes back names, for every step: the tools and the versions
        they ran at, the parameters as they were hashed, the command as
        written, and every reference by accession and checksum. A step whose
        output predates provenance is reported as a gap rather than left
        out, because nothing can reconstruct how it was produced — the
        command that would have said so is deliberately not in the hash.

        Args:
            bundle: Write a reproducibility bundle into this directory.
            verify: Check the bundle in this directory against what this
                machine just produced, and report where they differ.
            include: Files to copy into the bundle as they are, such as the
                pipeline itself.
            jobs: How many steps may run at once in the run beforehand.
            cache: Where to keep the cache. Defaults to `.ligase` under the
                working directory.

        Returns:
            How every step was produced, with the bundle written and the
            verification done, where either was asked for.

        Raises:
            CheckError: If the pipeline does not type-check. Nothing runs.
            RuntimeError: If a step failed, since a pipeline that did not
                finish has no methods section to write; or if a bundle
                cannot be written.
            ValueError: If the directory to verify against is not a bundle.
        """
        self.check()
        tasks, callbacks = self._plan()
        directory = Path(cache) if cache is not None else Path.cwd() / CACHE_DIR

        described, written, checked = self._graph.methods(
            self._roots,
            tasks,
            str(directory),
            callbacks,
            (
                None if bundle is None else str(bundle),
                None if verify is None else str(verify),
                [str(path) for path in include],
            ),
            jobs,
        )

        (
            groups,
            gaps,
            references,
            citations,
            uncited,
            engine,
            platform,
            summary,
            text,
        ) = described

        return Methods(
            steps=[
                StepMethods(
                    step=step,
                    nodes=nodes,
                    scopes=list(scopes),
                    scope_kind=scope_kind,
                    tools=list(tools),
                    parameters=list(parameters),
                    command=command,
                )
                for (
                    step,
                    nodes,
                    scopes,
                    scope_kind,
                    tools,
                    parameters,
                    command,
                ) in groups
            ],
            gaps=[
                MethodsGap(step=step, scopes=list(scopes), nodes=nodes)
                for step, scopes, nodes in gaps
            ],
            references=[
                PinnedReference(
                    param=param,
                    path=path,
                    build=build,
                    accession=accession,
                    contigs=contigs,
                    length=length,
                    checksum=checksum,
                    fingerprint=fingerprint,
                )
                for (
                    param,
                    path,
                    build,
                    accession,
                    contigs,
                    length,
                    checksum,
                    fingerprint,
                ) in references
            ],
            citations=list(citations),
            uncited=list(uncited),
            engine=engine,
            platform=platform,
            summary=summary,
            text=text,
            bundle=_bundle(written),
            verification=_verification(checked),
        )

    def _plan(self) -> tuple[list[_Task], list[_Callback]]:
        """Ask every step how its output is produced, before anything runs.

        This is the one moment a step body is called, and it is called for
        its *answer* rather than for its effect: a command to render, or a
        function to call later. Nothing has been computed, so a body that
        reaches for an input's contents here is a body that cannot work.
        """
        tasks: list[_Task] = []
        callbacks: list[_Callback] = []

        for node, produced in sorted(self._handles.items()):
            recipe = _recipe(produced)
            callback: int | None = None
            command: str | None = None
            if isinstance(recipe, Shell):
                command = recipe.command
            else:
                callback = len(callbacks)
                callbacks.append(_bind(recipe, produced))

            step = produced.step
            tasks.append(
                (
                    node,
                    _source(step.function),
                    command,
                    callback,
                    list(step.tools),
                    [
                        (parameter.name, repr(produced.arguments[parameter.name]))
                        for parameter in step.plain_parameters
                    ],
                    list(step.citations),
                )
            )

        return tasks, callbacks

    def _sample(self, node: int) -> str | None:
        """The one sample a node stands for, if it stands for exactly one."""
        found = self._graph.dataset(node)
        if found is None:
            return None
        samples = found.samples()
        if samples is None or len(samples) != 1:
            return None
        return samples[0]

    def __len__(self) -> int:
        """How many nodes the graph holds, inputs included."""
        return len(self._graph)

    def __repr__(self) -> str:
        """Identify the pipeline by its size."""
        return f"<Pipeline {len(self.steps)} steps, {len(self)} nodes>"

steps property

steps: tuple[str, ...]

The steps this pipeline runs, in dependency order.

__init__

__init__(outputs: Dataset | Iterable[Dataset]) -> None

Build the graph that produces these outputs.

Parameters:

Name	Type	Description	Default
outputs	Dataset | Iterable[Dataset]	One handle, or an iterable of them — what samples.map(...) returns. A type checker sees handles as the data types their steps return, which is how a mis-wired pipeline is caught before check() ever runs.	required

Raises:

Type	Description
ValueError	If no outputs were given.
TypeError	If a step was passed something that is neither loaded data nor another step's output.

Source code in python/ligase/graph.py

def __init__(self, outputs: Dataset | Iterable[Dataset]) -> None:
    """Build the graph that produces these outputs.

    Args:
        outputs: One handle, or an iterable of them — what
            `samples.map(...)` returns. A type checker sees handles as
            the data types their steps return, which is how a mis-wired
            pipeline is caught before `check()` ever runs.

    Raises:
        ValueError: If no outputs were given.
        TypeError: If a step was passed something that is neither
            loaded data nor another step's output.
    """
    self._graph = _core.Graph()
    self._handles: dict[int, StepOutput] = {}
    self._outputs: list[object] = list(_as_outputs(outputs))
    if not self._outputs:
        raise ValueError(
            "a Pipeline needs at least one output; pass the handle a step "
            "returned, or the list `samples.map(...)` gives you"
        )
    # Keyed by identity: two datasets loaded from the same file are two
    # nodes, because the user wrote them as two.
    placed: dict[int, int] = {}
    pinned: list[object] = []
    self._roots = [self._place(out, placed, pinned) for out in self._outputs]

report

report() -> Report

Check every step and return everything found, without raising.

Source code in python/ligase/graph.py

def report(self) -> Report:
    """Check every step and return everything found, without raising."""
    raw, summary, text = self._graph.check(self._roots)
    return Report(build(raw), summary, text)

errors

errors() -> list[CheckError]

Return every problem in this pipeline, without raising.

Returns:

Type	Description
list[CheckError]	One error per problem, in the order the steps run. Empty when
list[CheckError]	the pipeline is consistent. Nothing is computed.

Source code in python/ligase/graph.py

def errors(self) -> list[CheckError]:
    """Return every problem in this pipeline, without raising.

    Returns:
        One error per problem, in the order the steps run. Empty when
        the pipeline is consistent. Nothing is computed.
    """
    return self.report().errors

check

check() -> None

Type-check the whole pipeline. Nothing is computed.

Every problem is found before anything is raised, so the raised error's errors attribute holds all of them.

Raises:

Type	Description
CheckError	If anything is wrong. The subclass says what: DimensionError and its own subclasses for data that disagrees, StepTypeError for a step wired to the wrong kind of data.

Source code in python/ligase/graph.py

def check(self) -> None:
    """Type-check the whole pipeline. Nothing is computed.

    Every problem is found before anything is raised, so the raised
    error's `errors` attribute holds all of them.

    Raises:
        CheckError: If anything is wrong. The subclass says what:
            `DimensionError` and its own subclasses for data that
            disagrees, `StepTypeError` for a step wired to the wrong
            kind of data.
    """
    errors = self.errors()
    if errors:
        raise errors[0]

run

run(*, jobs: int | None = None, force: bool = False, cache: str | Path | None = None) -> RunReport

Check the pipeline, then compute whatever is not already known.

The pipeline is checked first, every time: running an unchecked pipeline is how an overnight run ends in a dimension error.

A step runs when something that could change its result has changed — its code, a parameter, an input's contents, or the version of a tool it calls — and not otherwise. Independent steps run at once, and a step whose rerun produced the same output as before leaves everything below it alone.

Parameters:

Name	Type	Description	Default
jobs	int | None	How many steps may run at once. Defaults to the number of cores.	None
force	bool	Run every step, cache or no cache. For the one case the cache cannot see: a file rewritten in place, keeping its length and its timestamp.	False
cache	str | Path | None	Where to keep the cache. Defaults to .ligase under the working directory.	None

Returns:

Type	Description
RunReport	What every step did, and where its output is.

Raises:

Type	Description
CheckError	If the pipeline does not type-check. Nothing runs.
RunError	If any step failed. Everything that did succeed is on the error's report, which for a cohort is usually most of it.

Source code in python/ligase/graph.py

def run(
    self,
    *,
    jobs: int | None = None,
    force: bool = False,
    cache: str | Path | None = None,
) -> RunReport:
    """Check the pipeline, then compute whatever is not already known.

    The pipeline is checked first, every time: running an unchecked
    pipeline is how an overnight run ends in a dimension error.

    A step runs when something that could change its result has changed
    — its code, a parameter, an input's contents, or the version of a
    tool it calls — and not otherwise. Independent steps run at once,
    and a step whose rerun produced the same output as before leaves
    everything below it alone.

    Args:
        jobs: How many steps may run at once. Defaults to the number of
            cores.
        force: Run every step, cache or no cache. For the one case the
            cache cannot see: a file rewritten in place, keeping its
            length and its timestamp.
        cache: Where to keep the cache. Defaults to `.ligase` under the
            working directory.

    Returns:
        What every step did, and where its output is.

    Raises:
        CheckError: If the pipeline does not type-check. Nothing runs.
        RunError: If any step failed. Everything that did succeed is on
            the error's `report`, which for a cohort is usually most of
            it.
    """
    self.check()
    tasks, callbacks = self._plan()
    directory = Path(cache) if cache is not None else Path.cwd() / CACHE_DIR

    finished, failed, summary, text, counted = self._graph.run(
        self._roots,
        tasks,
        str(directory),
        callbacks,
        jobs,
        force,
    )

    steps = []
    for (
        node,
        name,
        outcome,
        key,
        digest,
        canonical,
        output,
        duration,
        notes,
    ) in finished:
        # Tell the handle where its output landed, so a later
        # `lg.handle(x).path` and the next run's bodies can read it.
        produced = self._handles.get(node)
        if produced is not None:
            produced._produced = output
        steps.append(
            StepRun(
                step=name,
                sample=self._sample(node),
                outcome=outcome,
                output=output,
                key=key,
                hash=digest,
                canonical=canonical,
                duration_ms=duration,
                node=node,
                notes=tuple(notes),
            )
        )

    hashed, prefiltered, canonicalised = counted
    report = RunReport(
        steps=steps,
        failures=[
            StepFailure(
                step=name,
                sample=self._sample(node),
                message=message,
                node=node,
            )
            for node, name, message in failed
        ],
        summary=summary,
        text=text,
        hashed=hashed,
        prefiltered=prefiltered,
        canonicalised=canonicalised,
    )
    if report.failures:
        raise RunError(report)
    return report

audit

audit(step: str, *, sample: str | None = None, jobs: int | None = None, cache: str | Path | None = None) -> AuditReport

Run one step twice and report whether it is reproducible.

The pipeline is run first, normally and through the cache, so that the step being audited has its inputs. Then each node of that step is run twice more, into scratch files that are thrown away, and the two outputs are compared both as bytes and as meaning.

Three verdicts, because two would not do. A step whose bytes move while its result does not — which is every tool that records its own command line — is reproducible, not a failure. Only a step whose result moves is not reproducible.

Parameters:

Name	Type	Description	Default
step	str	The step's name, as written. Under per-sample fan-out there is one node of it per sample, and all of them are audited.	required
sample	str | None	Audit only the node for this sample.	None
jobs	int | None	How many steps may run at once in the preliminary run. The audited steps are taken in turn whatever this says, so that the durations reported mean something.	None
cache	str | Path | None	Where to keep the cache. Defaults to .ligase under the working directory.	None

Returns:

Type	Description
AuditReport	What two runs of each node showed.

Raises:

Type	Description
CheckError	If the pipeline does not type-check. Nothing runs.
ValueError	If no step of that name is in the pipeline, or no node of it is for that sample.

Source code in python/ligase/graph.py

def audit(
    self,
    step: str,
    *,
    sample: str | None = None,
    jobs: int | None = None,
    cache: str | Path | None = None,
) -> AuditReport:
    """Run one step twice and report whether it is reproducible.

    The pipeline is run first, normally and through the cache, so that
    the step being audited has its inputs. Then each node of that step
    is run twice more, into scratch files that are thrown away, and the
    two outputs are compared both as bytes and as meaning.

    Three verdicts, because two would not do. A step whose bytes move
    while its result does not — which is every tool that records its own
    command line — is `reproducible`, not a failure. Only a step whose
    *result* moves is `not reproducible`.

    Args:
        step: The step's name, as written. Under per-sample fan-out
            there is one node of it per sample, and all of them are
            audited.
        sample: Audit only the node for this sample.
        jobs: How many steps may run at once in the preliminary run.
            The audited steps are taken in turn whatever this says, so
            that the durations reported mean something.
        cache: Where to keep the cache. Defaults to `.ligase` under the
            working directory.

    Returns:
        What two runs of each node showed.

    Raises:
        CheckError: If the pipeline does not type-check. Nothing runs.
        ValueError: If no step of that name is in the pipeline, or no
            node of it is for that sample.
    """
    self.check()
    tasks, callbacks = self._plan()
    directory = Path(cache) if cache is not None else Path.cwd() / CACHE_DIR

    audited, failed, verdict, summary, text = self._graph.audit(
        self._roots,
        tasks,
        str(directory),
        callbacks,
        (step, sample),
        jobs,
    )

    return AuditReport(
        steps=[
            StepAudit(
                step=name,
                sample=self._sample(node),
                scope=scope,
                verdict=step_verdict,
                first=first,
                second=second,
                first_canonical=first_canonical,
                second_canonical=second_canonical,
                first_duration_ms=first_ms,
                second_duration_ms=second_ms,
                node=node,
            )
            for (
                node,
                name,
                scope,
                step_verdict,
                first,
                second,
                first_canonical,
                second_canonical,
                first_ms,
                second_ms,
            ) in audited
        ],
        failures=[
            StepFailure(
                step=name,
                sample=self._sample(node),
                message=message,
                node=node,
            )
            for node, name, message in failed
        ],
        verdict=verdict,
        summary=summary,
        text=text,
    )

methods

methods(*, bundle: str | Path | None = None, verify: str | Path | None = None, include: Sequence[str | Path] = (), jobs: int | None = None, cache: str | Path | None = None) -> Methods

Say how this pipeline's results were produced.

The pipeline is run first, normally and through the cache, because a provenance record is filed under a step's key and computing keys is what a run does. A pipeline already computed costs a cache lookup per step, not a recomputation.

What comes back names, for every step: the tools and the versions they ran at, the parameters as they were hashed, the command as written, and every reference by accession and checksum. A step whose output predates provenance is reported as a gap rather than left out, because nothing can reconstruct how it was produced — the command that would have said so is deliberately not in the hash.

Parameters:

Name	Type	Description	Default
bundle	str | Path | None	Write a reproducibility bundle into this directory.	None
verify	str | Path | None	Check the bundle in this directory against what this machine just produced, and report where they differ.	None
include	Sequence[str | Path]	Files to copy into the bundle as they are, such as the pipeline itself.	()
jobs	int | None	How many steps may run at once in the run beforehand.	None
cache	str | Path | None	Where to keep the cache. Defaults to .ligase under the working directory.	None

Returns:

Type	Description
Methods	How every step was produced, with the bundle written and the
Methods	verification done, where either was asked for.

Raises:

Type	Description
CheckError	If the pipeline does not type-check. Nothing runs.
RuntimeError	If a step failed, since a pipeline that did not finish has no methods section to write; or if a bundle cannot be written.
ValueError	If the directory to verify against is not a bundle.

Source code in python/ligase/graph.py

def methods(
    self,
    *,
    bundle: str | Path | None = None,
    verify: str | Path | None = None,
    include: Sequence[str | Path] = (),
    jobs: int | None = None,
    cache: str | Path | None = None,
) -> Methods:
    """Say how this pipeline's results were produced.

    The pipeline is run first, normally and through the cache, because a
    provenance record is filed under a step's key and computing keys is
    what a run does. A pipeline already computed costs a cache lookup
    per step, not a recomputation.

    What comes back names, for every step: the tools and the versions
    they ran at, the parameters as they were hashed, the command as
    written, and every reference by accession and checksum. A step whose
    output predates provenance is reported as a gap rather than left
    out, because nothing can reconstruct how it was produced — the
    command that would have said so is deliberately not in the hash.

    Args:
        bundle: Write a reproducibility bundle into this directory.
        verify: Check the bundle in this directory against what this
            machine just produced, and report where they differ.
        include: Files to copy into the bundle as they are, such as the
            pipeline itself.
        jobs: How many steps may run at once in the run beforehand.
        cache: Where to keep the cache. Defaults to `.ligase` under the
            working directory.

    Returns:
        How every step was produced, with the bundle written and the
        verification done, where either was asked for.

    Raises:
        CheckError: If the pipeline does not type-check. Nothing runs.
        RuntimeError: If a step failed, since a pipeline that did not
            finish has no methods section to write; or if a bundle
            cannot be written.
        ValueError: If the directory to verify against is not a bundle.
    """
    self.check()
    tasks, callbacks = self._plan()
    directory = Path(cache) if cache is not None else Path.cwd() / CACHE_DIR

    described, written, checked = self._graph.methods(
        self._roots,
        tasks,
        str(directory),
        callbacks,
        (
            None if bundle is None else str(bundle),
            None if verify is None else str(verify),
            [str(path) for path in include],
        ),
        jobs,
    )

    (
        groups,
        gaps,
        references,
        citations,
        uncited,
        engine,
        platform,
        summary,
        text,
    ) = described

    return Methods(
        steps=[
            StepMethods(
                step=step,
                nodes=nodes,
                scopes=list(scopes),
                scope_kind=scope_kind,
                tools=list(tools),
                parameters=list(parameters),
                command=command,
            )
            for (
                step,
                nodes,
                scopes,
                scope_kind,
                tools,
                parameters,
                command,
            ) in groups
        ],
        gaps=[
            MethodsGap(step=step, scopes=list(scopes), nodes=nodes)
            for step, scopes, nodes in gaps
        ],
        references=[
            PinnedReference(
                param=param,
                path=path,
                build=build,
                accession=accession,
                contigs=contigs,
                length=length,
                checksum=checksum,
                fingerprint=fingerprint,
            )
            for (
                param,
                path,
                build,
                accession,
                contigs,
                length,
                checksum,
                fingerprint,
            ) in references
        ],
        citations=list(citations),
        uncited=list(uncited),
        engine=engine,
        platform=platform,
        summary=summary,
        text=text,
        bundle=_bundle(written),
        verification=_verification(checked),
    )

__len__

__len__() -> int

How many nodes the graph holds, inputs included.

Source code in python/ligase/graph.py

def __len__(self) -> int:
    """How many nodes the graph holds, inputs included."""
    return len(self._graph)

__repr__

__repr__() -> str

Identify the pipeline by its size.

Source code in python/ligase/graph.py

def __repr__(self) -> str:
    """Identify the pipeline by its size."""
    return f"<Pipeline {len(self.steps)} steps, {len(self)} nodes>"

Step

Bases: Generic[P, R]

A function that records a call instead of running.

Attributes:

Name	Type	Description
function		The undecorated callable. A run calls it once, before anything executes, to ask how the output is produced.
tool		The external tools the step shells out to, as they were given.
tools		Those tools as a tuple, whose versions join the cache key.
coordinates		The coordinate convention the step writes, for the one output kind that does not imply one.
citations		References the step asks to be cited by.

Source code in python/ligase/graph.py

class Step(Generic[P, R]):
    """A function that records a call instead of running.

    Attributes:
        function: The undecorated callable. A run calls it once, before
            anything executes, to ask how the output is produced.
        tool: The external tools the step shells out to, as they were given.
        tools: Those tools as a tuple, whose versions join the cache key.
        coordinates: The coordinate convention the step writes, for the one
            output kind that does not imply one.
        citations: References the step asks to be cited by.
    """

    __slots__ = (
        "_signature",
        "citations",
        "coordinates",
        "function",
        "output",
        "parameters",
        "tool",
        "tools",
    )

    def __init__(
        self,
        function: Callable[P, R],
        *,
        tool: str | Sequence[str] | None = None,
        coordinates: str | None = None,
        cite: str | Sequence[str] | None = None,
    ) -> None:
        """Read a function's signature, refusing one a graph cannot use."""
        self.function = function
        self.tool = tool
        self.tools = _tools(tool)
        self.coordinates = coordinates
        self.citations = _citations(cite)
        self._signature = inspect.signature(function)
        self.output = _output_kind(function)
        self.parameters = _parameters(function, self._signature)

    @property
    def name(self) -> str:
        """The step's name, as it appears in messages."""
        return self.function.__name__

    @property
    def dataset_parameters(self) -> tuple[Parameter, ...]:
        """The parameters that take data, in declaration order.

        Order matters: a dimension is inherited from the first argument that
        states it.
        """
        return tuple(p for p in self.parameters if p.kind is not None)

    @property
    def plain_parameters(self) -> tuple[Parameter, ...]:
        """The parameters that take plain values, in declaration order."""
        return tuple(p for p in self.parameters if p.kind is None)

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
        """Record a call and return a handle on what it would produce.

        The body is not run. Missing or surplus arguments still fail here,
        exactly as they would for the undecorated function.
        """
        bound = self._signature.bind(*args, **kwargs)
        bound.apply_defaults()
        # The handle is not really an `R`. It stands in for one, which is
        # what lets a type checker follow a pipeline it will never run.
        return cast("R", StepOutput(self, dict(bound.arguments)))

    def __repr__(self) -> str:
        """Identify the step by name and what it produces."""
        return f"<Step {self.name} -> {self.output}>"

name property

name: str

The step's name, as it appears in messages.

dataset_parameters property

dataset_parameters: tuple[Parameter, ...]

The parameters that take data, in declaration order.

Order matters: a dimension is inherited from the first argument that states it.

plain_parameters property

plain_parameters: tuple[Parameter, ...]

The parameters that take plain values, in declaration order.

__init__

__init__(function: Callable[P, R], *, tool: str | Sequence[str] | None = None, coordinates: str | None = None, cite: str | Sequence[str] | None = None) -> None

Read a function's signature, refusing one a graph cannot use.

Source code in python/ligase/graph.py

def __init__(
    self,
    function: Callable[P, R],
    *,
    tool: str | Sequence[str] | None = None,
    coordinates: str | None = None,
    cite: str | Sequence[str] | None = None,
) -> None:
    """Read a function's signature, refusing one a graph cannot use."""
    self.function = function
    self.tool = tool
    self.tools = _tools(tool)
    self.coordinates = coordinates
    self.citations = _citations(cite)
    self._signature = inspect.signature(function)
    self.output = _output_kind(function)
    self.parameters = _parameters(function, self._signature)

__call__

__call__(*args: args, **kwargs: kwargs) -> R

Record a call and return a handle on what it would produce.

The body is not run. Missing or surplus arguments still fail here, exactly as they would for the undecorated function.

Source code in python/ligase/graph.py

def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R:
    """Record a call and return a handle on what it would produce.

    The body is not run. Missing or surplus arguments still fail here,
    exactly as they would for the undecorated function.
    """
    bound = self._signature.bind(*args, **kwargs)
    bound.apply_defaults()
    # The handle is not really an `R`. It stands in for one, which is
    # what lets a type checker follow a pipeline it will never run.
    return cast("R", StepOutput(self, dict(bound.arguments)))

__repr__

__repr__() -> str

Identify the step by name and what it produces.

Source code in python/ligase/graph.py

def __repr__(self) -> str:
    """Identify the step by name and what it produces."""
    return f"<Step {self.name} -> {self.output}>"

step

step(function: Callable[P, R]) -> Step[P, R]

step(*, tool: str | Sequence[str] | None = None, coordinates: str | None = None, cite: str | Sequence[str] | None = None) -> Callable[[Callable[P, R]], Step[P, R]]

step(function: Callable[P, R] | None = None, /, *, tool: str | Sequence[str] | None = None, coordinates: str | None = None, cite: str | Sequence[str] | None = None) -> Step[P, R] | Callable[[Callable[P, R]], Step[P, R]]

Make a function a pipeline step, without ever running it.

The signature is read once, here. Each parameter annotated with a Ligase data type becomes an edge in the graph; the rest are plain values. The return annotation is what says the step produces an Alignment, since nothing will ask the body.

Parameters:

Name	Type	Description	Default
function	Callable[P, R] | None	The function to decorate, when used bare as @lg.step.	None
tool	str | Sequence[str] | None	The external tool the step shells out to, e.g. minimap2, or several of them: tool=("minimap2", "samtools") for a body that pipes one into the other. Each one's --version joins the cache key, so an upgrade that changes a result reruns the steps it could have changed.	None
coordinates	str | None	The coordinate convention the step writes, one of "0-based half-open" or "1-based closed". Only meaningful for a step that produces Intervals: alignments and variant calls are 1-based closed by specification, while a tool that writes BED from a VCF changes the convention and must say so.	None
cite	str | Sequence[str] | None	How the step asks to be cited, as one reference or several. ligase methods knows the reference pipeline's own tools; a step using anything else names its reference here, or the methods section reports that tool as uncited. A citation never joins the cache key — it cannot change a result, so adding one must not rerun anything.	None

Returns:

Type	Description
Step[P, R] | Callable[[Callable[P, R]], Step[P, R]]	A Step, which a type checker sees as the function it decorated.

Raises:

Type	Description
StepDefinitionError	If the signature is one a graph cannot use — an unannotated parameter, a missing or non-Ligase return annotation, or *args/**kwargs.
ValueError	If coordinates is not one of the two conventions, or a tool was named as an empty string.

Source code in python/ligase/graph.py

def step(
    function: Callable[P, R] | None = None,
    /,
    *,
    tool: str | Sequence[str] | None = None,
    coordinates: str | None = None,
    cite: str | Sequence[str] | None = None,
) -> Step[P, R] | Callable[[Callable[P, R]], Step[P, R]]:
    """Make a function a pipeline step, without ever running it.

    The signature is read once, here. Each parameter annotated with a Ligase
    data type becomes an edge in the graph; the rest are plain values. The
    return annotation is what says the step produces an `Alignment`, since
    nothing will ask the body.

    Args:
        function: The function to decorate, when used bare as `@lg.step`.
        tool: The external tool the step shells out to, e.g. `minimap2`,
            or several of them: `tool=("minimap2", "samtools")` for a body
            that pipes one into the other. Each one's `--version` joins the
            cache key, so an upgrade that changes a result reruns the steps
            it could have changed.
        coordinates: The coordinate convention the step writes, one of
            `"0-based half-open"` or `"1-based closed"`. Only meaningful for
            a step that produces `Intervals`: alignments and variant calls
            are 1-based closed by specification, while a tool that writes
            BED from a VCF changes the convention and must say so.
        cite: How the step asks to be cited, as one reference or several.
            `ligase methods` knows the reference pipeline's own tools; a
            step using anything else names its reference here, or the
            methods section reports that tool as uncited. A citation never
            joins the cache key — it cannot change a result, so adding one
            must not rerun anything.

    Returns:
        A `Step`, which a type checker sees as the function it decorated.

    Raises:
        StepDefinitionError: If the signature is one a graph cannot use —
            an unannotated parameter, a missing or non-Ligase return
            annotation, or `*args`/`**kwargs`.
        ValueError: If `coordinates` is not one of the two conventions, or
            a tool was named as an empty string.
    """
    if coordinates is not None and coordinates not in COORDINATES:
        raise ValueError(
            f"`{coordinates}` is not a coordinate system; expected "
            + " or ".join(COORDINATES)
        )

    def decorate(inner: Callable[P, R]) -> Step[P, R]:
        return Step(inner, tool=tool, coordinates=coordinates, cite=cite)

    if function is None:
        return decorate
    return decorate(function)

sh

sh(command: str) -> R

Produce the output by running a shell command.

The command is not run here and the string is not formatted here: the engine fills it in once it knows where the inputs are, and runs it at whatever width the machine allows.

Placeholders name the step's own parameters:

• {ref} — the file bound to the parameter ref.
• {reads.r1}, {reads.r2} — the mates of a paired input.
• {min_freq} — a plain parameter's value.
• {out} — the file the step must write. A step that writes anywhere else produces nothing, as far as the engine is concerned.

Every value is substituted as exactly one shell word, so a path with a space in it stays one path. For a literal brace — an awk '{{print $1}}' — double it.

Parameters:

Name	Type	Description	Default
command	str	The command, with placeholders.	required

Returns:

Type	Description
R	A recipe, typed as the data the step declares, so a body reads as
R	return lg.sh(...) under a -> Alignment annotation.

Source code in python/ligase/recipes.py

def sh(command: str) -> R:  # type: ignore[type-var]
    """Produce the output by running a shell command.

    The command is not run here and the string is not formatted here: the
    engine fills it in once it knows where the inputs are, and runs it at
    whatever width the machine allows.

    Placeholders name the step's own parameters:

    - `{ref}` — the file bound to the parameter `ref`.
    - `{reads.r1}`, `{reads.r2}` — the mates of a paired input.
    - `{min_freq}` — a plain parameter's value.
    - `{out}` — the file the step must write. A step that writes anywhere
      else produces nothing, as far as the engine is concerned.

    Every value is substituted as exactly one shell word, so a path with a
    space in it stays one path. For a literal brace — an `awk '{{print $1}}'`
    — double it.

    Args:
        command: The command, with placeholders.

    Returns:
        A recipe, typed as the data the step declares, so a body reads as
        `return lg.sh(...)` under a `-> Alignment` annotation.
    """
    # Typed as the step's own return so that user pipelines type-check as
    # written; at runtime it is a `Shell`, which is what the engine wants.
    return Shell(command)  # type: ignore[return-value]

write

write(function: Callable[[Path], None]) -> R

Produce the output in Python, by writing the file yourself.

function is called with the path to write to. It is the one thing in a run that takes the GIL, so steps written this way do not run in parallel with each other — which is the right trade for an interpretation step and the wrong one for anything that does real work. Prefer lg.sh.

Parameters:

Name	Type	Description	Default
function	Callable[[Path], None]	Called with the path to write to.	required

Returns:

Type	Description
R	A recipe, typed as the data the step declares.

Source code in python/ligase/recipes.py

def write(function: Callable[[Path], None]) -> R:  # type: ignore[type-var]
    """Produce the output in Python, by writing the file yourself.

    `function` is called with the path to write to. It is the one thing in a
    run that takes the GIL, so steps written this way do not run in parallel
    with each other — which is the right trade for an interpretation step
    and the wrong one for anything that does real work. Prefer `lg.sh`.

    Args:
        function: Called with the path to write to.

    Returns:
        A recipe, typed as the data the step declares.
    """
    return Written(function)  # type: ignore[return-value]

handle

handle(output: object) -> StepOutput

The handle behind what a step returned.

A type checker sees align(reads, ref) as returning an Alignment, which is what lets it follow a pipeline and catch a mis-wired one while you are still editing. At runtime the value is a StepOutput — data that does not exist yet — and this is how to reach it without writing a cast at every call site.

Parameters:

Name	Type	Description	Default
output	object	What a step returned.	required

Returns:

Type	Description
StepOutput	The handle, with its step, its arguments and (once a Pipeline has
StepOutput	been built) the dimensions the output will carry.

Raises:

Type	Description
TypeError	If this is not a step's output — loaded data, say, which is already the thing itself.

Source code in python/ligase/graph.py

def handle(output: object) -> StepOutput:
    """The handle behind what a step returned.

    A type checker sees `align(reads, ref)` as returning an `Alignment`,
    which is what lets it follow a pipeline and catch a mis-wired one while
    you are still editing. At runtime the value is a `StepOutput` — data that
    does not exist yet — and this is how to reach it without writing a cast
    at every call site.

    Args:
        output: What a step returned.

    Returns:
        The handle, with its step, its arguments and (once a `Pipeline` has
        been built) the dimensions the output will carry.

    Raises:
        TypeError: If this is not a step's output — loaded data, say, which
            is already the thing itself.
    """
    if not isinstance(output, StepOutput):
        raise TypeError(
            f"{output!r} is not a step's output; only a value a step returned "
            "has a handle behind it"
        )
    return output

Checking without running

ligase

Ligase: type-safe, reproducible computational biology pipelines.

Biological data types carry dimensions (organism, reference build, coordinate system, sample set), and the engine refuses to connect steps whose dimensions do not match -- like units in physics.

>>> import ligase as lg
>>> from ligase.types import Alignment, Reads, Reference
>>> hxb2 = lg.reference("refs/HXB2.fasta", organism="HIV-1", build="HXB2")
>>> samples = lg.samples("data/*.fastq.gz", paired=True)
>>> @lg.step(tool="minimap2")
... def align(reads: Reads, ref: Reference) -> Alignment: ...
>>> pipe = lg.Pipeline(samples.map(lambda s: align(s, hxb2)))
>>> pipe.check()   # type-checks every step; nothing runs
>>> pipe.run()     # parallel, cached, per-sample fan-out

check_step

check_step(step: str, bindings: Mapping[str, Dataset]) -> None

Check a step's inputs, raising on the first disagreement.

Every disagreement is found before anything is raised, so the raised error's errors attribute holds all of them. Nothing is computed.

Parameters:

Name	Type	Description	Default
step	str	The step's name, used in the messages.	required
bindings	Mapping[str, Dataset]	The step's inputs, by the name the step knows each by.	required

Raises:

Type	Description
CheckError	If any two inputs disagree. The subclass says what kind of disagreement it is: ContigNamingError, ContigOrderError or UndeclaredDimensionError, all of which are DimensionErrors.

Source code in python/ligase/__init__.py

def check_step(step: str, bindings: Mapping[str, Dataset]) -> None:
    """Check a step's inputs, raising on the first disagreement.

    Every disagreement is found before anything is raised, so the raised
    error's `errors` attribute holds all of them. Nothing is computed.

    Args:
        step: The step's name, used in the messages.
        bindings: The step's inputs, by the name the step knows each by.

    Raises:
        CheckError: If any two inputs disagree. The subclass says what kind
            of disagreement it is: `ContigNamingError`, `ContigOrderError`
            or `UndeclaredDimensionError`, all of which are
            `DimensionError`s.
    """
    errors = dimension_errors(step, bindings)
    if errors:
        raise errors[0]

dimension_errors

dimension_errors(step: str, bindings: Mapping[str, Dataset]) -> list[CheckError]

Return every dimension disagreement among a step's inputs.

One step on its own, for when you are not building a pipeline. A whole pipeline is checked with lg.Pipeline(...).check(), which also knows what each step requires of the ones before it.

Parameters:

Name	Type	Description	Default
step	str	The step's name, used in the messages.	required
bindings	Mapping[str, Dataset]	The step's inputs, by the name the step knows each by.	required

Returns:

Type	Description
list[CheckError]	One error per disagreement, most general first. Empty if the step's
list[CheckError]	inputs are consistent. Every error carries the full list in its
list[CheckError]	errors attribute.

Source code in python/ligase/__init__.py

def dimension_errors(step: str, bindings: Mapping[str, Dataset]) -> list[CheckError]:
    """Return every dimension disagreement among a step's inputs.

    One step on its own, for when you are not building a pipeline. A whole
    pipeline is checked with `lg.Pipeline(...).check()`, which also knows
    what each step requires of the ones before it.

    Args:
        step: The step's name, used in the messages.
        bindings: The step's inputs, by the name the step knows each by.

    Returns:
        One error per disagreement, most general first. Empty if the step's
        inputs are consistent. Every error carries the full list in its
        `errors` attribute.
    """
    inputs = [(name, data._core) for name, data in bindings.items()]
    return _build_errors(_core.check_step(step, inputs))

Data types

ligase

Ligase: type-safe, reproducible computational biology pipelines.

Biological data types carry dimensions (organism, reference build, coordinate system, sample set), and the engine refuses to connect steps whose dimensions do not match -- like units in physics.

>>> import ligase as lg
>>> from ligase.types import Alignment, Reads, Reference
>>> hxb2 = lg.reference("refs/HXB2.fasta", organism="HIV-1", build="HXB2")
>>> samples = lg.samples("data/*.fastq.gz", paired=True)
>>> @lg.step(tool="minimap2")
... def align(reads: Reads, ref: Reference) -> Alignment: ...
>>> pipe = lg.Pipeline(samples.map(lambda s: align(s, hxb2)))
>>> pipe.check()   # type-checks every step; nothing runs
>>> pipe.run()     # parallel, cached, per-sample fan-out

Dataset

Data of a known kind, carrying the dimensions inferred from it.

Source code in python/ligase/types.py

class Dataset:
    """Data of a known kind, carrying the dimensions inferred from it."""

    __slots__ = ("_core",)

    def __init__(self, core: _core.Dataset) -> None:
        """Wrap a dataset loaded by the core engine."""
        self._core = core

    @property
    def kind(self) -> str:
        """The name of the kind of data this holds."""
        return self._core.kind

    @property
    def path(self) -> str | None:
        """The file this was loaded from."""
        return self._core.path

    @property
    def file(self) -> str:
        """The file this data is, for code that has to open it.

        `path` is the honest answer to "where did this come from, if
        anywhere?", so it can be `None` — a step's output carries dimensions
        worked out before anything ran, and has no file of its own until it
        does. This is the answer for the other question, the one a
        `lg.write` body asks:

            return lg.write(lambda out: interpret(v.file, out))

        Inside that function the steps above have run, so there is a file,
        and saying so in the type is what keeps a body from having to prove
        it.

        Raises:
            ValueError: If this data has no file behind it at all.
        """
        found = self._core.path
        if found is None:
            raise ValueError(
                f"this {self.kind} has no file behind it, so there is "
                "nothing to open. A step's output has none until the step "
                "has run; `path` is the accessor that admits to that"
            )
        return found

    @property
    def applicable(self) -> tuple[str, ...]:
        """The dimensions that mean anything for this kind of data.

        A dimension outside this set is not a gap to be filled in: asking a
        FASTA for its quality encoding is a category error.
        """
        return tuple(self._core.applicable())

    @property
    def dimensions(self) -> Mapping[str, DimensionValue]:
        """Every dimension that has a known value, by name."""
        found = {}
        for name in self._core.applicable():
            pair = self._core.dimension(name)
            if pair is not None:
                found[name] = DimensionValue(name, pair[0], pair[1])
        return MappingProxyType(found)

    def dimension(self, name: str) -> DimensionValue | None:
        """One dimension's value, or None if it is unknown or inapplicable."""
        pair = self._core.dimension(name)
        if pair is None:
            return None
        return DimensionValue(name, pair[0], pair[1])

    def unknown_reason(self, name: str) -> str | None:
        """Why an applicable dimension has no value.

        None when the dimension is known, or when it does not apply to this
        kind of data at all.
        """
        return self._core.unknown_reason(name)

    @property
    def contigs(self) -> list[Contig] | None:
        """The contigs of the reference this data is on, in file order."""
        contigs = self._core.contigs()
        if contigs is None:
            return None
        return [Contig(name, length, md5) for name, length, md5 in contigs]

    @property
    def samples(self) -> list[str] | None:
        """The samples this data covers, in order."""
        return self._core.samples()

    def __repr__(self) -> str:
        """Identify the dataset by kind and origin."""
        where = self.path or "<in memory>"
        return f"<{type(self).__name__} {self.kind} {where}>"

kind property

kind: str

The name of the kind of data this holds.

path property

path: str | None

The file this was loaded from.

file property

file: str

The file this data is, for code that has to open it.

path is the honest answer to "where did this come from, if anywhere?", so it can be None — a step's output carries dimensions worked out before anything ran, and has no file of its own until it does. This is the answer for the other question, the one a lg.write body asks:

return lg.write(lambda out: interpret(v.file, out))

Inside that function the steps above have run, so there is a file, and saying so in the type is what keeps a body from having to prove it.

Raises:

Type	Description
ValueError	If this data has no file behind it at all.

applicable property

applicable: tuple[str, ...]

The dimensions that mean anything for this kind of data.

A dimension outside this set is not a gap to be filled in: asking a FASTA for its quality encoding is a category error.

dimensions property

dimensions: Mapping[str, DimensionValue]

Every dimension that has a known value, by name.

contigs property

contigs: list[Contig] | None

The contigs of the reference this data is on, in file order.

samples property

samples: list[str] | None

The samples this data covers, in order.

__init__

__init__(core: Dataset) -> None

Wrap a dataset loaded by the core engine.

Source code in python/ligase/types.py

def __init__(self, core: _core.Dataset) -> None:
    """Wrap a dataset loaded by the core engine."""
    self._core = core

dimension

dimension(name: str) -> DimensionValue | None

One dimension's value, or None if it is unknown or inapplicable.

Source code in python/ligase/types.py

def dimension(self, name: str) -> DimensionValue | None:
    """One dimension's value, or None if it is unknown or inapplicable."""
    pair = self._core.dimension(name)
    if pair is None:
        return None
    return DimensionValue(name, pair[0], pair[1])

unknown_reason

unknown_reason(name: str) -> str | None

Why an applicable dimension has no value.

None when the dimension is known, or when it does not apply to this kind of data at all.

Source code in python/ligase/types.py

def unknown_reason(self, name: str) -> str | None:
    """Why an applicable dimension has no value.

    None when the dimension is known, or when it does not apply to this
    kind of data at all.
    """
    return self._core.unknown_reason(name)

__repr__

__repr__() -> str

Identify the dataset by kind and origin.

Source code in python/ligase/types.py

def __repr__(self) -> str:
    """Identify the dataset by kind and origin."""
    where = self.path or "<in memory>"
    return f"<{type(self).__name__} {self.kind} {where}>"

Reference

Bases: Dataset

A reference sequence (FASTA).

Source code in python/ligase/types.py

class Reference(Dataset):
    """A reference sequence (FASTA)."""

    __slots__ = ()

Reads

Bases: Dataset

Unaligned sequencing reads (FASTQ).

Source code in python/ligase/types.py

class Reads(Dataset):
    """Unaligned sequencing reads (FASTQ)."""

    __slots__ = ()

Samples

The samples a glob matched, in the order their files appeared.

Iterable, indexable and sized, so it can be used as an ordinary sequence of Reads. map is the part the golden example uses.

Source code in python/ligase/samples.py

class Samples:
    """The samples a glob matched, in the order their files appeared.

    Iterable, indexable and sized, so it can be used as an ordinary sequence
    of `Reads`. `map` is the part the golden example uses.
    """

    __slots__ = ("_samples",)

    def __init__(self, samples: list[Reads]) -> None:
        """Hold loaded reads, one entry per sample."""
        self._samples = samples

    @property
    def ids(self) -> tuple[str, ...]:
        """Each sample's identifier, in order.

        Taken from the filenames, with any mate marker removed, unless the
        reads were loaded with the identifier declared.
        """
        return tuple(_identify(sample) for sample in self._samples)

    def map(self, build: Callable[[Reads], T]) -> list[T]:
        """Apply a chain of steps to every sample.

        Nothing is executed. `build` is called once per sample and returns
        whatever the last step in the chain returns — a handle, not data —
        so the result is one output per sample, ready for `lg.Pipeline`.

        Args:
            build: What to do with one sample's reads.

        Returns:
            One result per sample, in order.
        """
        return [build(sample) for sample in self._samples]

    def __len__(self) -> int:
        """How many samples were found."""
        return len(self._samples)

    def __iter__(self) -> Iterator[Reads]:
        """Iterate the samples in order."""
        return iter(self._samples)

    @overload
    def __getitem__(self, index: int) -> Reads: ...
    @overload
    def __getitem__(self, index: slice) -> list[Reads]: ...

    def __getitem__(self, index: int | slice) -> Reads | list[Reads]:
        """One sample, or a slice of them."""
        return self._samples[index]

    def __repr__(self) -> str:
        """Identify the set by how many samples it holds."""
        return f"<Samples {len(self)} samples: {', '.join(self.ids)}>"

ids property

ids: tuple[str, ...]

Each sample's identifier, in order.

Taken from the filenames, with any mate marker removed, unless the reads were loaded with the identifier declared.

__init__

__init__(samples: list[Reads]) -> None

Hold loaded reads, one entry per sample.

Source code in python/ligase/samples.py

def __init__(self, samples: list[Reads]) -> None:
    """Hold loaded reads, one entry per sample."""
    self._samples = samples

map

map(build: Callable[[Reads], T]) -> list[T]

Apply a chain of steps to every sample.

Nothing is executed. build is called once per sample and returns whatever the last step in the chain returns — a handle, not data — so the result is one output per sample, ready for lg.Pipeline.

Parameters:

Name	Type	Description	Default
build	Callable[[Reads], T]	What to do with one sample's reads.	required

Returns:

Type	Description
list[T]	One result per sample, in order.

Source code in python/ligase/samples.py

def map(self, build: Callable[[Reads], T]) -> list[T]:
    """Apply a chain of steps to every sample.

    Nothing is executed. `build` is called once per sample and returns
    whatever the last step in the chain returns — a handle, not data —
    so the result is one output per sample, ready for `lg.Pipeline`.

    Args:
        build: What to do with one sample's reads.

    Returns:
        One result per sample, in order.
    """
    return [build(sample) for sample in self._samples]

__len__

__len__() -> int

How many samples were found.

Source code in python/ligase/samples.py

def __len__(self) -> int:
    """How many samples were found."""
    return len(self._samples)

__iter__

__iter__() -> Iterator[Reads]

Iterate the samples in order.

Source code in python/ligase/samples.py

def __iter__(self) -> Iterator[Reads]:
    """Iterate the samples in order."""
    return iter(self._samples)

__getitem__

__getitem__(index: int) -> Reads

__getitem__(index: slice) -> list[Reads]

__getitem__(index: int | slice) -> Reads | list[Reads]

One sample, or a slice of them.

Source code in python/ligase/samples.py

def __getitem__(self, index: int | slice) -> Reads | list[Reads]:
    """One sample, or a slice of them."""
    return self._samples[index]

__repr__

__repr__() -> str

Identify the set by how many samples it holds.

Source code in python/ligase/samples.py

def __repr__(self) -> str:
    """Identify the set by how many samples it holds."""
    return f"<Samples {len(self)} samples: {', '.join(self.ids)}>"

Alignment

Bases: Dataset

Reads placed on a reference (SAM/BAM).

Source code in python/ligase/types.py

class Alignment(Dataset):
    """Reads placed on a reference (SAM/BAM)."""

    __slots__ = ()

Variants

Bases: Dataset

Variant calls (VCF).

Source code in python/ligase/types.py

class Variants(Dataset):
    """Variant calls (VCF)."""

    __slots__ = ()

Intervals

Bases: Dataset

Genomic intervals (BED/GFF).

The one type whose coordinates dimension varies by format: BED is 0-based half-open, GFF is 1-based closed. Feeding one to a tool expecting the other produces results that are plausible and off by one.

Source code in python/ligase/types.py

class Intervals(Dataset):
    """Genomic intervals (BED/GFF).

    The one type whose `coordinates` dimension varies by format: BED is
    0-based half-open, GFF is 1-based closed. Feeding one to a tool
    expecting the other produces results that are plausible and off by one.
    """

    __slots__ = ()

Contig dataclass

One sequence in a reference.

Attributes:

Name	Type	Description
name	str	The name as written: chr1, 1, K03455.1.
length	int | None	Length in bases, or None when the file records no length.
md5	str | None	MD5 of the uppercased sequence, as the SAM specification defines @SQ M5, or None when neither recorded nor computable.

Source code in python/ligase/types.py

@dataclass(frozen=True, slots=True)
class Contig:
    """One sequence in a reference.

    Attributes:
        name: The name as written: `chr1`, `1`, `K03455.1`.
        length: Length in bases, or None when the file records no length.
        md5: MD5 of the uppercased sequence, as the SAM specification defines
            `@SQ M5`, or None when neither recorded nor computable.
    """

    name: str
    length: int | None
    md5: str | None

DimensionValue dataclass

A dimension's value, and where it came from.

Attributes:

Name	Type	Description
name	str	The dimension's name, e.g. reference.
value	str	The value, rendered for a human.
source	str	Where it came from, e.g. from BAM @SQ header: data/x.bam. This is the part that makes an error actionable.

Source code in python/ligase/types.py

@dataclass(frozen=True, slots=True)
class DimensionValue:
    """A dimension's value, and where it came from.

    Attributes:
        name: The dimension's name, e.g. `reference`.
        value: The value, rendered for a human.
        source: Where it came from, e.g. `from BAM @SQ header: data/x.bam`.
            This is the part that makes an error actionable.
    """

    name: str
    value: str
    source: str

StepOutput

A handle on data a step would produce.

Returned by calling a Step. It holds no data and has never touched a file; what it knows is which step would make it, and from what. Once a Pipeline has been built from it, dataset holds the dimensions the engine propagated — the dimensions the output will have, worked out without running anything.

Attributes:

Name	Type	Description
step		The step that would produce it.
arguments		Everything the step was called with, by parameter name.

Source code in python/ligase/graph.py

class StepOutput:
    """A handle on data a step would produce.

    Returned by calling a `Step`. It holds no data and has never touched a
    file; what it knows is which step would make it, and from what. Once a
    `Pipeline` has been built from it, `dataset` holds the dimensions the
    engine propagated — the dimensions the output *will* have, worked out
    without running anything.

    Attributes:
        step: The step that would produce it.
        arguments: Everything the step was called with, by parameter name.
    """

    __slots__ = ("_graph", "_node", "_produced", "arguments", "step")

    def __init__(self, step: Step[Any, Any], arguments: dict[str, object]) -> None:
        """Record a deferred call."""
        self.step = step
        self.arguments = arguments
        self._graph: _core.Graph | None = None
        self._node: int | None = None
        self._produced: str | None = None

    @property
    def kind(self) -> str:
        """The kind of data the step declares it produces."""
        return self.step.output

    @property
    def dataset(self) -> Dataset | None:
        """The propagated output dataset, once a `Pipeline` has been built.

        `None` before that: propagation is the graph's work, and there is no
        graph until a pipeline asks for one.
        """
        if self._graph is None or self._node is None:
            return None
        found = self._graph.dataset(self._node)
        if found is None:
            return None
        return KINDS.get(self.kind, Dataset)(found)

    @property
    def file(self) -> str:
        """Where the output is, once the step has produced it.

        The same name `Dataset.file` has, because a body cannot tell which
        it was handed: a step's input is loaded data for the first step of a
        pipeline and another step's handle after that. A `lg.write` body
        reads it inside the function it hands back, by which time the engine
        has run whatever comes above it:

            return lg.write(lambda out: interpret(v.file, out))

        Raises:
            RuntimeError: Before the step has run, when there is no file to
                name. A shell command should use `{placeholders}` instead,
                which the engine fills in for it.
        """
        if self._produced is None:
            raise RuntimeError(
                f"step `{self.step.name}` has not produced anything yet, so "
                "its output has no file. A shell command names its inputs "
                "with placeholders and lets the engine fill them in; a "
                "`lg.write` body reads `.file` inside the function it "
                "returns, which runs after the steps above it"
            )
        return self._produced

    def inputs(self) -> tuple[tuple[Parameter, object], ...]:
        """The data arguments, paired with the parameters they filled."""
        return tuple(
            (parameter, self.arguments[parameter.name])
            for parameter in self.step.dataset_parameters
        )

    def _attach(self, graph: _core.Graph, node: int) -> None:
        self._graph = graph
        self._node = node

    def __repr__(self) -> str:
        """Identify the handle by the step that would produce it."""
        return f"<StepOutput {self.kind} from step `{self.step.name}`>"

kind property

kind: str

The kind of data the step declares it produces.

dataset property

dataset: Dataset | None

The propagated output dataset, once a Pipeline has been built.

None before that: propagation is the graph's work, and there is no graph until a pipeline asks for one.

file property

file: str

Where the output is, once the step has produced it.

The same name Dataset.file has, because a body cannot tell which it was handed: a step's input is loaded data for the first step of a pipeline and another step's handle after that. A lg.write body reads it inside the function it hands back, by which time the engine has run whatever comes above it:

return lg.write(lambda out: interpret(v.file, out))

Raises:

Type	Description
RuntimeError	Before the step has run, when there is no file to name. A shell command should use {placeholders} instead, which the engine fills in for it.

__init__

__init__(step: Step[Any, Any], arguments: dict[str, object]) -> None

Record a deferred call.

Source code in python/ligase/graph.py

def __init__(self, step: Step[Any, Any], arguments: dict[str, object]) -> None:
    """Record a deferred call."""
    self.step = step
    self.arguments = arguments
    self._graph: _core.Graph | None = None
    self._node: int | None = None
    self._produced: str | None = None

inputs

inputs() -> tuple[tuple[Parameter, object], ...]

The data arguments, paired with the parameters they filled.

Source code in python/ligase/graph.py

def inputs(self) -> tuple[tuple[Parameter, object], ...]:
    """The data arguments, paired with the parameters they filled."""
    return tuple(
        (parameter, self.arguments[parameter.name])
        for parameter in self.step.dataset_parameters
    )

__repr__

__repr__() -> str

Identify the handle by the step that would produce it.

Source code in python/ligase/graph.py

def __repr__(self) -> str:
    """Identify the handle by the step that would produce it."""
    return f"<StepOutput {self.kind} from step `{self.step.name}`>"

Written dataclass

A step whose work is Python, and writes its own output.

Attributes:

Name	Type	Description
function	Callable[[Path], None]	Called with the path to write to, once the engine has decided what that path is.

Source code in python/ligase/recipes.py

@dataclass(frozen=True, slots=True)
class Written:
    """A step whose work is Python, and writes its own output.

    Attributes:
        function: Called with the path to write to, once the engine has
            decided what that path is.
    """

    function: Callable[[Path], None]

Resistance interpretation

ligase

Ligase: type-safe, reproducible computational biology pipelines.

Biological data types carry dimensions (organism, reference build, coordinate system, sample set), and the engine refuses to connect steps whose dimensions do not match -- like units in physics.

>>> import ligase as lg
>>> from ligase.types import Alignment, Reads, Reference
>>> hxb2 = lg.reference("refs/HXB2.fasta", organism="HIV-1", build="HXB2")
>>> samples = lg.samples("data/*.fastq.gz", paired=True)
>>> @lg.step(tool="minimap2")
... def align(reads: Reads, ref: Reference) -> Alignment: ...
>>> pipe = lg.Pipeline(samples.map(lambda s: align(s, hxb2)))
>>> pipe.check()   # type-checks every step; nothing runs
>>> pipe.run()     # parallel, cached, per-sample fan-out

ResistanceReport

Bases: Dataset

An interpreted drug-resistance report.

Source code in python/ligase/types.py

class ResistanceReport(Dataset):
    """An interpreted drug-resistance report."""

    __slots__ = ()

Provenance and auditing

ligase

Ligase: type-safe, reproducible computational biology pipelines.

Biological data types carry dimensions (organism, reference build, coordinate system, sample set), and the engine refuses to connect steps whose dimensions do not match -- like units in physics.

>>> import ligase as lg
>>> from ligase.types import Alignment, Reads, Reference
>>> hxb2 = lg.reference("refs/HXB2.fasta", organism="HIV-1", build="HXB2")
>>> samples = lg.samples("data/*.fastq.gz", paired=True)
>>> @lg.step(tool="minimap2")
... def align(reads: Reads, ref: Reference) -> Alignment: ...
>>> pipe = lg.Pipeline(samples.map(lambda s: align(s, hxb2)))
>>> pipe.check()   # type-checks every step; nothing runs
>>> pipe.run()     # parallel, cached, per-sample fan-out

Methods dataclass

How a set of results was produced.

Attributes:

Name	Type	Description
steps	list[StepMethods]	Each distinct way a step was run, in the order the pipeline runs them.
gaps	list[MethodsGap]	Steps nothing is known about.
references	list[PinnedReference]	Every reference any step was given, pinned, with no duplicates.
citations	list[str]	The reference list, as a methods section prints it.
uncited	list[str]	Tools with nothing to cite — named rather than omitted, so a short reference list cannot pass for a complete one.
engine	str	The version of Ligase that produced the results.
platform	str	The platform they were produced on.
summary	str	One line — 12 steps, all accounted for.
text	str	The paragraph and the reference list, as the command line prints them.
bundle	Bundle | None	The bundle written, where one was asked for.
verification	Verification | None	What checking a bundle found, where one was asked for.

Source code in python/ligase/provenance.py

@dataclass(frozen=True, slots=True)
class Methods:
    """How a set of results was produced.

    Attributes:
        steps: Each distinct way a step was run, in the order the pipeline
            runs them.
        gaps: Steps nothing is known about.
        references: Every reference any step was given, pinned, with no
            duplicates.
        citations: The reference list, as a methods section prints it.
        uncited: Tools with nothing to cite — named rather than omitted, so
            a short reference list cannot pass for a complete one.
        engine: The version of Ligase that produced the results.
        platform: The platform they were produced on.
        summary: One line — `12 steps, all accounted for.`
        text: The paragraph and the reference list, as the command line
            prints them.
        bundle: The bundle written, where one was asked for.
        verification: What checking a bundle found, where one was asked for.
    """

    steps: list[StepMethods] = field(default_factory=list)
    gaps: list[MethodsGap] = field(default_factory=list)
    references: list[PinnedReference] = field(default_factory=list)
    citations: list[str] = field(default_factory=list)
    uncited: list[str] = field(default_factory=list)
    engine: str = ""
    platform: str = ""
    summary: str = ""
    text: str = ""
    bundle: Bundle | None = None
    verification: Verification | None = None

    @property
    def accounted_for(self) -> bool:
        """Whether every step's production is recorded.

        False means the paragraph has a hole in it and says so.
        """
        return not self.gaps

    @property
    def tools(self) -> list[tuple[str, str]]:
        """Every tool any step used, with the version it ran at."""
        seen: list[tuple[str, str]] = []
        for step in self.steps:
            for tool in step.tools:
                if tool not in seen:
                    seen.append(tool)
        return sorted(seen)

    def __repr__(self) -> str:
        """Identify the methods section by its summary."""
        return f"<Methods {self.summary}>"

accounted_for property

accounted_for: bool

Whether every step's production is recorded.

False means the paragraph has a hole in it and says so.

tools property

tools: list[tuple[str, str]]

Every tool any step used, with the version it ran at.

__repr__

__repr__() -> str

Identify the methods section by its summary.

Source code in python/ligase/provenance.py

def __repr__(self) -> str:
    """Identify the methods section by its summary."""
    return f"<Methods {self.summary}>"

StepMethods dataclass

One way a step was run.

One written step becomes one node per sample, and all of them are one entry here. Where two nodes of a step genuinely were run differently they are two entries, because then the difference is the thing worth reporting.

Attributes:

Name	Type	Description
step	str	The step's name, as written.
nodes	int	How many nodes this covers.
scopes	list[str]	What tells them apart, e.g. the sample identifiers.
scope_kind	str | None	What kind of thing that is, e.g. sample.
tools	list[tuple[str, str]]	Each tool and the whole version it printed.
parameters	list[tuple[str, str]]	Each plain parameter and its value, as it was hashed.
command	str | None	The command as written in the pipeline, placeholders and all, or None for a step whose body is Python.

Source code in python/ligase/provenance.py

@dataclass(frozen=True, slots=True)
class StepMethods:
    """One way a step was run.

    One written step becomes one node per sample, and all of them are one
    entry here. Where two nodes of a step genuinely were run differently
    they are two entries, because then the difference is the thing worth
    reporting.

    Attributes:
        step: The step's name, as written.
        nodes: How many nodes this covers.
        scopes: What tells them apart, e.g. the sample identifiers.
        scope_kind: What kind of thing that is, e.g. `sample`.
        tools: Each tool and the whole version it printed.
        parameters: Each plain parameter and its value, as it was hashed.
        command: The command as written in the pipeline, placeholders and
            all, or `None` for a step whose body is Python.
    """

    step: str
    nodes: int
    scopes: list[str]
    scope_kind: str | None
    tools: list[tuple[str, str]]
    parameters: list[tuple[str, str]]
    command: str | None

MethodsGap dataclass

A step nothing is known about.

Its output is in the cache from a run that recorded no provenance, and nothing can reconstruct how it was produced: the command that would have said so is not in the hash, by design. ligase run --force runs it again and records it.

Attributes:

Name	Type	Description
step	str	The step's name.
scopes	list[str]	What tells its nodes apart.
nodes	int	How many of them are unaccounted for.

Source code in python/ligase/provenance.py

@dataclass(frozen=True, slots=True)
class MethodsGap:
    """A step nothing is known about.

    Its output is in the cache from a run that recorded no provenance, and
    nothing can reconstruct how it was produced: the command that would have
    said so is not in the hash, by design. `ligase run --force` runs it
    again and records it.

    Attributes:
        step: The step's name.
        scopes: What tells its nodes apart.
        nodes: How many of them are unaccounted for.
    """

    step: str
    scopes: list[str]
    nodes: int

Bundle dataclass

A reproducibility bundle that was written.

Attributes:

Name	Type	Description
root	str	Where it is.
files	list[str]	Everything in it, relative to the root.
steps	int	How many steps it accounts for.
gaps	int	How many it could not.

Source code in python/ligase/provenance.py

@dataclass(frozen=True, slots=True)
class Bundle:
    """A reproducibility bundle that was written.

    Attributes:
        root: Where it is.
        files: Everything in it, relative to the root.
        steps: How many steps it accounts for.
        gaps: How many it could not.
    """

    root: str
    files: list[str]
    steps: int
    gaps: int

    def __repr__(self) -> str:
        """Identify the bundle by where it is and how much it covers."""
        return f"<Bundle {self.root}, {self.steps} steps, {len(self.files)} files>"

__repr__

__repr__() -> str

Identify the bundle by where it is and how much it covers.

Source code in python/ligase/provenance.py

def __repr__(self) -> str:
    """Identify the bundle by where it is and how much it covers."""
    return f"<Bundle {self.root}, {self.steps} steps, {len(self.files)} files>"

Verification dataclass

What checking a bundle against this machine found.

Attributes:

Name	Type	Description
results	list[ResultCheck]	Every result the bundle recorded.
inputs	list[InputCheck]	Every input it recorded.
reproduced	bool	Whether every recorded result came out the same.
verdict	str	The worst thing found.
summary	str	One line — 12 results: 4 identical, 8 reproduced.
text	str	The whole report as the command line prints it.

Source code in python/ligase/provenance.py

@dataclass(frozen=True, slots=True)
class Verification:
    """What checking a bundle against this machine found.

    Attributes:
        results: Every result the bundle recorded.
        inputs: Every input it recorded.
        reproduced: Whether every recorded result came out the same.
        verdict: The worst thing found.
        summary: One line — `12 results: 4 identical, 8 reproduced.`
        text: The whole report as the command line prints it.
    """

    results: list[ResultCheck] = field(default_factory=list)
    inputs: list[InputCheck] = field(default_factory=list)
    reproduced: bool = False
    verdict: str = MISSING
    summary: str = ""
    text: str = ""

    @property
    def differing(self) -> list[ResultCheck]:
        """The results that did not come out the same."""
        return [check for check in self.results if not check.reproduced]

    @property
    def moved(self) -> list[InputCheck]:
        """The inputs that are not the ones the bundle was built from.

        Worth reading before any `differs` verdict: a changed input explains
        a changed output, and a bundle cannot fetch its own data.
        """
        return [check for check in self.inputs if check.presence != SAME]

    def __repr__(self) -> str:
        """Identify the verification by its summary."""
        return f"<Verification {self.summary}>"

differing property

differing: list[ResultCheck]

The results that did not come out the same.

moved property

moved: list[InputCheck]

The inputs that are not the ones the bundle was built from.

Worth reading before any differs verdict: a changed input explains a changed output, and a bundle cannot fetch its own data.

__repr__

__repr__() -> str

Identify the verification by its summary.

Source code in python/ligase/provenance.py

def __repr__(self) -> str:
    """Identify the verification by its summary."""
    return f"<Verification {self.summary}>"

PinnedReference dataclass

A reference sequence, pinned as firmly as its file allows.

Nothing here is declared or guessed. The accession is the sequence name where the name is one — K03455.1 carries its own version — and the checksum is the MD5 the SAM specification defines for @SQ M5, over the uppercased sequence with whitespace removed.

Attributes:

Name	Type	Description
param	str	The parameter it filled.
path	str	Where it was when the step ran.
build	str | None	The label it was declared under, e.g. HXB2. Display only.
accession	str | None	The accession with its version, where the sequence name is one, and None where it is not — chr1 is a name, not an accession, and a methods section must not cite it as one.
contigs	int	How many sequences it holds.
length	int	Their total length in bases.
checksum	str | None	The MD5 of the sequence, for a single-sequence reference.
fingerprint	str	A short digest of the whole identity, always present.

Source code in python/ligase/provenance.py

@dataclass(frozen=True, slots=True)
class PinnedReference:
    """A reference sequence, pinned as firmly as its file allows.

    Nothing here is declared or guessed. The accession is the sequence name
    where the name is one — `K03455.1` carries its own version — and the
    checksum is the MD5 the SAM specification defines for `@SQ M5`, over the
    uppercased sequence with whitespace removed.

    Attributes:
        param: The parameter it filled.
        path: Where it was when the step ran.
        build: The label it was declared under, e.g. `HXB2`. Display only.
        accession: The accession with its version, where the sequence name
            is one, and `None` where it is not — `chr1` is a name, not an
            accession, and a methods section must not cite it as one.
        contigs: How many sequences it holds.
        length: Their total length in bases.
        checksum: The MD5 of the sequence, for a single-sequence reference.
        fingerprint: A short digest of the whole identity, always present.
    """

    param: str
    path: str
    build: str | None
    accession: str | None
    contigs: int
    length: int
    checksum: str | None
    fingerprint: str

AuditReport dataclass

What an audit found.

Attributes:

Name	Type	Description
steps	list[StepAudit]	Every node audited, in graph order.
failures	list[StepFailure]	Every node that could not be audited because it failed.
verdict	str	The worst verdict any node got — the worst rather than the commonest, since one irreproducible sample in a cohort is an irreproducible step.
summary	str	One line — 4 steps: 1 deterministic, 3 reproducible.
text	str	The whole report as the command line prints it, worded by the engine so that every surface reads the same.

Source code in python/ligase/audit.py

@dataclass(frozen=True, slots=True)
class AuditReport:
    """What an audit found.

    Attributes:
        steps: Every node audited, in graph order.
        failures: Every node that could not be audited because it failed.
        verdict: The worst verdict any node got — the worst rather than the
            commonest, since one irreproducible sample in a cohort is an
            irreproducible step.
        summary: One line — `4 steps: 1 deterministic, 3 reproducible.`
        text: The whole report as the command line prints it, worded by the
            engine so that every surface reads the same.
    """

    steps: list[StepAudit] = field(default_factory=list)
    failures: list[StepFailure] = field(default_factory=list)
    verdict: str = DETERMINISTIC
    summary: str = ""
    text: str = ""

    @property
    def reproducible(self) -> bool:
        """Whether every node audited agreed with itself."""
        return all(step.reproducible for step in self.steps) and not self.failures

    @property
    def unreproducible(self) -> list[StepAudit]:
        """The nodes whose two runs disagreed about the result."""
        return [step for step in self.steps if not step.reproducible]

    def __repr__(self) -> str:
        """Identify the report by its summary."""
        return f"<AuditReport {self.summary}>"

reproducible property

reproducible: bool

Whether every node audited agreed with itself.

unreproducible property

unreproducible: list[StepAudit]

The nodes whose two runs disagreed about the result.

__repr__

__repr__() -> str

Identify the report by its summary.

Source code in python/ligase/audit.py

def __repr__(self) -> str:
    """Identify the report by its summary."""
    return f"<AuditReport {self.summary}>"

StepAudit dataclass

What two runs of one step showed.

Attributes:

Name	Type	Description
step	str	The step's name.
sample	str | None	The sample it was for, or None for a step that spans more than one.
scope	str | None	How a report names this node where its step's name is shared — sample SRR8525904 — or None where the name is already unambiguous.
verdict	str	deterministic, reproducible, or not reproducible.
first	str	The first run's output hash.
second	str	The second run's output hash.
first_canonical	str	What the first run's output means.
second_canonical	str	What the second run's output means.
first_duration_ms	int	How long the first run took.
second_duration_ms	int	How long the second run took.
node	int	Which node of the graph it was.

Source code in python/ligase/audit.py

@dataclass(frozen=True, slots=True)
class StepAudit:
    """What two runs of one step showed.

    Attributes:
        step: The step's name.
        sample: The sample it was for, or `None` for a step that spans more
            than one.
        scope: How a report names this node where its step's name is shared
            — `sample SRR8525904` — or `None` where the name is already
            unambiguous.
        verdict: `deterministic`, `reproducible`, or `not reproducible`.
        first: The first run's output hash.
        second: The second run's output hash.
        first_canonical: What the first run's output means.
        second_canonical: What the second run's output means.
        first_duration_ms: How long the first run took.
        second_duration_ms: How long the second run took.
        node: Which node of the graph it was.
    """

    step: str
    sample: str | None
    scope: str | None
    verdict: str
    first: str
    second: str
    first_canonical: str
    second_canonical: str
    first_duration_ms: int
    second_duration_ms: int
    node: int

    @property
    def reproducible(self) -> bool:
        """Whether the two runs agreed on the result.

        True for a step whose bytes moved and whose meaning did not: that is
        what reproducible means here.
        """
        return self.verdict != NOT_REPRODUCIBLE

reproducible property

reproducible: bool

Whether the two runs agreed on the result.

True for a step whose bytes moved and whose meaning did not: that is what reproducible means here.

Report dataclass

What a check found.

Attributes:

Name	Type	Description
errors	list[CheckError]	One error per failing step, in the order the steps run. A problem that hits four samples is four errors, each naming its own sample.
summary	str	One line summarising the lot — 4 errors in 4 steps. — and the steps it counts are places rather than names, since one written step is many steps under fan-out.
text	str	The whole report as the command line prints it: each distinct problem stated once, with every sample it affects named, and the summary at the end. Worded by the engine so that every surface reports a check the same way.

Source code in python/ligase/graph.py

@dataclass(frozen=True, slots=True)
class Report:
    """What a check found.

    Attributes:
        errors: One error per *failing step*, in the order the steps run. A
            problem that hits four samples is four errors, each naming its
            own sample.
        summary: One line summarising the lot — `4 errors in 4 steps.` — and
            the steps it counts are places rather than names, since one
            written step is many steps under fan-out.
        text: The whole report as the command line prints it: each distinct
            problem stated once, with every sample it affects named, and the
            summary at the end. Worded by the engine so that every surface
            reports a check the same way.
    """

    errors: list[CheckError]
    summary: str
    text: str

RunReport dataclass

Everything a run did.

Attributes:

Name	Type	Description
steps	list[StepRun]	Every step that produced an output, in graph order.
failures	list[StepFailure]	Every step that did not.
summary	str	One line — 12 steps: 8 ran, 4 from cache. — counting places rather than names, since one written step is many steps under fan-out.
text	str	The whole report as the command line prints it, worded by the engine so that every surface reads the same.
hashed	int	Input files that had to be read and hashed.
prefiltered	int	Input files whose hash was already known from their size and mtime, and so were not read at all.
canonicalised	int	Files that had to be parsed to find out what they mean. Like hashed, this is zero on a run that found everything it needed already known.

Source code in python/ligase/run.py

@dataclass(frozen=True, slots=True)
class RunReport:
    """Everything a run did.

    Attributes:
        steps: Every step that produced an output, in graph order.
        failures: Every step that did not.
        summary: One line — `12 steps: 8 ran, 4 from cache.` — counting
            places rather than names, since one written step is many steps
            under fan-out.
        text: The whole report as the command line prints it, worded by the
            engine so that every surface reads the same.
        hashed: Input files that had to be read and hashed.
        prefiltered: Input files whose hash was already known from their
            size and mtime, and so were not read at all.
        canonicalised: Files that had to be parsed to find out what they
            mean. Like `hashed`, this is zero on a run that found everything
            it needed already known.
    """

    steps: list[StepRun] = field(default_factory=list)
    failures: list[StepFailure] = field(default_factory=list)
    summary: str = ""
    text: str = ""
    hashed: int = 0
    prefiltered: int = 0
    canonicalised: int = 0

    @property
    def ran(self) -> list[StepRun]:
        """The steps that actually ran, in graph order."""
        return [step for step in self.steps if step.outcome == "ran"]

    @property
    def cached(self) -> list[StepRun]:
        """The steps that were served from the cache."""
        return [step for step in self.steps if step.outcome == "cached"]

    @property
    def skipped(self) -> list[StepRun]:
        """The steps an upstream rerun turned out not to change."""
        return [step for step in self.steps if step.outcome == "skipped"]

    def outputs(self, step: str) -> list[str]:
        """Where one step's outputs are, in graph order.

        Args:
            step: The step's name. Under fan-out there is one output per
                sample.

        Returns:
            One path per node of that name.
        """
        return [run.output for run in self.steps if run.step == step]

    def __repr__(self) -> str:
        """Identify the report by its summary."""
        return f"<RunReport {self.summary}>"

ran property

ran: list[StepRun]

The steps that actually ran, in graph order.

cached property

cached: list[StepRun]

The steps that were served from the cache.

skipped property

skipped: list[StepRun]

The steps an upstream rerun turned out not to change.

outputs

outputs(step: str) -> list[str]

Where one step's outputs are, in graph order.

Parameters:

Name	Type	Description	Default
step	str	The step's name. Under fan-out there is one output per sample.	required

Returns:

Type	Description
list[str]	One path per node of that name.

Source code in python/ligase/run.py

def outputs(self, step: str) -> list[str]:
    """Where one step's outputs are, in graph order.

    Args:
        step: The step's name. Under fan-out there is one output per
            sample.

    Returns:
        One path per node of that name.
    """
    return [run.output for run in self.steps if run.step == step]

__repr__

__repr__() -> str

Identify the report by its summary.

Source code in python/ligase/run.py

def __repr__(self) -> str:
    """Identify the report by its summary."""
    return f"<RunReport {self.summary}>"

StepRun dataclass

One step that produced an output.

Attributes:

Name	Type	Description
step	str	The step's name.
sample	str | None	The sample it was for, or None for a step that spans more than one — naming such a step after one of its samples would send you to the wrong data.
outcome	str	ran, cached, or skipped. skipped is early cutoff: the step's key was already known although something above it reran, because that rerun produced the same output as before.
output	str	Where the output is, in the content-addressed store.
key	str	The cache key, as hex.
hash	str	The hash of the output's bytes, as hex, which is also its name in the store.
canonical	str	The hash of what the output means, as hex: its records sorted and its volatile headers stripped. This is what the steps below were keyed on, so two runs that agree here skip everything downstream however far their bytes diverged.
duration_ms	int	How long it took, when it ran; zero otherwise.
node	int	Which node of the graph it was.
notes	tuple[str, ...]	What canonicalising the output noticed and its hash cannot say — one sentence each, and almost always none. A note is not a failure: the step ran and its result stands. The only note in v0.1 reports a sample identifier that is a file name rather than a sample, which makes this output's canonical hash depend on where that file was.

Source code in python/ligase/run.py

@dataclass(frozen=True, slots=True)
class StepRun:
    """One step that produced an output.

    Attributes:
        step: The step's name.
        sample: The sample it was for, or `None` for a step that spans more
            than one — naming such a step after one of its samples would
            send you to the wrong data.
        outcome: `ran`, `cached`, or `skipped`. `skipped` is early cutoff:
            the step's key was already known although something above it
            reran, because that rerun produced the same output as before.
        output: Where the output is, in the content-addressed store.
        key: The cache key, as hex.
        hash: The hash of the output's bytes, as hex, which is also its name
            in the store.
        canonical: The hash of what the output *means*, as hex: its records
            sorted and its volatile headers stripped. This is what the steps
            below were keyed on, so two runs that agree here skip everything
            downstream however far their bytes diverged.
        duration_ms: How long it took, when it ran; zero otherwise.
        node: Which node of the graph it was.
        notes: What canonicalising the output noticed and its hash cannot
            say — one sentence each, and almost always none. A note is not
            a failure: the step ran and its result stands. The only note in
            v0.1 reports a sample identifier that is a file name rather than
            a sample, which makes this output's canonical hash depend on
            where that file was.
    """

    step: str
    sample: str | None
    outcome: str
    output: str
    key: str
    hash: str
    canonical: str
    duration_ms: int
    node: int
    notes: tuple[str, ...] = ()

    @property
    def ran(self) -> bool:
        """Whether this step actually ran, rather than being served."""
        return self.outcome == "ran"

ran property

ran: bool

Whether this step actually ran, rather than being served.

Errors

Every failure mode has its own type, so a caller can tell a reference mismatch from a naming mismatch without reading the message.

ligase

Ligase: type-safe, reproducible computational biology pipelines.

Biological data types carry dimensions (organism, reference build, coordinate system, sample set), and the engine refuses to connect steps whose dimensions do not match -- like units in physics.

>>> import ligase as lg
>>> from ligase.types import Alignment, Reads, Reference
>>> hxb2 = lg.reference("refs/HXB2.fasta", organism="HIV-1", build="HXB2")
>>> samples = lg.samples("data/*.fastq.gz", paired=True)
>>> @lg.step(tool="minimap2")
... def align(reads: Reads, ref: Reference) -> Alignment: ...
>>> pipe = lg.Pipeline(samples.map(lambda s: align(s, hxb2)))
>>> pipe.check()   # type-checks every step; nothing runs
>>> pipe.run()     # parallel, cached, per-sample fan-out

LoadError module-attribute

LoadError = _core.LoadError

CheckError

Bases: Exception

A pipeline will not work, found before anything ran.

Attributes:

Name	Type	Description
step		The step the problem was found in.
dimension		The name of the dimension involved, where one is. None for a StepTypeError, which blames no dimension.
scope		What distinguishes that step from the others of the same name, as a (kind, value) pair — ("sample", "SRR32217714") under per-sample fan-out. None for a step that stands for no single sample, such as one that spans a whole cohort. Collecting these is how you ask which samples a problem affects: affected = {e.scope[1] for e in pipe.errors() if e.scope}
errors	list[CheckError]	Every problem found in the same pass, this one included. Checking reports all of them at once rather than one per run, because fixing errors one per overnight run is the problem Ligase exists to remove.

Source code in python/ligase/errors.py

class CheckError(Exception):
    """A pipeline will not work, found before anything ran.

    Attributes:
        step: The step the problem was found in.
        dimension: The name of the dimension involved, where one is. `None`
            for a `StepTypeError`, which blames no dimension.
        scope: What distinguishes that step from the others of the same name,
            as a `(kind, value)` pair — `("sample", "SRR32217714")` under
            per-sample fan-out. `None` for a step that stands for no single
            sample, such as one that spans a whole cohort. Collecting these
            is how you ask which samples a problem affects:

                affected = {e.scope[1] for e in pipe.errors() if e.scope}

        errors: Every problem found in the same pass, this one included.
            Checking reports all of them at once rather than one per run,
            because fixing errors one per overnight run is the problem
            Ligase exists to remove.
    """

    def __init__(
        self,
        message: str,
        step: str,
        dimension: str | None,
        scope: tuple[str, str] | None = None,
    ) -> None:
        """Record a problem, its step, its dimension and which step it was."""
        super().__init__(message)
        self.step = step
        self.dimension = dimension
        self.scope = scope
        self.errors: list[CheckError] = []

__init__

__init__(message: str, step: str, dimension: str | None, scope: tuple[str, str] | None = None) -> None

Record a problem, its step, its dimension and which step it was.

Source code in python/ligase/errors.py

def __init__(
    self,
    message: str,
    step: str,
    dimension: str | None,
    scope: tuple[str, str] | None = None,
) -> None:
    """Record a problem, its step, its dimension and which step it was."""
    super().__init__(message)
    self.step = step
    self.dimension = dimension
    self.scope = scope
    self.errors: list[CheckError] = []

DimensionError

Bases: CheckError

Two inputs to a step disagree about a dimension.

The message names the step, both conflicting values, where each value came from, and asks a leading question. The provenance is the part that makes it actionable: knowing two references disagree sends you hunting, while knowing one came from a BAM @SQ header and the other from the FASTA you passed tells you what to change.

Source code in python/ligase/errors.py

class DimensionError(CheckError):
    """Two inputs to a step disagree about a dimension.

    The message names the step, both conflicting values, where each value
    came from, and asks a leading question. The provenance is the part that
    makes it actionable: knowing two references disagree sends you hunting,
    while knowing one came from a BAM `@SQ` header and the other from the
    FASTA you passed tells you what to change.
    """

ContigNamingError

Bases: DimensionError

The same contigs under different names, such as chr1 against 1.

Distinct from a plain DimensionError because the fix is distinct: a renaming step, not a different reference. Still a DimensionError, so code that wants to catch any dimension problem can.

Source code in python/ligase/errors.py

class ContigNamingError(DimensionError):
    """The same contigs under different names, such as `chr1` against `1`.

    Distinct from a plain `DimensionError` because the fix is distinct: a
    renaming step, not a different reference. Still a `DimensionError`, so
    code that wants to catch any dimension problem can.
    """

ContigOrderError

Bases: DimensionError

The same contigs in a different order.

htslib matches contigs by name, so bcftools does not mind. GATK requires sequence dictionaries in matching order and will reject this.

Source code in python/ligase/errors.py

class ContigOrderError(DimensionError):
    """The same contigs in a different order.

    htslib matches contigs by name, so `bcftools` does not mind. GATK
    requires sequence dictionaries in matching order and will reject this.
    """

UndeclaredDimensionError

Bases: DimensionError

A dimension some step requires is unknown and must be declared.

Raised only because a step actually joins on it: an unknown dimension nothing downstream requires is allowed to stay unknown. The message names the data that has to declare it and the route by which it reaches the step that needs it, which for a value propagated through the graph are two different places.

Source code in python/ligase/errors.py

class UndeclaredDimensionError(DimensionError):
    """A dimension some step requires is unknown and must be declared.

    Raised only because a step actually joins on it: an unknown dimension
    nothing downstream requires is allowed to stay unknown. The message names
    the data that has to declare it and the route by which it reaches the
    step that needs it, which for a value propagated through the graph are
    two different places.
    """

StepTypeError

Bases: CheckError

A step was handed a kind of data it does not declare.

Not a DimensionError: nothing disagrees about the data, the pipeline is simply wired up wrong. Reported instead of that step's dimension errors rather than alongside them, because comparing the dimensions of bindings the step never declared would describe a pipeline you did not write.

Source code in python/ligase/errors.py

class StepTypeError(CheckError):
    """A step was handed a kind of data it does not declare.

    Not a `DimensionError`: nothing disagrees about the data, the pipeline is
    simply wired up wrong. Reported instead of that step's dimension errors
    rather than alongside them, because comparing the dimensions of bindings
    the step never declared would describe a pipeline you did not write.
    """

StepDefinitionError

Bases: TypeError

A decorated function cannot be used as a step.

Raised by @lg.step as the module is imported, not at check time: a missing annotation is a mistake in the pipeline's source, and the sooner it is said the less there is to read. A TypeError, because that is what a badly typed definition is.

Source code in python/ligase/errors.py

class StepDefinitionError(TypeError):
    """A decorated function cannot be used as a step.

    Raised by `@lg.step` as the module is imported, not at check time: a
    missing annotation is a mistake in the pipeline's source, and the sooner
    it is said the less there is to read. A `TypeError`, because that is what
    a badly typed definition is.
    """

RunError

Bases: Exception

A step failed.

Attributes:

Name	Type	Description
report		The whole run, including everything that did succeed — which for a cohort is usually most of it.
failures		The steps that failed, as a shortcut.

Source code in python/ligase/run.py

class RunError(Exception):
    """A step failed.

    Attributes:
        report: The whole run, including everything that did succeed —
            which for a cohort is usually most of it.
        failures: The steps that failed, as a shortcut.
    """

    def __init__(self, report: RunReport) -> None:
        """Carry the whole report, not just the first failure."""
        self.report = report
        self.failures = report.failures
        first = report.failures[0] if report.failures else None
        where = "" if first is None else f"step `{first.step}`: {first.message}"
        super().__init__(f"{report.summary} {where}".strip())

__init__

__init__(report: RunReport) -> None

Carry the whole report, not just the first failure.

Source code in python/ligase/run.py

def __init__(self, report: RunReport) -> None:
    """Carry the whole report, not just the first failure."""
    self.report = report
    self.failures = report.failures
    first = report.failures[0] if report.failures else None
    where = "" if first is None else f"step `{first.step}`: {first.message}"
    super().__init__(f"{report.summary} {where}".strip())

StepFailure dataclass

One step that did not produce anything.

A step whose input failed is in neither list: one cause, one failure.

Attributes:

Name	Type	Description
step	str	The step's name.
sample	str | None	The sample it was for, where it stands for one.
message	str	What went wrong, in the tool's own words where there are any.
node	int	Which node of the graph it was.

Source code in python/ligase/run.py

@dataclass(frozen=True, slots=True)
class StepFailure:
    """One step that did not produce anything.

    A step whose input failed is in neither list: one cause, one failure.

    Attributes:
        step: The step's name.
        sample: The sample it was for, where it stands for one.
        message: What went wrong, in the tool's own words where there are
            any.
        node: Which node of the graph it was.
    """

    step: str
    sample: str | None
    message: str
    node: int

InputCheck dataclass

One input file, as a bundle recorded it and as it is here.

Attributes:

Name	Type	Description
step	str	The step it fed.
param	str	The parameter it filled.
path	str	Where the bundle said it was.
presence	str	same, changed or absent, by content rather than by name — which is the only way to tell the file a bundle was built from apart from a different file with the same name.

Source code in python/ligase/provenance.py

@dataclass(frozen=True, slots=True)
class InputCheck:
    """One input file, as a bundle recorded it and as it is here.

    Attributes:
        step: The step it fed.
        param: The parameter it filled.
        path: Where the bundle said it was.
        presence: `same`, `changed` or `absent`, by content rather than by
            name — which is the only way to tell the file a bundle was built
            from apart from a different file with the same name.
    """

    step: str
    param: str
    path: str
    presence: str

ResultCheck dataclass

One result, as a bundle recorded it and as this machine produced it.

Attributes:

Name	Type	Description
step	str	The step's name.
scope	str | None	What tells this node apart, where anything does.
verdict	str	identical, reproduced, differs or missing.
expected	str	The canonical hash the bundle recorded.
found	str | None	The canonical hash produced here, or None where this pipeline produced nothing for it.
same_key	bool	Whether the cache key matched too. False with a verdict of reproduced means something in the environment moved and the result did not, which is the interesting outcome.

Source code in python/ligase/provenance.py

@dataclass(frozen=True, slots=True)
class ResultCheck:
    """One result, as a bundle recorded it and as this machine produced it.

    Attributes:
        step: The step's name.
        scope: What tells this node apart, where anything does.
        verdict: `identical`, `reproduced`, `differs` or `missing`.
        expected: The canonical hash the bundle recorded.
        found: The canonical hash produced here, or `None` where this
            pipeline produced nothing for it.
        same_key: Whether the cache key matched too. False with a verdict of
            `reproduced` means something in the environment moved and the
            result did not, which is the interesting outcome.
    """

    step: str
    scope: str | None
    verdict: str
    expected: str
    found: str | None
    same_key: bool

    @property
    def reproduced(self) -> bool:
        """Whether the result came out the same, by either route."""
        return self.verdict in (IDENTICAL, REPRODUCED)

reproduced property

reproduced: bool

Whether the result came out the same, by either route.

Version

ligase

Ligase: type-safe, reproducible computational biology pipelines.

Biological data types carry dimensions (organism, reference build, coordinate system, sample set), and the engine refuses to connect steps whose dimensions do not match -- like units in physics.

>>> import ligase as lg
>>> from ligase.types import Alignment, Reads, Reference
>>> hxb2 = lg.reference("refs/HXB2.fasta", organism="HIV-1", build="HXB2")
>>> samples = lg.samples("data/*.fastq.gz", paired=True)
>>> @lg.step(tool="minimap2")
... def align(reads: Reads, ref: Reference) -> Alignment: ...
>>> pipe = lg.Pipeline(samples.map(lambda s: align(s, hxb2)))
>>> pipe.check()   # type-checks every step; nothing runs
>>> pipe.run()     # parallel, cached, per-sample fan-out

version

version() -> str

Return the version of the compiled Ligase core engine.

Source code in python/ligase/__init__.py

def version() -> str:
    """Return the version of the compiled Ligase core engine."""
    return _core.version()