Analysis pipeline API reference

DataStore.pipeline runs CyteArc’s standard RNA recipe and records every invocation in the datastore. The result is a durable, read-only PipelineRun whose outputs are exact immutable ArtifactRef values.

cytearc.PipelineRun	A datastore-bound handle for one durable pipeline invocation.
cytearc.PipelineExecutionError	A handled pipeline failure with a durable run identifier.
cytearc.datastore.pipeline_accessor.PipelineAccessor	Store-bound entry point for the durable basic RNA pipeline.
cytearc.datastore.pipeline_accessor.PipelineEvent
cytearc.datastore.pipeline_accessor.PipelineAccessor.run	Run the validated rich RNA recipe and return its durable handle.
cytearc.datastore.pipeline_accessor.PipelineAccessor.open	Open one durable run by ID or immutable label.
cytearc.datastore.pipeline_accessor.PipelineAccessor.list_runs	List recent runs, optionally filtered by their terminal status.
cytearc.datastore.pipeline_accessor.PipelineAccessor.abandon_label_claim	Mark an exact stopped finalizer interrupted so its label can be retried.

Run the RNA recipe

run = ds.pipeline.run(assay="RNA", label="baseline")

run["pca"]
run["cluster_selection"]
run["clusters"]
run.cells.to_pandas_dataframe(["umap_1", "umap_2", "clusters"])
ds.plots.embedding(run=run, layout="umap", color_by="clusters")
ds.markers.load(marker=run["markers"], group_id=0)

The public signature is:

def run(
    *,
    assay: str | None = None,
    label: str | None = None,
    cell_key: str = "I",
    filtering: bool | Mapping[str, object] = True,
    harmony_batch_columns: Sequence[str] | None = None,
    hvg_count: int = 1000,
    pca_dims: int = 21,
    neighbors_k: int = 11,
    umap: bool = True,
    leiden: Mapping[str, object] | bool = True,
    cell_cycle: bool = True,
    paris: bool = True,
    doublets: bool = True,
    markers: bool = True,
    snapshot_columns: Sequence[str] = (),
    params: Mapping[str, object] | None = None,
    callback: PipelineCallback | None = None,
) -> PipelineRun: ...

The default recipe snapshots its inputs, filters cells, scores cell cycle, selects highly variable genes, normalizes, runs PCA, builds ANN, neighbour, and connectivity artifacts, initializes UMAP, runs UMAP, evaluates Leiden at 0.5, 0.75, 1.0, and 1.25, runs Paris, selects a clustering, scores doublets, and searches for markers. Stages run one at a time in this fixed order on the calling thread, so each stage keeps its own wall time, sampled resident memory, and artifact receipt.

Harmony is enabled by a non-empty harmony_batch_columns sequence. The main graph then uses the Harmony coordinates. When doublet scoring is enabled, its graph branch still uses the uncorrected PCA coordinates. The branch artifacts appear in their stage receipt, not as convenience entries in the run’s top-level output mapping.

Boolean stage options disable that stage when set to False. True uses that stage’s defaults. A mapping configures the stage. None is rejected. An empty filtering mapping uses the same automatic defaults as filtering=True. An empty Leiden mapping is invalid; a custom Leiden request contains exactly partitions:

run = ds.pipeline.run(leiden={"partitions": [0.4, 0.8]})

Doublets and markers require at least one Leiden candidate. Paris can still run as a diagnostic output when those stages are disabled; a Paris-only run has run["paris"] and no run["clusters"]. Setting umap=False skips UMAP, and the embedding initialization unless t-SNE, which uses it, is enabled; a skipped stage’s artifact does not appear in the completed run.

filtering=True uses MAD filtering over available assay QC columns. The default and method="auto" both resolve to method="mad". Set filtering to False to retain the captured input selection, or pass a configuration mapping. MAD filtering accepts attrs, n_mads, min_cells_per_sample, and optionally sample_column; without a sample column it uses pooled bounds. Groups with fewer than 20 active cells are retained with a warning by default, including small pooled selections. Pass method="gaussian" explicitly for the former Gaussian policy, with attrs, min_p, and max_p. Gaussian filtering does not accept a sample column. method="manual" requires aligned attrs, lows, and highs, with an optional Boolean keep_bounds value. Probability, MAD, and manual-bound values must be finite numbers when present; booleans and numeric strings are rejected rather than coerced. If filtering is requested and none of the default QC columns exists, validation raises instead of silently analyzing the unfiltered cells. Pass filtering=False only when that choice is deliberate.

