fasta_lake.quantification

fasta_lake.quantification

Selectable quantification of an existing, fixed study dictionary.

The original study outputs remain the source of assignments and accepted features. A new destination contains the selected matrix and method-specific diagnostics.

Optional backend: MannLabs directLFQ, https://github.com/MannLabs/directlfq. Ammar et al. (2023), Accurate Label-Free Quantification by directLFQ to Compare Unlimited Numbers of Proteomes. This adapter builds and checks the fixed-group ion input; directLFQ performs the quantification.

stamp

stamp(path)

Return basename, byte size and a streamed SHA256 for a pathlib.Path input.

Source code in fasta_lake/quantification.py

def stamp(path):
    """Return basename, byte size and a streamed SHA256 for a pathlib.Path input."""
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for block in iter(lambda: stream.read(8 * 1024 * 1024), b""):
            digest.update(block)
    return {"file": path.name, "sha256": digest.hexdigest(), "bytes": path.stat().st_size}

require_directlfq

require_directlfq()

Fail before running a workflow if the optional backend is unavailable.

Source code in fasta_lake/quantification.py

def require_directlfq():
    """Fail before running a workflow if the optional backend is unavailable."""
    try:
        version = importlib.metadata.version("directlfq")
        if version != DIRECTLFQ_VERSION:
            raise RuntimeError(f"Expected directlfq {DIRECTLFQ_VERSION}; found {version}")
        from directlfq import lfq_manager  # noqa: F401
    except ImportError as error:
        raise RuntimeError(
            "Install the optional backend: pip install 'fasta-lake[directlfq]'"
        ) from error
    return version

quantify_study

quantify_study(groups, output, *, method='sum', threads=1)

Quantify an existing study dictionary without assigning peptides again.

Parameters:

Name	Type	Description	Default
groups	path - like	Completed sequence-checked grouping bundle from group_searches.	required
output	path - like	New directory for the selected-method matrix, coverage and summary.	required
method	('sum', 'directlfq')	Sum copies the original matrix unchanged. directLFQ estimates quantities from assigned ions using the required backend version.	"sum"
threads	int	Positive worker count passed to directLFQ.	1

Returns:

Type	Description
dict	Method, acquisition roster, input/output hashes and coverage information also written to summary.json.

Raises:

Type	Description
FileExistsError	If the output already exists.
ValueError	For invalid settings, inconsistent group/feature data or changed inputs.
RuntimeError	If the requested directLFQ backend is unavailable or has the wrong version.

Notes

The fixed peptide-to-group dictionary and acquisition order are preserved. Unquantified output cells remain blank; no missing values are imputed.

Source code in fasta_lake/quantification.py

def quantify_study(groups, output, *, method="sum", threads=1):
    """Quantify an existing study dictionary without assigning peptides again.

    Parameters
    ----------
    groups : path-like
        Completed sequence-checked grouping bundle from group_searches.
    output : path-like
        New directory for the selected-method matrix, coverage and summary.
    method : {"sum", "directlfq"}
        Sum copies the original matrix unchanged. directLFQ estimates quantities
        from assigned ions using the required backend version.
    threads : int
        Positive worker count passed to directLFQ.

    Returns
    -------
    dict
        Method, acquisition roster, input/output hashes and coverage information
        also written to summary.json.

    Raises
    ------
    FileExistsError
        If the output already exists.
    ValueError
        For invalid settings, inconsistent group/feature data or changed inputs.
    RuntimeError
        If the requested directLFQ backend is unavailable or has the wrong version.

    Notes
    -----
    The fixed peptide-to-group dictionary and acquisition order are preserved.
    Unquantified output cells remain blank; no missing values are imputed.
    """
    if method not in {"sum", "directlfq"}:
        raise ValueError("method must be sum or directlfq")
    if not isinstance(threads, int) or threads < 1:
        raise ValueError("threads must be a positive integer")
    source, out = Path(groups).resolve(), Path(output).absolute()
    if out.exists() or out.is_symlink():
        raise FileExistsError(f"Output exists: {out}; choose a new directory")
    out = out.resolve()
    if source == out or out in source.parents:
        raise ValueError("Output must not contain the source study directory")
    version = require_directlfq() if method == "directlfq" else None
    names = [
        "summary.json",
        "study_group_matrix.tsv",
        "peptide_to_study_group.tsv",
        "feature_assignments.tsv.gz",
        "acquisition_summary.tsv",
        "input_manifest.json",
    ]
    names += [
        n for n in ["peptide_evidence.tsv.gz", "peptide_ambiguity.tsv"] if (source / n).is_file()
    ]
    inputs = [stamp(source / name) for name in names]
    summary = json.loads((source / "summary.json").read_text())
    if summary.get("schema_version") != 1 or not summary.get(
        "representative_sequence_compatibility_checked"
    ):
        raise ValueError("Expected a completed, sequence-checked study grouping bundle")
    samples, counts = _baseline(source / "study_group_matrix.tsv")
    if samples != summary.get("acquisition_names"):
        raise ValueError("Study matrix acquisition roster differs from summary")
    out.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix=f".{out.name}-", dir=out.parent) as directory:
        temp = Path(directory)
        if method == "sum":
            shutil.copyfile(source / "study_group_matrix.tsv", temp / "study_group_matrix.tsv")
            details = {"sample_normalization": False, "quantified_groups": counts}
        else:
            details = _directlfq(source, temp, samples, threads)
        with (temp / "coverage.tsv").open("w", newline="") as stream:
            writer = csv.writer(stream, delimiter="\t")
            writer.writerow(["sample", "sum_groups", "selected_method_groups", "groups_lost"])
            writer.writerows(
                (s, a, b, a - b)
                for s, a, b in zip(samples, counts, details.pop("quantified_groups"))
            )
        # Detect changes to source artifacts during export/estimation.
        if inputs != [stamp(source / name) for name in names]:
            raise ValueError("Study inputs changed during quantification")
        record = {
            "schema_version": 1,
            "method": method,
            "directlfq_version": version,
            "acquisition_names": samples,
            "source_study_directory": str(source),
            "matrix": "study_group_matrix.tsv",
            "inputs": inputs,
            "implementation": stamp(Path(__file__)),
            **details,
            "missing_values": "blank in selected matrix; zero in directLFQ tables; no imputation",
            "scope": "Relative study-group quantities; no new identification or group FDR. "
            "Representative labels do not establish unique protein or organism origin.",
            "outputs": [stamp(p) for p in sorted(temp.iterdir())],
        }
        (temp / "summary.json").write_text(json.dumps(record, indent=2) + "\n")
        out.mkdir(exist_ok=False)
        for path in sorted(temp.iterdir(), key=lambda p: (p.name == "summary.json", p.name)):
            path.rename(out / path.name)
    return record