API reference¶

def download(
    self,
    output_dir: str | Path = ".",
    *,
    ncbi_parallel: int = DEFAULT_NCBI_PARALLEL,
    cores: int | None = None,
    threads_per_run: int = DEFAULT_THREADS_PER_RUN,
    select_srx: Iterable[str] | None = None,
    original_srx: Iterable[str] | None = None,
    max_size: str | None = None,
    retries: int | None = None,
    backoff: float | None = None,
    keep_sra: bool = False,
    prefetch_only: bool = False,
    verbose: bool = True,
) -> dict[str, bool]:
    """Download FASTQ for every run of every SRA experiment in this record.

    Lays the data out as ``<output_dir>/<accession>/<SRX>/<SRR>*.fastq.gz`` —
    one directory per experiment under a directory named for this record — via
    a Snakemake DAG (``prefetch → extract → compress``). Runs already marked
    done (a ``.<SRR>.success`` flag) are skipped, so this is safe to rerun on
    an unstable connection.

    Concurrency is resource-aware: at most ``ncbi_parallel`` runs ``prefetch``
    from NCBI at once (keeping the link busy without tripping throttling), while
    ``cores`` bounds total CPU across the local extraction/compression steps —
    so those two stages overlap instead of blocking each other.

    Pass ``select_srx`` to download only some of the record's experiments (see
    below).

    Parameters
    ----------
    output_dir : str or Path, default "."
        Parent directory; a ``<output_dir>/<accession>`` subtree is created for
        this record.
    ncbi_parallel : int, default :data:`DEFAULT_NCBI_PARALLEL`
        Maximum runs to ``prefetch`` (the NCBI network step) concurrently. Keep
        this modest to avoid NCBI throttling; it is independent of ``cores``.
    cores : int, optional
        Total CPU cores the DAG may use across all steps (defaults to the
        machine's CPU count). On an HPC allocation, pass the allotted cores
        (e.g. ``$SLURM_CPUS_ON_NODE``).
    threads_per_run : int, default :data:`DEFAULT_THREADS_PER_RUN`
        Threads handed to ``fasterq-dump``/``pigz`` for each run.
    select_srx : iterable of str, optional
        A whitelist of SRA experiment accessions (``SRX…``). When given, only
        the record's experiments whose accession is listed are downloaded and
        every other experiment is skipped; ``None`` (the default) downloads
        them all. Entries are matched case-insensitively and every one must be
        a well-formed ``SRX`` accession — anything else raises
        :class:`~labdata.exceptions.AccessionError`.
    original_srx : iterable of str, optional
        A set of SRA experiment accessions (``SRX…``) to fetch in *original
        format* instead of the default sra-tools path: every run of a listed
        experiment has its submitter-uploaded files downloaded as-is (no
        ``fasterq-dump``/gzip), because its ``.sra`` is missing or useless
        (e.g. a 10X run whose ``.sra`` stored only one read). This selects the
        *mode*; it does not by itself widen the selection — an experiment must
        still survive ``select_srx`` to be downloaded at all. Each entry must
        be a well-formed ``SRX`` accession or :class:`~labdata.exceptions.AccessionError`
        is raised.
    max_size : str, optional
        Passed to ``prefetch --max-size`` (defaults to :data:`DEFAULT_MAX_SIZE`);
        raise it for runs that exceed ``prefetch``'s 20G default.
    retries : int, optional
        Attempts for the resumable ``prefetch`` network step per run (defaults
        to :data:`DEFAULT_RETRIES`).
    backoff : float, optional
        Base seconds for the linear backoff between ``prefetch`` retries
        (defaults to :data:`DEFAULT_BACKOFF`).
    keep_sra : bool, default False
        Keep each run's downloaded ``.sra`` next to its FASTQ (as
        ``<accession>/<SRX>/<SRR>.sra``) instead of deleting it after extraction.
    prefetch_only : bool, default False
        Only ``prefetch`` each run's ``.sra`` (into ``<SRX>/<SRR>/``), skipping
        ``fasterq-dump``, gzip, and cleanup. A later full download resumes from
        extraction without re-fetching.
    verbose : bool, default True
        Print a download plan up front and one line per finished run to
        ``stderr``. Set ``False`` to download silently.

    Returns
    -------
    dict[str, bool]
        Maps each run accession to whether its data is present on completion.

    Examples
    --------
    >>> Series("GSE229022").download("./fastq", ncbi_parallel=3, cores=16)  # doctest: +SKIP
    {'SRR24084454': True, 'SRR24084455': True, ...}
    >>> BioProject("PRJNA1027859").download("./fastq", ncbi_parallel=3)  # doctest: +SKIP
    {'SRR26452081': True, 'SRR26452082': True, ...}
    >>> Series("GSE229022").download("./fastq", select_srx=["SRX19885398"])  # doctest: +SKIP
    {'SRR24084454': True}
    """
    # ``None`` overrides fall through to the sra-tools defaults.
    overrides = {
        key: value
        for key, value in {"max_size": max_size, "retries": retries, "backoff": backoff}.items()
        if value is not None
    }
    # Original-format experiments select the download *mode*, not the selection.
    original_set = _validate_srx_whitelist(original_srx, field="original_srx")
    base = Path(output_dir) / self.accession
    tasks: list[tuple[Run, Path]] = []
    original: set[str] = set()
    for experiment in self._selected_experiments(select_srx):
        srx_dir = base / experiment.accession
        srx_dir.mkdir(parents=True, exist_ok=True)
        is_original = experiment.accession in original_set
        for run in experiment.runs:
            tasks.append((run, srx_dir))
            if is_original:
                original.add(run.accession)
    return _run_plan(
        self.accession,
        tasks,
        output_root=base,
        verbose=verbose,
        ncbi_parallel=ncbi_parallel,
        cores=cores,
        threads_per_run=threads_per_run,
        keep_sra=keep_sra,
        prefetch_only=prefetch_only,
        original=original,
        **overrides,
    )