run = ds.pipeline.run(
    filtering={
        "method": "manual",
        "attrs": ["RNA_nCounts", "RNA_nFeatures"],
        "lows": [1000, 500],
        "highs": [15000, 4000],
    },
)

Configure stages with one mapping

params configures the run with one mapping. Each key names a stage, and its value holds the keyword arguments the pipeline forwards to that stage’s function. The pipeline still supplies the artifacts each stage consumes. A setting has exactly the meaning of the same keyword on the method in the table below, and omitting it uses that method’s default.

Settings are JSON values, so a run configuration can be saved and passed again; infinity and NaN are never settings. Some settings spell a derived default as None: hvg max_cells, which excludes genes detected in at least the number of selected cells minus 20; batch_size of every stage that takes it; cell_cycle ctrl_size and log_transform; paris min_cluster_size; and normalization log_transform and renormalize_subset, whose defaults follow the assay’s normalizer (Assays and metadata API reference). The run records params as given, and each artifact records the values its method resolved.

run = ds.pipeline.run(
    params={
        "filtering": {"method": "manual", "attrs": ["RNA_nCounts"], "lows": [500], "highs": [20000]},
        "hvg": {"min_mean": 0.01, "max_mean": 5.0, "keep_bounds": True},
        "pca": {"dims": 30},
        "neighbors": {"k": 15},
        "umap": {"n_epochs": 400, "min_dist": 1.0, "spread": 2.0, "parallel": True},
        "leiden": {"partitions": [0.6, 0.8, 1.0, 1.2, 1.4], "selected": 1.0},
        "membership_strength": True,
        "tsne": {"max_iter": 800, "early_iter": 200, "alpha": 10, "box_h": 0.7},
        "species": "homo_sapiens",
    },
)

Section	Forwards to	Settings
filtering	the filtering option	the filtering mapping above, or a bool
cell_cycle	cell-cycle scoring	s_genes, g2m_genes, ctrl_size, log_transform, n_bins, rand_seed
hvg	hvgs()	top_n, min_cells, max_cells, min_mean, max_mean, min_var, max_var, n_bins, lowess_frac, blacklist, keep_bounds, bin_strategy
normalization	normalize()	log_transform, renormalize_subset
pca	pca()	dims, feat_scaling, batch_size
harmony	harmony()	batch_columns (required), harmony_params, batch_size
ann_index	ann_index()	ann_metric, ann_efc, ann_ef, ann_m, ann_parallel, rand_state, batch_size
neighbors	neighbors()	k, batch_size
connectivity	connectivity()	local_connectivity, bandwidth
embedding_initialization	initialization()	n_centroids, rand_state, batch_size, kmeans_sampling, kmeans_batch_size
umap	umap()	umap_dims, spread, min_dist, n_epochs, repulsion_strength, initial_alpha, negative_sample_rate, use_density_map, dens_lambda, dens_frac, dens_var_shift, random_seed, parallel, symmetric_graph, graph_upper_only
tsne	tsne()	tsne_dims, lambda_scale, max_iter, early_iter, alpha, box_h, symmetric_graph, graph_upper_only
leiden	leiden()	partitions, selected, backend, random_seed, symmetric_graph, graph_upper_only
membership_strength	membership_strength()	none; a bool
paris	Paris clustering	n_clusters, min_cluster_size
doublets	doublets()	cluster_sample_fraction, max_cells_per_cluster, simulation_ratio, heterotypic_fraction, save_k, smoothing_t, normalize_scores, random_seed
markers	marker search	none; a bool
species	recorded with the run	a species key such as homo_sapiens or mus_musculus

