sc_tools.pipeline — Phase DAG#

Defines the pipeline phase graph as semantic slugs with explicit dependencies. Use this module to query available next steps and expected checkpoint paths.

from sc_tools.pipeline import get_available_next, get_phase_checkpoint

# What can run after QC and metadata?
completed = {"ingest_raw", "ingest_load", "qc_filter", "metadata_attach"}
next_phases = get_available_next(completed)
# -> {"preprocess"}

# Expected checkpoint path for a phase
path = get_phase_checkpoint("preprocess")
# -> "results/adata.normalized.h5ad"
class sc_tools.pipeline.PhaseSpec(label, depends_on=<factory>, branch='main', checkpoint=None, phase_group='data_processing', required_obs=<factory>, required_obsm=<factory>, x_format='', qc_report=None, old_code='', human_in_loop=False, optional=False, iterative=False)[source]#

Bases: object

Specification for a single pipeline phase.

Parameters:

  • label (str) – Human-readable name shown in reports and UIs.

  • depends_on (list[str | tuple[str, str]]) – List of (phase_group, subphase) tuples that must be complete before this phase can start. Empty list means this is a root phase. For backward compatibility, flat slug strings are also accepted and will be auto-converted to ("data_processing", slug) tuples at DAG construction time.

  • branch (str) – Conceptual branch name for grouping parallel tracks. Examples: “ingestion”, “main”, “scoring”, “demographics”, “meta”.

  • checkpoint (str | None) – Default output filename template. May contain {sample_id} or other format fields. None means the phase produces no single checkpoint file (e.g. figures-only or per-sample outputs).

  • phase_group (str) – The phase group this phase belongs to. Standard pipeline phases use "data_processing". Discovery phases use "discovery".

  • required_obs (list[str]) – obs columns that must exist after this phase completes.

  • required_obsm (list[str]) – obsm keys that must exist after this phase completes.

  • x_format (str) – Description of what X should contain (e.g. “raw counts”, “normalized”).

  • qc_report (str | None) – Filename template for the QC report produced by this phase (if any).

  • old_code (str) – Legacy phase code (e.g. “p0a”, “p1”, “p3.5b”) for documentation.

  • human_in_loop (bool) – If True, this phase requires human intervention to complete.

  • optional (bool) – If True, this phase can be skipped without breaking downstream phases.

  • iterative (bool) – If True, this phase can be re-entered (re-run) after completion without it being considered an error. Used for human-in-loop cycles such as manual cell typing.

label: str#
depends_on: list[str | tuple[str, str]]#
branch: str = 'main'#
checkpoint: str | None = None#
phase_group: str = 'data_processing'#
required_obs: list[str]#
required_obsm: list[str]#
x_format: str = ''#
qc_report: str | None = None#
old_code: str = ''#
human_in_loop: bool = False#
optional: bool = False#
iterative: bool = False#
sc_tools.pipeline.get_phase(slug)[source]#

Return the PhaseSpec for slug.

Accepts either a flat slug ("qc_filter") or a tuple (("data_processing", "qc_filter")).

Raises:

KeyError – If the slug is not registered.

Return type:

PhaseSpec

Parameters:

slug (str | tuple[str, str])

sc_tools.pipeline.get_available_next(completed)[source]#

Return phase keys whose dependencies are all satisfied.

Excludes phases that are already complete, except for iterative phases which can always be re-entered.

Parameters:

completed (list[str | tuple[str, str]]) – List of completed phases. Accepts flat slugs ("qc_filter") or tuples (("data_processing", "qc_filter")) or a mix.

Returns:

(phase_group, subphase) keys available to run next.

Return type:

list[tuple[str, str]]

sc_tools.pipeline.get_phase_checkpoint(slug, **kwargs)[source]#

Return the expected checkpoint path for a phase, with placeholders filled in.

Accepts either a flat slug or a (phase_group, subphase) tuple.

Parameters:

  • slug (str | tuple[str, str]) – Phase identifier (e.g. "qc_filter" or ("data_processing", "qc_filter")).

  • **kwargs (str) – Format arguments for the checkpoint template (e.g. sample_id="s1").

Returns:

Formatted checkpoint path, or None if the phase has no checkpoint.

Return type:

str or None

sc_tools.pipeline.get_dag()[source]#

Return a copy of the current phase DAG (standard + any custom phases).

Keys are (phase_group, subphase) tuples.

Return type:

dict[tuple[str, str], PhaseSpec]

sc_tools.pipeline.extend_dag(slug, spec)[source]#

Register a custom phase (project-specific or experimental).

Parameters:

  • slug (str) – Unique identifier for the phase (e.g. "spatial_regulon").

  • spec (PhaseSpec) – PhaseSpec describing the phase.

Raises:

ValueError – If any depends_on entry is not registered.

Return type:

None

sc_tools.pipeline.validate_dag()[source]#

Check that all depends_on references point to registered phases.

Returns:

Error messages (empty if the DAG is valid).

Return type:

list[str]