def to_bytes(self) -> bytes:
    """Return this record as canonical 4-line FASTQ bytes (trailing newline)."""
    return b"\n".join((self.header, self.seq, self.plus, self.qual)) + b"\n"

def read_indexes(self) -> list[int]:
    """Return the within-spot read indexes present, ascending."""
    return sorted(self.reads)

def to_fastq_bytes(self, index: int) -> bytes:
    """Return read ``index``'s records concatenated as FASTQ bytes."""
    return b"".join(record.to_bytes() for record in self.reads[index])

def content_hash(self, index: int) -> str:
    """Return the sha256 of read ``index``'s FASTQ bytes (a reproducible slice id)."""
    return hashlib.sha256(self.to_fastq_bytes(index)).hexdigest()

def download(
    self,
    output_dir: str | Path = ".",
    *,
    ncbi_parallel: int = DEFAULT_NCBI_PARALLEL,
    cores: int | None = None,
    threads_per_run: int = DEFAULT_THREADS_PER_RUN,
    select_srx: Iterable[str] | None = None,
    original_srx: Iterable[str] | None = None,
    max_size: str | None = None,
    retries: int | None = None,
    backoff: float | None = None,
    keep_sra: bool = False,
    prefetch_only: bool = False,
    verbose: bool = True,
) -> dict[str, bool]:
    """Download FASTQ for every run of every SRA experiment in this record.

    Lays the data out as ``<output_dir>/<accession>/<SRX>/<SRR>*.fastq.gz`` —
    one directory per experiment under a directory named for this record — via
    a Snakemake DAG (``prefetch → extract → compress``). Runs already marked
    done (a ``.<SRR>.success`` flag) are skipped, so this is safe to rerun on
    an unstable connection.

    Concurrency is resource-aware: at most ``ncbi_parallel`` runs ``prefetch``
    from NCBI at once (keeping the link busy without tripping throttling), while
    ``cores`` bounds total CPU across the local extraction/compression steps —
    so those two stages overlap instead of blocking each other.

    Pass ``select_srx`` to download only some of the record's experiments (see
    below).

    Parameters
    ----------
    output_dir : str or Path, default "."
        Parent directory; a ``<output_dir>/<accession>`` subtree is created for
        this record.
    ncbi_parallel : int, default :data:`DEFAULT_NCBI_PARALLEL`
        Maximum runs to ``prefetch`` (the NCBI network step) concurrently. Keep
        this modest to avoid NCBI throttling; it is independent of ``cores``.
    cores : int, optional
        Total CPU cores the DAG may use across all steps (defaults to the
        machine's CPU count). On an HPC allocation, pass the allotted cores
        (e.g. ``$SLURM_CPUS_ON_NODE``).
    threads_per_run : int, default :data:`DEFAULT_THREADS_PER_RUN`
        Threads handed to ``fasterq-dump``/``pigz`` for each run.
    select_srx : iterable of str, optional
        A whitelist of SRA experiment accessions (``SRX…``). When given, only
        the record's experiments whose accession is listed are downloaded and
        every other experiment is skipped; ``None`` (the default) downloads
        them all. Entries are matched case-insensitively and every one must be
        a well-formed ``SRX`` accession — anything else raises
        :class:`~labdata.exceptions.AccessionError`.
    original_srx : iterable of str, optional
        A set of SRA experiment accessions (``SRX…``) to fetch in *original
        format* instead of the default sra-tools path: every run of a listed
        experiment has its submitter-uploaded files downloaded as-is (no
        ``fasterq-dump``/gzip), because its ``.sra`` is missing or useless
        (e.g. a 10X run whose ``.sra`` stored only one read). This selects the
        *mode*; it does not by itself widen the selection — an experiment must
        still survive ``select_srx`` to be downloaded at all. Each entry must
        be a well-formed ``SRX`` accession or :class:`~labdata.exceptions.AccessionError`
        is raised.
    max_size : str, optional
        Passed to ``prefetch --max-size`` (defaults to :data:`DEFAULT_MAX_SIZE`);
        raise it for runs that exceed ``prefetch``'s 20G default.
    retries : int, optional
        Attempts for the resumable ``prefetch`` network step per run (defaults
        to :data:`DEFAULT_RETRIES`).
    backoff : float, optional
        Base seconds for the linear backoff between ``prefetch`` retries
        (defaults to :data:`DEFAULT_BACKOFF`).
    keep_sra : bool, default False
        Keep each run's downloaded ``.sra`` next to its FASTQ (as
        ``<accession>/<SRX>/<SRR>.sra``) instead of deleting it after extraction.
    prefetch_only : bool, default False
        Only ``prefetch`` each run's ``.sra`` (into ``<SRX>/<SRR>/``), skipping
        ``fasterq-dump``, gzip, and cleanup. A later full download resumes from
        extraction without re-fetching.
    verbose : bool, default True
        Print a download plan up front and one line per finished run to
        ``stderr``. Set ``False`` to download silently.

    Returns
    -------
    dict[str, bool]
        Maps each run accession to whether its data is present on completion.

    Examples
    --------
    >>> Series("GSE229022").download("./fastq", ncbi_parallel=3, cores=16)  # doctest: +SKIP
    {'SRR24084454': True, 'SRR24084455': True, ...}
    >>> BioProject("PRJNA1027859").download("./fastq", ncbi_parallel=3)  # doctest: +SKIP
    {'SRR26452081': True, 'SRR26452082': True, ...}
    >>> Series("GSE229022").download("./fastq", select_srx=["SRX19885398"])  # doctest: +SKIP
    {'SRR24084454': True}
    """
    # ``None`` overrides fall through to the sra-tools defaults.
    overrides = {
        key: value
        for key, value in {"max_size": max_size, "retries": retries, "backoff": backoff}.items()
        if value is not None
    }
    # Original-format experiments select the download *mode*, not the selection.
    original_set = _validate_srx_whitelist(original_srx, field="original_srx")
    base = Path(output_dir) / self.accession
    tasks: list[tuple[Run, Path]] = []
    original: set[str] = set()
    for experiment in self._selected_experiments(select_srx):
        srx_dir = base / experiment.accession
        srx_dir.mkdir(parents=True, exist_ok=True)
        is_original = experiment.accession in original_set
        for run in experiment.runs:
            tasks.append((run, srx_dir))
            if is_original:
                original.add(run.accession)
    return _run_plan(
        self.accession,
        tasks,
        output_root=base,
        verbose=verbose,
        ncbi_parallel=ncbi_parallel,
        cores=cores,
        threads_per_run=threads_per_run,
        keep_sra=keep_sra,
        prefetch_only=prefetch_only,
        original=original,
        **overrides,
    )