Optional stages also accept True or False: filtering, cell_cycle, harmony (False only), umap, tsne, leiden, membership_strength, paris, doublets, and markers. tsne and membership_strength are off unless requested. t-SNE uses the UMAP graph and embedding initialization and runs after UMAP; its coordinates appear as tsne_1, tsne_2, and so on. A new t-SNE embedding needs the optional sgtsnepi package of the tsne extra; without it the tsne stage fails, after the earlier stages have completed, and the run raises a PipelineExecutionError whose __cause__ is the ImportError that names the extra. A run whose t-SNE embedding the store already holds reuses it without the package, so a run that enables t-SNE without sgtsnepi is not refused: it logs a warning that names the extra before it creates a run record, and then runs. A saved configuration that sets parallel for t-SNE is rejected as an unknown setting when it runs. An existing run record still reopens and reports, because reopening does not validate its configuration against the current recipe. Membership strength is computed on the saved clustering and appears as membership_strength: the fraction of each cell’s graph neighbors that carry the cell’s own cluster label.

pca dims=0 skips the pca stage. The graph is then built on the normalized values of the selected features, one coordinate per feature: the ANN index, the neighbors, the embedding initialization, and cluster selection use run["normalized"] as their coordinates, and the run records no pca output. Harmony corrects reduced coordinates and doublet scoring needs a PCA graph, so dims=0 with harmony_batch_columns, or with doublet scoring on, which it is by default, raises ValueError before a run record is created; pass doublets=False. leiden selected names one of the partitions as the saved clustering, run["clusters"], in place of the silhouette choice, and the cluster-selection stage is skipped; the other partitions still run beside it.

With dims=0 the graph has one coordinate per highly variable gene, so the memory and time of the ANN index, the neighbour queries, the embedding initialization, and cluster selection scale with the HVG count instead of the PCA dimensions. The hnswlib index holds about 4 * hvg_count + 8 * ann_m + 130 bytes per cell on x86-64 Linux, about 4.5 GB for 1,000,000 cells with 1,000 HVGs and 8.5 GB with 2,000, and the ann_index and neighbors stages fail before they create or load it when it does not fit the memory budget: the run raises a PipelineExecutionError whose __cause__ is the MemoryError. By default the embedding initialization reads the coordinates in their stored row bands, and fits k-means in memory over every cell’s coordinates, 4 bytes per HVG per cell, when one band holds every cell; set params={"embedding_initialization": {"batch_size": ...}} below the cell count to select the streamed fit, which holds a sample of the cells and one block of coordinates at a time.

The doublets stage forces each simulated doublet, with probability heterotypic_fraction (0.8 by default), to pair parents from two different clusters, so it needs a saved clustering with two or more clusters. The silhouette choice never saves a partition with one cluster, but leiden selected can. The stage then fails with a PipelineExecutionError that names the settings to change: select a partition with two or more clusters, set params["doublets"]["heterotypic_fraction"] to 0 to simulate doublets from any two sampled cells, or pass doublets=False.

Unknown sections or settings, settings without a JSON value such as infinity, and True for a stage that always runs are rejected before a run record is created; hvg settings also get the checks of features.hvgs then. min_cells must be an integer of at least 0, top_n and n_bins integers of at least 1, lowess_frac a finite number from 0 to 1 for either bin_strategy, keep_bounds a boolean, and bin_strategy "adaptive" or "fixed"; an omitted setting is checked as its features.hvgs default. tsne settings get the checks of embeddings.tsne then too, because the stage runs after most others: tsne_dims, max_iter, and alpha must be integers of at least 1, early_iter an integer of at least 0, lambda_scale and box_h finite positive numbers, and symmetric_graph and graph_upper_only booleans; an omitted setting is checked as its embeddings.tsne default. The umap symmetric_graph and graph_upper_only settings must be booleans too, checked at the same time; embeddings.umap no longer accepts None for them. A setting cannot also be given through its shortcut argument: hvg_count with hvg.top_n, pca_dims with pca.dims, neighbors_k with neighbors.k, harmony_batch_columns with harmony, or a changed stage switch such as umap=False with params["umap"]. Each shortcut defaults to the default of the keyword it sets, so hvg_count, pca_dims, and neighbors_k follow features.hvgs top_n, reduction.pca dims, and graph.neighbors k. Every other stage checks its own values when it runs. The run’s configuration records the settings under params as JSON values.

The invocation is validated before a run record is created. Unknown options, missing columns, invalid stage combinations, reserved snapshot fields, and an already completed label fail without starting a run. A handled stage failure raises PipelineExecutionError. Use its run_id to inspect the persisted failure, and its stage to identify the failing stage. The original exception is retained as __cause__.

from cytearc import PipelineExecutionError

try:
    run = ds.pipeline.run(label="baseline")
except PipelineExecutionError as error:
    failed_run = ds.pipeline.open(run_id=error.run_id)
    report = failed_run.report(format="dict")

Automatic cluster selection

The cluster_selection stage scores enabled Leiden resolutions in the coordinates used to build the graph: PCA or Harmony coordinates, or with pca_dims=0 the normalized values. Paris remains run["paris"] for diagnosis and comparison; it is never the automatic run["clusters"] winner. Selection uses one deterministic shared sample of at most 10,000 selected cells with seed 4466. The sample reserves up to two seeded cells per cluster across every Leiden candidate, then fills remaining capacity without replacement. The sampled cells’ coordinates are read in blocks under the datastore memory budget, and pairwise distances are computed in chunks of a quarter of the budget, at most 1 GiB. Before it reads a coordinate, the stage fails when the sampled coordinates, in their stored dtype and as float64, with one block of the read or with the distance chunks, exceed the budget: the run raises a PipelineExecutionError whose __cause__ is the MemoryError. Reading a sample spread over a large normalized matrix, as with pca_dims=0, decodes most of its stored chunks.

This silhouette comparison is a reproducible provisional baseline. It is not biological validation or ground truth. Keep alternative Leiden refs and Paris when the study question needs other evidence.

The resulting cluster_selection artifact records candidate keys and refs, silhouette scores, the sample definition (sampleStrategy="sharedClusterQuota" and minClusterQuota=2), invalid-candidate reasons, deterministic tie order, and the selected key. run["clusters"] is the selected Leiden candidate’s exact ref. It is not a copied artifact or metadata column. If no Leiden candidate can be scored, the stage fails. A single valid Leiden candidate still receives a decision artifact.

Persisted cluster selection is intentionally pipeline-only. The fixed recipe owns the candidate set, deterministic sampling policy, and run-ledger decision. Granular callers can evaluate their own partitions with public APIs such as evaluate_cluster_separability in cytearc.metrics. There is no public “best clustering” function and no separate DataStore persistence method that could imply the pipeline policy outside a run. The agent orchestrator is a separate multi-metric workflow and does not replace this baseline.

Durable outputs and frozen views

A completed default run exposes these keys in order:

input_cell_selection, analysis_cell_selection, feature_universe, cell_cycle,
highly_variable_features, normalized, pca, ann_index, neighbors,
connectivity_map, embedding_initialization, umap,
leiden_0.5, leiden_0.75, leiden_1.0, leiden_1.25,
paris, cluster_selection, clusters, doublets, markers

harmony appears after pca when enabled. Disabled stages omit their outputs. Normal mapping operations such as run["pca"], list(run), and run.items() are available only on a successfully completed run.

run.cells and run.features are narrow read-only views over captured selections, metadata snapshots, and result artifacts:

view.columns
view.fetch(column)                 # rows selected by the run's stored I
view.fetch_all(column)             # values aligned to the complete stored axis
view.to_pandas_dataframe(columns)  # selected rows
view.head(n=5)

Cell I is the analysis selection. Feature I is backed by the immutable run["feature_universe"] selection created during the input stage, not the live feature I column. The views retain captured names and requested metadata even if live values or live I later change. They fail closed if ordered cell or feature identities change. There is no live overlay or mutation surface.

Pipeline execution writes immutable artifacts and run records only. It never inserts result columns or rewrites live I. Plotting, marker loading, and export stay on DataStore:

ds.plots.embedding(run=run, layout="umap", color_by="clusters")
markers = ds.markers.load(marker=run["markers"], group_id=0)
adata = ds.to_anndata(run=run)
# adata.obsm["X_umap"] holds frozen UMAP; cluster labels stay in adata.obs