def make_sra_run_table(self) -> pandas.DataFrame:
    """Build a tidy, run-level (``SRR``) table for this Series, like SRA Run Selector.

    Resolves the Series' SRA experiments and fetches NCBI's ``runinfo`` report
    in one batched EFetch per chunk, returning one row per sequencing run. The
    columns are SRA's ``runinfo`` fields (``Run``, ``BioSample``, ``Experiment``,
    ``LibraryName``, ``Platform``, …) with a few renamed to the Run-Selector
    names (``AvgSpotLen``, ``Bases``, ``Instrument``). The reported size is
    ``size_MB`` — E-utilities ``runinfo`` does not expose an exact byte count.

    A ``ReadStructure`` column (``{L1}+{L2}+…``, e.g. ``"28+94"``) is added
    from each run's per-read ``<Statistics>``, read from the full SRA XML with
    a second batched EFetch per chunk (see :attr:`Run.read_structure`); it is
    empty for runs SRA reports no read statistics for.

    Returns
    -------
    pandas.DataFrame
        One row per run, key columns ordered first. Empty when the Series has
        no linked SRA runs.

    Examples
    --------
    >>> table = Series("GSE229022").make_sra_run_table()  # doctest: +SKIP
    >>> table[["Run", "BioSample", "Instrument"]].head()  # doctest: +SKIP
    """
    import pandas

    sra_uids = self.client.elink(dbfrom="gds", db="sra", uid=self.uid)
    frames: list[pandas.DataFrame] = []
    structures: dict[str, str] = {}
    for chunk in _chunked(sra_uids, _RUNINFO_CHUNK):
        text = self.client.efetch(db="sra", ids=chunk, rettype="runinfo", retmode="text")
        if text.strip():
            frames.append(pandas.read_csv(io.StringIO(text)))
        xml = self.client.efetch(db="sra", ids=chunk, rettype="full", retmode="xml")
        for accession, block in _run_blocks(xml).items():
            structures[accession] = _read_structure(_read_lengths(block))
    if not frames:
        return pandas.DataFrame()
    table = pandas.concat(frames, ignore_index=True).rename(columns=_RUNINFO_RENAME)
    table["ReadStructure"] = [structures.get(run, "") for run in table["Run"]]
    leading = [column for column in _RUN_TABLE_COLUMNS if column in table.columns]
    remaining = [column for column in table.columns if column not in leading]
    return table.reindex(columns=leading + remaining)