The marker table of a run saved before CyteArc 1.0.0 cannot be read, because its fold_change column held sentinels: markers.load raises and names ds.markers.search(run["clusters"], features=run["feature_universe"]), which recomputes the table for the run’s clusters. A saved agent result keeps the table that it read, so its analysis must be run again.

Open, list, and report runs

A successful optional label is an immutable name for one run:

run = ds.pipeline.open(label="baseline")
same_run = ds.pipeline.open(run_id=run.run_id)

recent = ds.pipeline.list_runs(limit=20)
failed = ds.pipeline.list_runs(status="failed")
interrupted = ds.pipeline.list_runs(status="interrupted")

open requires exactly one of run_id or label. Only a successfully completed run acquires its requested label. Failed and interrupted attempts do not reserve it. list_runs returns newest first and includes all statuses unless filtered. Concurrent finalizers use an atomic label claim, so at most one completed run can acquire a name. A storage backend without atomic conditional creation rejects label= before a run record or computation starts; unlabeled runs are unaffected. An unclean incomplete finalizer blocks reuse of its requested label and fails closed.

After confirming that the owner process has stopped, explicitly abandon that exact claim before retrying the label:

interrupted = ds.pipeline.abandon_label_claim(
    label="baseline",
    run_id=stopped_run_id,
    reason="worker terminated after finalization began",
)
replacement = ds.pipeline.run(label="baseline")

The recovery call succeeds only for the exact current owner while complete=False, including a torn terminal payload, and records an abandoned_label_claim interruption. It refuses every complete=True owner and has no timeout or automatic stale-process heuristic.

Every status exposes identity, status, and run.report(format="dict" | "markdown"). Only a completed run exposes mapping outputs and frozen views. Reports include stage timing, sampled process-tree RSS, artifact plans with created or reused dispositions, failures, interruption details, and signal-guard availability. RSS peaks are sampled lower bounds. An unavailable measurement is reported as null with a reason. RSS comes from the Linux /proc filesystem; on platforms without it, such as macOS and Windows, nothing is sampled, so sampleCount is 0.

A hard process death can leave complete=False; this is reported as an unclean incomplete run. There is no resume, repair, or same-ID retry. A new invocation may reuse only complete artifacts.

Runs saved by other releases

A run completed under any 1.x release reopens under every later 1.x release. open by label or run ID, list_runs, run[key], run.report(), and the frozen run.cells and run.features views work as long as the artifacts the run references exist and satisfy their payload contracts. Running a saved configuration again recomputes a stage whose operation gained a revision that applies to it, or whose recorded provenance a later release changed; it never modifies the earlier run. Lineage reports of the earlier run’s outputs mark those that a revision supersedes stale.

A run that a newer CyteArc saved with record fields this release does not know fails closed: opening it by run ID raises a ValueError that names those fields. Upgrade CyteArc to open it. Within 1.x, releases only add record fields, so runs from earlier releases never meet this error. A 2.0 release may change run records and lists each change in its compatibility inventory.

Run and stage records are otherwise strict: a missing or unknown field, or a malformed value, fails closed. list_runs and open-by-label skip records they cannot read so healthy runs remain accessible, and log a warning that names each skipped run ID and why it could not be read. Opening that run by its exact ID raises its error.

Graceful interruption and callbacks

On the main interpreter thread, a pipeline temporarily installs cooperative handlers for available SIGTERM, SIGINT, and SIGHUP, while respecting signals already set to ignore. The first signal requests shutdown at a safe checkpoint. The pipeline persists completed or interrupted stage and run state, invokes interruption callbacks, restores prior handlers, and propagates the original signal behavior. A second catchable termination signal escalates immediately. Non-main-thread runs record that signal protection was unavailable.

KeyboardInterrupt and an escaped asyncio.CancelledError use the same durable interruption boundary. Ordinary stage exceptions produce a failed run instead. Run and stage records hold at most 512 characters of an error or interruption message. A longer message keeps its first 509 characters followed by ...; the raised exception keeps its full message. If a termination signal races with an ordinary failure or the final successful handoff, the pipeline preserves that durable outcome and still propagates the pending signal after cleanup.