def __repr__(self) -> str:
    """Return an unambiguous representation."""
    return f"{type(self).__name__}({self.experiment.accession!r}, {str(self.output_dir)!r})"

def download(
    self,
    *,
    ncbi_parallel: int = DEFAULT_NCBI_PARALLEL,
    cores: int | None = None,
    threads_per_run: int = DEFAULT_THREADS_PER_RUN,
    max_size: str = DEFAULT_MAX_SIZE,
    retries: int = DEFAULT_RETRIES,
    backoff: float = DEFAULT_BACKOFF,
    keep_sra: bool = False,
    prefetch_only: bool = False,
    original: bool = False,
    verbose: bool = True,
) -> dict[str, bool]:
    """Download every run of the experiment as a resource-aware Snakemake DAG.

    Creates :attr:`srx_dir`, then downloads each run's gzipped FASTQ into it.
    Runs with an existing ``.<SRR>.done`` marker are skipped; a run that fails
    is recorded as ``False`` without aborting the others.

    Pass ``original=True`` to fetch this experiment's *original-format* files
    instead — each run's submitter-uploaded files are downloaded as-is into
    ``<SRX>/<SRR>/`` (no ``fasterq-dump``/gzip), for runs whose ``.sra`` is
    missing or useless.

    Parameters
    ----------
    ncbi_parallel : int, default :data:`DEFAULT_NCBI_PARALLEL`
        Maximum runs to ``prefetch`` (the NCBI network step) concurrently;
        independent of ``cores``. Keep it modest to avoid NCBI throttling.
    cores : int, optional
        Total CPU cores the DAG may use across all steps (defaults to the
        machine's CPU count).
    threads_per_run : int, default :data:`DEFAULT_THREADS_PER_RUN`
        Threads handed to ``fasterq-dump``/``pigz`` for each run.
    max_size : str, default :data:`DEFAULT_MAX_SIZE`
        Passed to ``prefetch --max-size``; raise it for large runs (``prefetch``
        otherwise refuses anything over its 20G default).
    retries : int, default :data:`DEFAULT_RETRIES`
        Attempts for the resumable ``prefetch`` network step per run; a retry
        continues the partial download rather than restarting.
    backoff : float, default :data:`DEFAULT_BACKOFF`
        Base seconds for the linear backoff between ``prefetch`` retries.
    keep_sra : bool, default False
        Keep each run's downloaded ``.sra`` next to its FASTQ (as
        ``<SRX>/<SRR>.sra``) instead of deleting it after extraction.
    prefetch_only : bool, default False
        Only ``prefetch`` each run's ``.sra`` (into ``<SRX>/<SRR>/``), skipping
        ``fasterq-dump``, gzip, and cleanup. A later full download resumes from
        extraction without re-fetching.
    original : bool, default False
        Fetch the experiment's original-format files (download only) instead of
        the ``prefetch → extract → compress`` chain; see above.
    verbose : bool, default True
        Print a download plan up front and one line per finished run to
        ``stderr``. Set ``False`` to download silently.

    Returns
    -------
    dict[str, bool]
        Maps each run accession to whether its data is present on completion.
    """
    srx_dir = self.srx_dir
    srx_dir.mkdir(parents=True, exist_ok=True)
    tasks = [(run, srx_dir) for run in self.experiment.runs]
    original_runs = {run.accession for run in self.experiment.runs} if original else set()
    # output_root is the parent of the SRX dir so the run->srx map (and the
    # Snakefile's ``{srx}`` wildcard) sees the experiment accession as a single
    # path component, matching the whole-record layout.
    return _run_plan(
        self.experiment.accession,
        tasks,
        output_root=self.output_dir,
        verbose=verbose,
        ncbi_parallel=ncbi_parallel,
        cores=cores,
        threads_per_run=threads_per_run,
        max_size=max_size,
        retries=retries,
        backoff=backoff,
        keep_sra=keep_sra,
        prefetch_only=prefetch_only,
        original=original_runs,
    )

def clear_cache(self) -> None:
    """Drop all cached responses (forcing fresh requests on the next call)."""
    self._cache.clear()

def retrieve(self, accession: str) -> list[RemoteFile]:
    """Return every file SDL lists for ``accession``.

    Parameters
    ----------
    accession : str
        An SRA run accession, e.g. ``"SRR20172067"``.

    Returns
    -------
    list of RemoteFile
        Both the normalized SRA archive file(s) and any original/submitted
        files (e.g. a 10X ``possorted_genome_bam``). Cached per client.

    Raises
    ------
    SdlError
        If the request fails, the response is not valid JSON, or SDL reports a
        non-OK status for the accession (e.g. unknown or non-public run).
    """
    cached = self._cache.get(accession)
    if cached is None:
        payload = self._fetch(accession)
        cached = self._cache[accession] = _parse_files(payload, accession)
    return list(cached)

def __repr__(self) -> str:
    """Return the root and the build."""
    return f"Store({str(self.root)!r}, build={self.settings.get('build')!r})"

def open_many(self, accessions: Iterable[str]) -> dict[str, Record | LabdataError]:
    """Open many records: one record lookup and one link lookup for the whole batch.

    Parameters
    ----------
    accessions : iterable of str
        Public accessions of any record type.

    Returns
    -------
    dict
        Each accession, in input order, mapped to its ``Record`` or to the error that
        stopped it (``AccessionError``, ``NotMirrored`` or ``NotFound``).
    """
    order = [_norm(a) for a in accessions]
    out: dict[str, Record | LabdataError] = {}
    valid: list[str] = []
    for accession in order:
        try:
            kind = kind_of(accession)
            if kind == PAPER:
                raise AccessionError(
                    f"{accession} is a paper, not a record; follow 'PubMed' or 'DOI'"
                )
            if kind == ANNOTATION:
                raise AccessionError(
                    f"{accession} is a GEO named annotation, not a record; "
                    "follow 'Named Annotation'"
                )
            valid.append(accession)
        except AccessionError as err:
            out[accession] = err
    records = self.reader.records(valid) if valid else {}
    by_end = self._links(records) if records else {}
    absent = self._absent_errors(set(valid) - set(records))
    for accession in valid:
        record = records.get(accession)
        out[accession] = (
            absent[accession]
            if record is None
            else replace(record, links=tuple(by_end[accession]))
        )
    return {a: out[a] for a in order}