SIGKILL, SIGSTOP, out-of-memory termination, power loss, and expired shutdown grace periods cannot perform cleanup. Their durable contract is an incomplete artifact or run, followed by a new invocation that reuses only complete artifacts.

Callbacks provide serialized progress notifications without controlling execution:

from cytearc.datastore.pipeline_accessor import PipelineEvent

events: list[tuple[str, str]] = []


def record_event(event: PipelineEvent) -> None:
    events.append((event.kind, event.stage))


run = ds.pipeline.run(callback=record_event)

Enabled stages emit stage_started once their record is written, then stage_completed, stage_failed, or stage_interrupted once their outcome is recorded, before the next stage starts. A handled interruption also emits pipeline_interrupted after durable state is written. A skipped stage is present in the report and emits no stage event, unless an error or interruption arrives between writing its record and recording it as skipped; it is then recorded as failed or interrupted and emits stage_failed or stage_interrupted without a preceding stage_started. Callback errors are logged and cannot block durable status.

Public types

class cytearc.datastore.pipeline_accessor.PipelineAccessor(store)

Store-bound entry point for the durable basic RNA pipeline.

abandon_label_claim(*, label, run_id, reason)

Mark an exact stopped finalizer interrupted so its label can be retried.

The caller must first confirm that the process executing run_id has stopped. CyteArc does not infer abandonment from elapsed time because a slow live finalizer must never lose its label claim.

Return type:

PipelineRun

list_runs(*, status=None, limit=20)

List recent runs, optionally filtered by their terminal status.

Return type:

tuple[PipelineRun, ...]

open(*, run_id=None, label=None)

Open one durable run by ID or immutable label.

Return type:

PipelineRun

run(*, assay=None, label=None, cell_key='I', filtering=True, harmony_batch_columns=None, hvg_count=1000, pca_dims=21, neighbors_k=11, umap=True, leiden=True, cell_cycle=True, paris=True, doublets=True, markers=True, snapshot_columns=(), params=None, callback=None)

Run the validated rich RNA recipe and return its durable handle.

params maps a stage name to the keyword arguments the pipeline forwards to that stage’s function, so one mapping can configure a run:

{"filtering": {"method": "manual", "lows": [...], "highs": [...]}, "hvg": {"min_mean": 0.01, "keep_bounds": True}, "pca": {"dims": 30}, "umap": {"n_epochs": 400}, "leiden": {"partitions": [0.8, 1.0], "selected": 1.0}, "tsne": {"max_iter": 800}}

Sections:

• filtering: the filtering mapping, or a bool.

• cell_cycle, umap, paris, doublets, markers: a mapping of settings, or a bool that switches the stage.

• hvg, normalization, pca, ann_index, neighbors, connectivity, embedding_initialization: settings for stages that always run. pca dims=0 skips PCA and builds the graph on the normalized values of the selected features; it requires doublets=False and no Harmony batch columns.

• harmony: batch_columns plus Harmony settings, or False.

• leiden: partitions, an optional selected resolution that becomes the saved clustering in place of the silhouette choice, and Leiden settings; or a bool.

• tsne and membership_strength: off unless requested. t-SNE uses the UMAP graph and initialization; membership strength is computed on the saved clustering.

• species: a species key recorded with the run.

Unknown sections or settings raise before anything is written, and a setting cannot also be given through its shortcut argument, such as hvg_count with params["hvg"]["top_n"]. Each stage validates its own values. The run records the resolved settings.

Raises:

UnmeasuredCellsError – If the assay did not measure a cell of cell_key.

Return type:

PipelineRun

class cytearc.datastore.pipeline_accessor.PipelineEvent(kind, stage, error=None)

class cytearc.PipelineRun(owner, record)

A datastore-bound handle for one durable pipeline invocation.

report(*, format='dict')

Return deterministic persisted facts for any run status.

Overloads:

• self, format (Literal[‘dict’]) → dict[str, Any]

• self, format (Literal[‘markdown’]) → str

class cytearc.PipelineExecutionError(run_id, stage, cause)

A handled pipeline failure with a durable run identifier.

See Quick start for an executable first run and Graph construction API reference for explicit stage-by-stage APIs.