bioimageflow ¶

set[InvalidatedSelection]

Parameters:

name (str)
source (Any)
id (str | None)

topological_order()[source]¶

Return node names in dependency order. Raises on cycle.

Thin wrapper over bioimageflow.engine.topological_order(). If the graph may contain a cycle, call validate() first.

Return type:: list[str]

invalidate(node_ids, *, cascade=True)[source]¶

Remove cache selections for the given nodes.

Returns the selections whose current.json pointers were removed. cascade=True (the default) also removes selections for every node transitively downstream of each input node, so a subsequent run recomputes or reselects everything that depended on the changed node. Immutable records/<record-id>/ directories are retained.

KeyError is raised if any name in node_ids is not registered with this workflow — matching the existing behavior of downstream_of().

Return type:

Parameters:

node_ids (Iterable[str])
cascade (bool)

Concurrency¶

This method is not safe to call concurrently with compute() on the same workflow. The library does not currently expose a public lock primitive; callers that need to invalidate while a compute is in flight must coordinate externally (e.g., cancel + join + invalidate).

downstream_of(node_name)[source]¶

Return node names transitively downstream of node_name.

Excludes node_name itself. Useful for callers (GUIs, external schedulers) that need to mark dependents as cache-invalidated after a parameter change.

Return type:: set[str]
Parameters:: node_name (str)

plan(*, dev_mode=False)[source]¶

Return a per-node cache-status plan.

Instantiates a non-Wetlands DefaultEngine and calls its plan(). No tool code runs.

Return type:: dict[str, NodePlan]
Parameters:: dev_mode (bool)

create_engine(*, environment_lifetime='execution', env_manager=None)[source]¶

Create an engine preserving this workflow’s execution configuration.

environment_lifetime controls whether Wetlands workers stop after each execution ("execution"), remain warm until engine.close() ("engine"), or are owned entirely by the caller ("external"). An existing manager can be injected so multiple workflows and engines share the same worker environments.

Return type:

DefaultEngine

Parameters:

environment_lifetime (EnvironmentLifetime | str)
env_manager (WetlandsEnvManager | None)

validate(*, dev_mode=False, _recursion_stack=())[source]¶

Return all domain-level problems in this workflow.

Runs, in order:

Cycle detection (one error per cycle).
Type compatibility on every column binding.
Missing-required-input check for every node.
Pydantic validation of every node’s supplied constants.
Recursive validation of workflow invocations (path is prefixed with the parent’s node name).

Steps 1–3 are already enforced by Node.__init__ during construction; this method exists so GUIs that built the workflow via capture_errors() / from_dict() can re-check after the fact. Step 4 (constant Pydantic validation) only runs here — it is intentionally not performed at construction time, so a GUI editing one field at a time does not need every other field to be valid yet.

Parameters:

dev_mode (bool) – Accepted for symmetry with plan(); unused by validate.
_recursion_stack (tuple[int, ...])

Returns:

Deduplicated, sorted by (path, node, field, kind).

Return type:

list[ValidationError]

capture_errors()[source]¶

Capture node-construction errors as ValidationError.

Usage:

wf = Workflow()
with wf, wf.capture_errors() as errors:
    MyTool()(input=upstream["bad_col"])
# errors: list[ValidationError]

Nested blocks push their own list; the outer list is restored on exit. Disables the “raise on first error” behavior of Node construction only for the duration of the block.

Return type:: Iterator[list[ValidationError]]

property nodes: dict[str, Node]¶

property errors: list[ValidationError]¶

Build-time errors accumulated during from_dict().

Empty when the workflow was constructed programmatically (via the context-manager / call-tools pattern) or when from_dict was called in strict mode.

property failed_nodes: dict[str, ValidationError]¶

Map of node name → ValidationError for nodes that failed to construct during from_dict().

Populated only when from_dict is called with partial=True and a node’s tool resolution or construction raised. Empty otherwise.

property is_partial: bool¶

Whether the workflow is missing nodes that the input dict described.

True when at least one entry in the source data["nodes"] is absent from nodes (typically because it failed to construct in collect mode). False for fully-built workflows and for workflows constructed without from_dict().

disable(*nodes)[source]¶

Disable nodes by reference or name.

Return type:: None
Parameters:: nodes (Node | str)

enable(*nodes)[source]¶

Enable nodes by reference or name.

Return type:: None
Parameters:: nodes (Node | str)

cancel()[source]¶

Request cancellation of the running workflow.

Return type:: None

property cancel_requested: bool¶: Whether cancellation has been requested.

get_environment(target)[source]¶

Get the launch configuration proxy for an environment.

Parameters:: target (ProcessingTool | EnvironmentSpec | str) – A ProcessingTool instance, an EnvironmentSpec, or an env name string.
Return type:: WorkflowEnvironment
Returns:: A shared WorkflowEnvironment proxy. Multiple calls with tools sharing the same environment return the same object.

compute(*targets, inputs=None, dev_mode=False, engine=None)[source]¶

Execute the workflow and return results.

Parameters:

dev_mode (bool) – Development mode flag
engine (Optional[DefaultEngine]) – Optional pre-configured engine to use. If None, the configured engine backend and execution policy are used. Providing an engine allows post-execution inspection and testing.
targets (Node)
inputs (Mapping[str, Any] | None)

Return type:

compute_steps(*targets, inputs=None, dev_mode=False, engine=None)[source]¶

Execute the workflow step by step, yielding a NodeStep for each node in topological (dependency) order.

Parameters:

dev_mode (bool) – Development mode flag
engine (Optional[DefaultEngine]) – Optional pre-configured engine to use. If None, a default DefaultEngine is created.
targets (Node)
inputs (Mapping[str, Any] | None)

Return type:

Generator[NodeStep, None, None]

The engine stays alive between yields so Wetlands environments remain warm — ideal for interactive debugging.

Usage:

for step in wf.compute_steps(results):
    print(f"Next: {step.node_name}")
    step.prepare()     # optional: launches env — attach debugger here
    df = step.execute()
    print(df.head())

If step.execute() is not called before advancing to the next iteration, the step auto-executes to keep downstream nodes consistent.

export_outputs(*, mode='symlink', scope='latest', run_id=None)[source]¶

Materialize human-facing output files from the portable JSON views.

Return type:

list[Path]

Parameters:

mode (Literal['pointer', 'symlink', 'copy', 'hardlink'])
scope (Literal['latest', 'runs', 'both'])
run_id (str | None)

to_dict(*, include_custom_tools=False)[source]¶

Serialize the strict recursive schema-version-1 graph.

Return type:: dict[str, Any]
Parameters:: include_custom_tools (bool)

export(path)[source]¶

Serialize the workflow to a JSON file or BioImageFlow zip archive.

Return type:: None
Parameters:: path (str | Path)

classmethod load(path)[source]¶

Deserialize a workflow from a JSON file.

Thin wrapper around from_dict(). Preserves the original behavior: raises on the first error, auto-installs missing versioned packages.

Return type:: Workflow
Parameters:: path (str | Path)

classmethod from_python(path_or_module)[source]¶

Execute a trusted module’s exact build_workflow factory once.

Return type:: Workflow
Parameters:: path_or_module (str | Path | Any)

classmethod import_archive(path, destination)[source]¶

Extract a BioImageFlow zip archive to destination and load it.

Return type:

Workflow

Parameters:

path (str | Path)
destination (str | Path)

classmethod from_dict(data, *, validate_only=False, partial=False, auto_install=True, storage_path_override=None, on_progress=None, engine=None, execution=None, wetlands_config=None)[source]¶

Reconstruct a Workflow from a serialized dict.

Parameters:

data (dict[str, Any]) – A schema-version-1 recursive graph produced by to_dict(), or a portable archive envelope produced by export().
validate_only (bool) – Drives the return type. When True, returns a (workflow, errors) tuple; the workflow may be partial. When False (default), returns the Workflow directly and aggregates any captured errors into a raised exception.
partial (bool) – Drives error suppression / continuation. When True, per-node failures are captured as ValidationError entries and construction continues; the workflow may be best-effort partially wired. When False (default), construction stops at the first failure.
auto_install (bool) – When True (default), missing versioned packages are installed automatically. When False, missing packages produce an unknown_tool error (when captured) or raise.
storage_path_override (Union[str, Path, None]) – Override data["config"]["storage_path"] without mutating the dict. Useful for GUIs that validate a graph against a specific cache path.
on_progress (Optional[Callable[[ProgressEvent], None]]) – Passed to Workflow. None means “use the values from data['config'] (or defaults)”.
engine (Optional[str]) – Passed to Workflow. None means “use the values from data['config'] (or defaults)”.
execution (Optional[str]) – Passed to Workflow. None means “use the values from data['config'] (or defaults)”.
wetlands_config (Optional[dict[str, Any]]) – Passed to Workflow. None means “use the values from data['config'] (or defaults)”.

Return type:

Workflow | tuple[Workflow, list[ValidationError]]

Notes

The partial=False, validate_only=True combination returns a (workflow, errors) tuple where errors contains at most one entry (the first failure) and the workflow may be empty — useful as a fail-fast diagnostic.

Node¶

Node and ColumnRef — graph construction primitives.

exception bioimageflow.node.ColumnNotFoundError[source]¶

Bases: Exception

Raised when a column reference targets a non-existent column.

to_validation_error(node, field=None)[source]¶

Return type:

ValidationError

Parameters:

node (str)
field (str | None)

exception bioimageflow.node.BindingError[source]¶

Bases: Exception

Raised when a required input field has no source.

to_validation_error(node, field=None, kind='missing_input')[source]¶

Return type:

ValidationError

Parameters:

node (str)
field (str | None)
kind (Literal['cycle', 'type_mismatch', 'missing_input', 'unknown_input', 'column_not_found', 'parameter_invalid', 'unknown_tool', 'duplicate_name', 'construction_failed', 'source_tool_upstream'])

exception bioimageflow.node.IndexAlignmentError[source]¶

Bases: Exception

Raised when upstream indices are incompatible.

to_validation_error(node, field=None)[source]¶

Return type:

ValidationError

Parameters:

node (str)
field (str | None)

exception bioimageflow.node.SourceToolUpstreamError[source]¶

Bases: Exception

Raised when a source DataFrameTool (accepts_upstream = False) is constructed with positional upstream arguments.

to_validation_error(node=None, field=None)[source]¶

Return type:

ValidationError

Parameters:

node (str | None)
field (str | None)

class bioimageflow.node.ColumnRef[source]¶

Bases: object

References a specific column from a specific upstream node.

node: Any¶

column: str¶

__init__(node, column)¶

Parameters:

node (Any)
column (str)

Return type:

None

bioimageflow.node.scoped_node_names(names)[source]¶

Expose immutable execution paths without changing structural names.

Parameters:: names (dict[Node, str])

bioimageflow.node.get_active_workflow()[source]¶

Return type:: Any

bioimageflow.node.set_active_workflow(wf)[source]¶

Return type:: None
Parameters:: wf (Any)

class bioimageflow.node.Node[source]¶

Bases: object

A node in the computation DAG. Wraps a tool and its configuration.

__init__(tool, kwargs=None, args=None, name=None, output_templates=None)[source]¶

Parameters:

tool (BaseTool)
kwargs (dict[str, Any] | None)
args (list[Any] | None)
name (str | None)
output_templates (dict[str, str] | None)

Return type:

None

property name: str¶

get_output_schema()[source]¶

Resolve this node’s output column schema as currently configured.

Algorithm: :rtype: dict[str, dict[str, Any]] | None

DataFrameTool with a resolve_merge_schema override (built-in merge tools): collect upstream schemas via each positional arg’s get_output_schema() and call tool.resolve_merge_schema(upstream_schemas, kwargs).
DataFrameTool (or subclass) with resolve_outputs → tool.resolve_outputs(kwargs).
ProcessingTool → static serialize_output_schema(type(tool)).

Returns None when the schema is unresolvable (any required upstream returns None, or the tool has no Outputs and no override). Idempotent and side-effect free.

Return type:: dict[str, dict[str, Any]] | None

disable()[source]¶

Disable this node so it is skipped during execution.

Return type:: None

enable()[source]¶

Re-enable this node for execution.

Return type:: None

compute(**kwargs)[source]¶

Shorthand: create/use a Workflow and compute this node.

Return type:: Any
Parameters:: kwargs (Any)

Engine¶

Execution engines for BioImageFlow workflows.

class bioimageflow.engine.EnvironmentLifetime[source]¶

Bases: str, Enum

Ownership policy for Wetlands environments used by an engine.

EXECUTION preserves the historical behavior and stops environments after every DefaultEngine.execute() or DefaultEngine.execute_steps() call. ENGINE retains them until DefaultEngine.close(). EXTERNAL leaves cleanup entirely to the owner of the injected environment manager.

EXECUTION = 'execution'¶

ENGINE = 'engine'¶

EXTERNAL = 'external'¶

__new__(value)¶

bioimageflow.engine.topological_order(workflow)[source]¶

Return the names of the nodes in workflow in dependency order.

Uses graphlib.TopologicalSorter over all registered nodes. Raises CycleError if the graph contains a cycle; callers that want to tolerate cycles should catch that or use Workflow.validate().

Return type:: list[str]
Parameters:: workflow (Any)

bioimageflow.engine.source_processing_signature_material(node)[source]¶

Return the cache-signature material for source ProcessingTool nodes.

Return type:: dict[str, Any]
Parameters:: node (Node)

exception bioimageflow.engine.DisabledNodeError[source]¶

Bases: Exception

Raised when all requested target nodes are disabled or unreachable.

exception bioimageflow.engine.WorkflowCancelledError[source]¶

Bases: Exception

Raised when a workflow execution is cancelled via Workflow.cancel().

exception bioimageflow.engine.CycleInWorkflowError[source]¶

Bases: ValueError

Raised by DefaultEngine.plan() / Workflow.plan() when the graph contains a cycle.

Use Workflow.validate() for non-fatal cycle reporting (it returns a ValidationError with kind="cycle" instead of raising).

__init__(nodes)[source]¶

Parameters:: nodes (list[str])
Return type:: None

class bioimageflow.engine.NodePlanStatus[source]¶

Bases: str, Enum

Per-node status returned alongside NodePlan.

Use this enum (or its string values) instead of inspecting the cache directory layout — the platform should not need to know how the library stores cached node results.

Values¶

CACHED: The node result key has a valid selected current record; compute() would short-circuit.
PRIOR_SELECTION_MISS: The planned result key has no selected current record, but the same node has another selected current record from prior result-key material. compute() would re-execute and publish or select a record.
UNEXECUTED: No known cache record exists for this node.
SKIPPED: The node is disabled, or has a disabled upstream that prevents execution. final_result_key and selected_record_id are None.
PENDING_UPSTREAM: At least one upstream selected record is not known yet, so the node’s final result key cannot be determined from a stable record graph snapshot.

CACHED = 'cached'¶

PRIOR_SELECTION_MISS = 'prior_selection_miss'¶

UNEXECUTED = 'unexecuted'¶

SKIPPED = 'skipped'¶

PENDING_UPSTREAM = 'pending_upstream'¶

__new__(value)¶

class bioimageflow.engine.NodePlan[source]¶

Bases: object

A single node’s pre-execution plan entry, returned by DefaultEngine.plan().

Variables:

node_name – Scoped node name ("workflow-node/internal" for nested workflow internals, plain name otherwise).
final_result_key – V1 result key for this node when it can be computed from known selected upstream records; None for skipped and pending nodes.
selected_record_id – Currently selected record ID for final_result_key when the node is cached; otherwise None.
status – Per-node NodePlanStatus — the canonical signal for external callers wanting to display “cached / prior selection miss / unexecuted / skipped” in a GUI.
upstream – Scoped names of this node’s direct upstreams.
pending_upstreams – Upstream node names whose selected records are not known yet.
logical_signature – Diagnostic logical signature computed by the same helpers as execution. It is not the public cache identity field.
accessors (cached and skipped are read-only convenience)
CACHED, (derived from status (cached == status is)
SKIPPED). (skipped == status is)

node_name: str¶

logical_signature: str¶

status: NodePlanStatus¶

upstream: tuple[str, ...]¶

final_result_key: str | None = None¶

selected_record_id: str | None = None¶

pending_upstreams: tuple[str, ...] = ()¶

property cached: bool¶

property skipped: bool¶

__init__(node_name, logical_signature, status, upstream, final_result_key=None, selected_record_id=None, pending_upstreams=())¶

Parameters:

node_name (str)
logical_signature (str)
status (NodePlanStatus)
upstream (tuple[str, ...])
final_result_key (str | None)
selected_record_id (str | None)
pending_upstreams (tuple[str, ...])

Return type:

None

exception bioimageflow.engine.WorkerTimeoutError[source]¶

Bases: RuntimeError

Raised when the engine-side safety timeout fires on a Wetlands task.

This is a last-resort timeout that wraps task.wait_for() in _dispatch_via_wetlands. It only fires when the Wetlands-side health monitor fails to catch a dead or hung worker within worker_timeout * 1.5 (or worker_timeout + 60, whichever is larger).

exception bioimageflow.engine.WorkerTaskError[source]¶

Bases: RuntimeError

Raised when a Wetlands worker task fails while executing a node.

__init__(message=None, *, node_name='', tool_class='', environment_name='', row_index=None, original=None, task_status=None, task_traceback=None)[source]¶

Parameters:

message (str | None)
node_name (str)
tool_class (str)
environment_name (str)
row_index (str | None)
original (BaseException | None)
task_status (Any | None)
task_traceback (Any | None)

Return type:

None

class bioimageflow.engine.NodeStep[source]¶

Bases: object

Handle for a single node in a stepped workflow execution.

Yielded by DefaultEngine.execute_steps(). The caller may optionally call prepare() (to warm up the Wetlands environment before execution — useful for attaching a debugger) and must call execute() to run the node (or it will auto-execute when the generator advances to the next step).

__init__(node, engine, results, sig_hashes, workflow, skipped=False)[source]¶

Parameters:

node (Node)
engine (DefaultEngine)
results (dict[Node, DataFrame])
sig_hashes (dict[Node, str])
workflow (Any)
skipped (bool)

Return type:

None

property skipped: bool¶: True if the node is disabled or has a disabled upstream.

property node_name: str¶: Name of the node about to be executed.

property tool: Any¶: The tool instance associated with this node.

property environment: Any¶: The EnvironmentSpec for ProcessingTools, None for DataFrameTools.

property cached: bool¶

True if the node’s result is already in the cache.

The first access triggers logical-signature computation and cache lookup; subsequent accesses reuse the result.

prepare()[source]¶

Launch the tool’s Wetlands environment (ProcessingTool only).

No-op for DataFrameTools, when Wetlands is disabled, or when the node’s result is already cached (no environment needed). After this call the environment is running and a debugger can be attached to it before execute() triggers the actual computation.

Return type:: None

execute()[source]¶

Execute the node and return its output DataFrame.

Idempotent — calling more than once returns the cached result. If the cache was already checked (via cached or prepare()) and a hit was found, returns it directly without re-entering the engine. Raises DisabledNodeError if the node is skipped.

Return type:: DataFrame

class bioimageflow.engine.DefaultEngine[source]¶

Bases: object

Executes workflow nodes with optional parallelism.

When _force_sequential is False (default), independent DAG branches can execute concurrently and intra-node rows can run in parallel across Wetlands workers. When True (set by SequentialEngine), execution is strictly sequential — useful for debugging and deterministic reproduction.

The non-Wetlands _dispatch_direct() path is a testing/development fallback and does not support sub-row progress reporting (Feature 3) or cooperative cancellation (Feature 4).

__init__(use_wetlands=False, wetlands_config=None, force_sequential=False, env_manager=None, environment_lifetime=EnvironmentLifetime.EXECUTION)[source]¶

Parameters:

use_wetlands (bool)
wetlands_config (dict[str, Any] | None)
force_sequential (bool)
env_manager (WetlandsEnvManager | None)
environment_lifetime (EnvironmentLifetime | str)

Return type:

None

property environment_lifetime: EnvironmentLifetime¶: Return the engine’s Wetlands environment ownership policy.

property environment_manager: WetlandsEnvManager | None¶: Return the Wetlands manager used by this engine, if any.

close()[source]¶

Close the engine and stop environments it owns.

The method is idempotent. Externally owned managers are never shut down; their owner must call WetlandsEnvManager.shutdown_all().

Return type:: None

execute(targets, workflow)[source]¶

Execute the workflow, returning results for target nodes.

Return type:

dict[str, DataFrame]

Parameters:

targets (list[Node])
workflow (Any)

execute_steps(targets, workflow)[source]¶

Yield a NodeStep for each node in topological order.

The engine (and any Wetlands environments) stays alive between yields. The caller controls execution via step.prepare() / step.execute(). If execute() is not called before advancing, the step auto-executes to keep the results chain consistent for downstream nodes.

WorkflowNodes are expanded: their internal nodes are yielded individually with scoped names (workflow_node/internal_name).

Cleanup runs when the generator is exhausted or closed (early break).

Return type:

Generator[NodeStep, None, None]

Parameters:

targets (list[Node])
workflow (Any)

plan(workflow)[source]¶

Return the cache status and diagnostic plan state of every node.

Walks the graph in topological order, computes diagnostic logical signatures with the same helpers as execute(), and reports result-key/current-record state when enough upstream cache selections are known. No tool code runs, and no Wetlands environment is launched.

Nested workflow tools appear under scoped names ("workflow_node/internal_name"), matching execute_steps().

Disabled nodes and nodes downstream of disabled nodes are returned with skipped=True and no final result key.

Raises:: CycleInWorkflowError – If the graph contains a cycle. Use Workflow.validate() for non-fatal cycle reporting.
Return type:: dict[str, NodePlan]
Parameters:: workflow (Any)

class bioimageflow.engine.SequentialEngine[source]¶

Bases: DefaultEngine

Forces sequential execution — useful for debugging and deterministic reproduction.

Inherits from DefaultEngine but forces _force_sequential=True and overrides worker resolution to always use a single worker with no worker_env.

__init__(**kwargs)[source]¶

Parameters:: kwargs (Any)
Return type:: None

DataFrameTool¶

DataFrameTool base class — main process only.

class bioimageflow.dataframe_tool.Passthrough[source]¶

Bases: IOModel

Marker base class for DataFrameTool Outputs that preserve input columns.

class bioimageflow.dataframe_tool.DataFrameTool[source]¶

Bases: BaseTool

Tool that transforms DataFrames in the main process.

accepts_upstream: bool = True¶

Whether this tool accepts positional upstream Nodes.

Set to False on source tools (e.g. Files, Generate) that produce a DataFrame from constants alone. A source tool raises SourceToolUpstreamError if any positional argument is passed.

merge_dataframes(dfs, arguments)[source]¶

Default: inner join on index.

Return type:

Parameters:

dfs (list[Any])
arguments (Any)

transform(df, arguments)[source]¶

Default: identity (passthrough).

Return type:

Parameters:

df (Any)
arguments (Any)

classmethod resolve_outputs(inputs=None)[source]¶

Return the output column schema for the given (possibly partial) inputs.

Each value has the per-field shape produced by serialize_output_schema(): {"type": str, "default": Any | None, "image_spec": dict | None}.

Default implementation: :rtype: dict[str, dict[str, Any]] | None

Outputs declared as IOModel → returns the static schema.
Outputs is a Passthrough subclass → returns the marker dict {"_passthrough": True, **declared_extra_fields}.
Otherwise → returns None (schema unknown without input information).

Tools whose output column names are derived from their inputs (e.g. Generate, where column_name is a runtime parameter) override this method. Merge tools whose output schema depends on upstream schemas (rather than just inputs) instead override resolve_merge_schema().

Must be pure (no I/O, no side effects). Returning None or a partial dict is allowed when inputs is incomplete.

Parameters:: inputs (dict[str, Any] | None)
Return type:: dict[str, dict[str, Any]] | None

classmethod resolve_merge_schema(upstream_schemas, inputs=None)[source]¶

Return the merged output schema given upstream schemas.

Override this on merge tools whose output columns are computed from upstream column sets (InnerJoin, CrossJoin, JoinOnColumn, Concat, Collect).

Returns None when any required upstream schema is None (caller should treat that as “schema unresolvable”). The default implementation returns None so Node.get_output_schema() falls back to resolve_outputs() for non-merge tools.

Each entry has the same shape as the values returned by resolve_outputs().

Return type:

dict[str, dict[str, Any]] | None

Parameters:

upstream_schemas (list[dict[str, dict[str, Any]] | None])
inputs (dict[str, Any] | None)

Cache¶

Hashing, caching, and provenance.

bioimageflow.cache.normalize_dependencies(dependencies)[source]¶

Normalize dependencies for consistent hashing.

Return type:: dict[str, Any]
Parameters:: dependencies (dict[str, Any])

bioimageflow.cache.compute_env_hash(dependencies)[source]¶

SHA256 of normalized dependencies.

Return type:: str
Parameters:: dependencies (dict[str, Any])

bioimageflow.cache.deterministic_serialize(obj)[source]¶

Serialize an object deterministically for hashing.

Return type:: str
Parameters:: obj (Any)

bioimageflow.cache.compute_signature_hash(tool_name, tool_version, env_hash, resolved_params, upstream_hashes, source_hash=None)[source]¶

Compute the logical digest for a node.

Return type:

Parameters:

tool_name (str)
tool_version (str)
env_hash (str)
resolved_params (Any)
upstream_hashes (dict[str, str])
source_hash (str | None)

bioimageflow.cache.cache_load(cache_path)[source]¶

Load a DataFrame from cache.

Accepts either a .parquet or .csv path.

Return type:: DataFrame
Parameters:: cache_path (Path)

bioimageflow.cache.dataframe_result_key(node_name, sig_hash)[source]¶

Return the result key for a DataFrameTool node.

Return type:

Parameters:

node_name (str)
sig_hash (str)

bioimageflow.cache.processing_result_key(node_name, sig_hash)[source]¶

Return the result key for a ProcessingTool node.

Return type:

Parameters:

node_name (str)
sig_hash (str)

bioimageflow.cache.iter_dataframe_result_metadata(storage_path, known_node_signatures=None)[source]¶

Return DataFrameTool result metadata, inferring metadata-less records when possible.

Return type:

list[dict[str, Any]]

Parameters:

storage_path (str | Path)
known_node_signatures (dict[str, set[str]] | None)

bioimageflow.cache.iter_processing_result_metadata(storage_path, known_node_signatures=None)[source]¶

Return ProcessingTool result metadata, inferring metadata-less records when possible.

Return type:

list[dict[str, Any]]

Parameters:

storage_path (str | Path)
known_node_signatures (dict[str, set[str]] | None)

bioimageflow.cache.dataframe_lookup(storage_path, node_name, sig_hash)[source]¶

Load a DataFrameTool cache hit, or return None on miss.

Return type:

DataFrame | None

Parameters:

storage_path (str | Path)
node_name (str)
sig_hash (str)

bioimageflow.cache.dataframe_publish(storage_path, node_name, sig_hash, df)[source]¶

Publish a DataFrameTool result through the immutable record model.

Return type:

DataFrame

Parameters:

storage_path (str | Path)
node_name (str)
sig_hash (str)
df (DataFrame)

bioimageflow.cache.processing_prepare_attempt(storage_path, node_name, sig_hash)[source]¶

Create an attempt staging tree for a ProcessingTool node.

Return type:

tuple[str, str, Path, Path]

Parameters:

storage_path (str | Path)
node_name (str)
sig_hash (str)

bioimageflow.cache.processing_lookup(storage_path, node_name, sig_hash, path_columns, shared_array_columns=None, hydrate_assets=True)[source]¶

Load a ProcessingTool cache hit, or return None on miss.

Return type:

DataFrame | None

Parameters:

storage_path (str | Path)
node_name (str)
sig_hash (str)
path_columns (set[str])
shared_array_columns (set[str] | None)
hydrate_assets (bool)

bioimageflow.cache.processing_publish(storage_path, node_name, sig_hash, df, *, result_key, attempt_id, staging_dir, staging_assets_dir, path_columns, owned_path_columns, shared_array_columns=None, declared_owned_artifact_paths=None, declared_scalar_outputs=None)[source]¶

Publish a source ProcessingTool attempt as an immutable record.

Return type:

DataFrame

Parameters:

storage_path (str | Path)
node_name (str)
sig_hash (str)
df (DataFrame)
result_key (str)
attempt_id (str)
staging_dir (Path)
staging_assets_dir (Path)
path_columns (set[str])
owned_path_columns (set[str])
shared_array_columns (set[str] | None)
declared_owned_artifact_paths (Iterable[tuple[str, Any, str | PathLike[str]]] | None)
declared_scalar_outputs (Iterable[tuple[str, Any, Any]] | None)

Storage¶

Low-level cache storage primitives for the current cache/v1 storage layout.

Output/cache storage primitives.

The on-disk cache schema is versioned, but this module is the clean storage implementation used by the current runtime.

exception bioimageflow.storage.CacheCorruptionError[source]¶

Bases: RuntimeError

Raised when cache metadata points to corrupt or unsafe state.

class bioimageflow.storage.OutputViewCapability[source]¶

Bases: object

Structured result from probing one output-view materialization mode.

mode: str¶

supported: bool¶

code: Literal['ok', 'permission_denied', 'filesystem_unsupported', 'invalid_mode', 'io_error']¶

detail: str | None = None¶

__init__(mode, supported, code, detail=None)¶

Parameters:

mode (str)
supported (bool)
code (Literal['ok', 'permission_denied', 'filesystem_unsupported', 'invalid_mode', 'io_error'])
detail (str | None)

Return type:

None

bioimageflow.storage.canonical_json_bytes(value)[source]¶

Return canonical UTF-8 JSON bytes for hashing.

Return type:: bytes
Parameters:: value (Any)

bioimageflow.storage.make_result_key(material)[source]¶

Create a deterministic result key from canonical material.

Return type:: str
Parameters:: material (Any)

bioimageflow.storage.result_shard_parts(result_key)[source]¶

Return deterministic nested shard path parts for a result key.

Return type:: tuple[str, str]
Parameters:: result_key (str)

bioimageflow.storage.make_node_keys(names)[source]¶

Return storage-safe node keys, disambiguating normalized collisions.

Return type:: dict[str, str]
Parameters:: names (list[str])

bioimageflow.storage.validate_relative_posix_path(path)[source]¶

Validate and normalize a record-relative POSIX path.

Return type:: str
Parameters:: path (str)

bioimageflow.storage.canonical_scalar_payload(value)[source]¶

Return the canonical payload for scalar manifest metadata.

Return type:: dict[str, Any]
Parameters:: value (Any)

bioimageflow.storage.canonical_dataframe_digest(df, *, declared_columns=None, column_kinds=None)[source]¶

Return a stable digest for the supported dataframe surface.

Return type:

Parameters:

df (DataFrame)
declared_columns (Sequence[str] | None)
column_kinds (Mapping[str, str] | None)

bioimageflow.storage.asset_digest_and_size(path)[source]¶

Return deterministic size and digest metadata for a file or directory asset.

Return type:: tuple[int, str]
Parameters:: path (Path)

bioimageflow.storage.make_record_id(manifest_material)[source]¶

Create a record ID from content material, excluding execution metadata.

Return type:: str
Parameters:: manifest_material (dict[str, Any])

class bioimageflow.storage.RecordManifest[source]¶

Bases: object

Structured manifest helper for a reusable record.

result_key: str¶

record_id: str¶

dataframe_digest: str¶

outputs: list[dict[str, Any]]¶

schema: str = 'bioimageflow.cache.record.v1'¶

to_dict()[source]¶

Return type:: dict[str, Any]

classmethod from_dict(value)[source]¶

Return type:: RecordManifest
Parameters:: value (dict[str, Any])

validate(record_dir, *, expected_result_key=None)[source]¶

Return type:

Parameters:

record_dir (Path)
expected_result_key (str | None)

__init__(result_key, record_id, dataframe_digest, outputs, schema='bioimageflow.cache.record.v1')¶

Parameters:

result_key (str)
record_id (str)
dataframe_digest (str)
outputs (list[dict[str, Any]])
schema (str)

Return type:

None

class bioimageflow.storage.CurrentPointer[source]¶

Bases: object

Structured helper for current.json.

result_key: str¶

record_id: str¶

manifest: str¶

attempt_id: str¶

run_id: str¶

policy: str = 'first-valid'¶

schema: str = 'bioimageflow.cache.current.v1'¶

selected_at: str | None = None¶

to_dict()[source]¶

Return type:: dict[str, Any]

classmethod from_dict(value)[source]¶

Return type:: CurrentPointer
Parameters:: value (dict[str, Any])

__init__(result_key, record_id, manifest, attempt_id, run_id, policy='first-valid', schema='bioimageflow.cache.current.v1', selected_at=None)¶

Parameters:

result_key (str)
record_id (str)
manifest (str)
attempt_id (str)
run_id (str)
policy (str)
schema (str)
selected_at (str | None)

Return type:

None

class bioimageflow.storage.Storage[source]¶

Bases: object

Filesystem paths and guarded metadata updates for versioned storage.

__init__(storage_path)[source]¶

Parameters:: storage_path (str | Path)
Return type:: None

property cache_root: Path¶

property views_root: Path¶

property runs_root: Path¶

property latest_root: Path¶

property outputs_root: Path¶

probe_output_view_mode(mode)[source]¶

Probe an output-view mode on the workflow storage filesystem.

Return type:: OutputViewCapability
Parameters:: mode (str)

result_dir(result_key)[source]¶

Return type:: Path
Parameters:: result_key (str)

run_dir(run_id)[source]¶

Return type:: Path
Parameters:: run_id (str)

run_node_dir(run_id, node_key)[source]¶

Return type:

Parameters:

run_id (str)
node_key (str)

new_attempt_id()[source]¶

Return type:: str

write_run_metadata(run_id, *, workflow_identity, engine, status, target_nodes, started_at=None, completed_at=None)[source]¶

Write workflow-level run metadata.

Return type:

Parameters:

run_id (str)
workflow_identity (str)
engine (str)
status (str)
target_nodes (Sequence[str])
started_at (str | None)
completed_at (str | None)

write_run_node_result(run_id, node_key, *, result_key, record_id, cache_hit)[source]¶

Write a run-local view over the selected current record for one node.

Return type:

Parameters:

run_id (str)
node_key (str)
result_key (str)
record_id (str)
cache_hit (bool)

update_latest_node(node_key, run_id)[source]¶

Atomically point views/latest/<node-key> at a run-node view.

Return type:

Parameters:

node_key (str)
run_id (str)

update_latest_success_run(run_id)[source]¶

Atomically point views/runs/latest-success at a successful run view.

Return type:: Path
Parameters:: run_id (str)

materialize_run_outputs(run_id, mode)[source]¶

Materialize owned assets for one run under outputs/runs.

Return type:

list[Path]

Parameters:

run_id (str)
mode (str)

materialize_latest_node_outputs(node_key, mode)[source]¶

Materialize owned assets for one latest node pointer under outputs/latest.

Return type:

list[Path]

Parameters:

node_key (str)
mode (str)

materialize_latest_outputs(mode)[source]¶

Materialize owned assets for all latest node pointers under outputs/latest.

Return type:: list[Path]
Parameters:: mode (str)

latest_success_run_id()[source]¶

Return the latest successful run ID from views/runs if present.

Return type:: str | None

load_current(result_key)[source]¶

Return type:: CurrentPointer | None
Parameters:: result_key (str)

select_current_record(result_key, *, candidate_record_id, attempt_id, run_id)[source]¶

Return type:

CurrentPointer

Parameters:

result_key (str)
candidate_record_id (str)
attempt_id (str)
run_id (str)

Template¶

Output templating engine for ProcessingTool.

bioimageflow.template.get_output_templates(outputs_cls, inputs_cls, overrides=None)[source]¶

Extract templates from explicit Template defaults.

Path outputs without a Template default use the built-in default template. Non-path outputs may not declare Template defaults.

Return type:

dict[str, str]

Parameters:

outputs_cls (type[IOModel])
inputs_cls (type[IOModel])
overrides (dict[str, str] | None)

bioimageflow.template.validate_template(template_str, input_annotations)[source]¶

Check that all template references are valid. :rtype: None

{field.xxx} — field must exist in Inputs, xxx must be a valid property (stem, name, ext, exts).
{bare} — must be a special variable, a column: reference, or an input field. Unknown bare references emit a warning.

Parameters:

template_str (str)
input_annotations (dict[str, Any])

Return type:

None

bioimageflow.template.resolve_template(template_str, context)[source]¶

Resolve a template string using the provided context dict.

Context should include: - node_name, row_index, timestamp - Input field values (for .stem, .name, .ext, .exts properties) - column:<name> lookups

Return type:

Parameters:

template_str (str)
context (dict[str, Any])

Tool registry¶

Public tool-registry abstraction.

A ToolRegistry wraps load_versioned_package(), resolve_tool_class(), and serialize_input_schema() / serialize_output_schema() behind a single object that GUIs and other consumers can use without rebuilding metadata serialization or package-resolution layers themselves.

The registry deliberately separates install (a slow, network-bound side effect) from register (a fast, in-process index lookup):

ToolRegistry.install_package() installs a package into the tool store, doing the network work.
ToolRegistry.register_package() only loads what is already installed and indexes its tools — never installs.

GUIs that validate on every keystroke must call register_package on hot paths and install_package only from a user-initiated action.

Workflow exports may also carry embedded custom tool modules. GUIs can call ToolRegistry.register_workflow() to discover those local tools for the specific workflow being edited without promoting them to a versioned tool package.

class bioimageflow.registry.ToolMetadata[source]¶

Bases: object

Public, serialized description of a tool class.

Produced by ToolRegistry.register_package() for every BaseTool subclass found in the loaded package.

Variables:

package – The tool’s package name (e.g. "my_tools").
version – Pinned package version (e.g. "1.2.3").
module – Canonical module path (e.g. "my_tools.alpha").
class_name – The class name as written in the source.
inputs_schema – Output of serialize_input_schema() for this class.
outputs_schema – Output of serialize_output_schema() for this class.
display_name – Human-readable label declared on the class, or the class name.
tags – Free-form tags declared on the class (empty list if none).

package: str¶

version: str¶

module: str¶

class_name: str¶

inputs_schema: dict[str, Any]¶

outputs_schema: dict[str, Any]¶

display_name: str¶

tags: tuple[str, ...]¶

__init__(package, version, module, class_name, inputs_schema, outputs_schema, display_name, tags=<factory>)¶

Parameters:

package (str)
version (str)
module (str)
class_name (str)
inputs_schema (dict[str, Any])
outputs_schema (dict[str, Any])
display_name (str)
tags (tuple[str, ...])

Return type:

None

class bioimageflow.registry.ToolRegistry[source]¶

Bases: object

Stateful index of tool classes loaded from packages or workflows.

__init__(*, store_path=None)[source]¶

Parameters:: store_path (Path | None)
Return type:: None

install_package(name, version)[source]¶

Install a versioned package into the tool store.

This is the slow, network-bound side effect — it does not load or index anything. Call register_package() separately once the install completes.

name is the import name (the module a workflow expects to import). ensure_installed infers the PyPI name from this by replacing underscores with hyphens, matching the convention used by Workflow.from_dict()’s auto-installer.

Return type:

Parameters:

name (str)
version (str)

register_package(name, version)[source]¶

Load an already installed package and index its tools.

Returns the metadata for every BaseTool subclass discovered in the package. Raises FileNotFoundError if the package is not present in the store — the caller must install it first via install_package() (or skip registration on the hot path).

Return type:

list[ToolMetadata]

Parameters:

name (str)
version (str)

register_workflow(workflow)[source]¶

Index custom tools carried by a workflow.

workflow may be either a live bioimageflow.Workflow instance or an exported workflow dict. Live workflows are inspected recursively; exported archives carry one deduplicated custom_sources table. Package tools referenced by the workflow are not loaded or installed here; use register_package() for those.

Return type:: list[ToolMetadata]
Parameters:: workflow (Any)

get_class(class_name, *, package=None, version=None, module=None)[source]¶

Return a registered class, or None if not registered.

When several package versions expose the same class name, callers can pass package / version / module to select the intended class. A name-only lookup keeps the historical behavior and returns the most recently registered matching class.

Return type:

type | None

Parameters:

class_name (str)
package (str | None)
version (str | None)
module (str | None)

get_metadata(class_name, *, package=None, version=None, module=None)[source]¶

Return registered ToolMetadata, or None.

Return type:

ToolMetadata | None

Parameters:

class_name (str)
package (str | None)
version (str | None)
module (str | None)

list_tools()[source]¶

Return all registered tool metadata, in insertion order.

Return type:: list[ToolMetadata]

forget(class_name, *, package=None, version=None, module=None)[source]¶

Drop matching classes from the registry. No-op if none match.

Return type:

Parameters:

class_name (str)
package (str | None)
version (str | None)
module (str | None)

Workflow session¶

Incremental editing session for GUI clients.

A WorkflowSession is a parallel data model to Workflow that holds the wire-format dict and exposes mutation operations. It materializes a Workflow on demand and caches it across edits, selectively rebuilding only when structural changes (add/remove node/edge) demand it. Edits that don’t change the graph topology (set_constant, set_enabled) mutate the cached workflow in place — so a constant edit followed by validate() does not re-resolve any tool class.

Why a separate class? Workflow builds nodes eagerly during construction, with _upstream_nodes references and column bindings wired at __init__ time. Retrofitting incremental mutation onto that model would require invasive changes to Node. A dict-backed session, materialized to a Workflow only when needed, is both simpler and matches what GUIs actually want to send over the wire.

class bioimageflow.session.WorkflowSession[source]¶

Bases: object

A mutable, dict-backed editing session for a workflow.

The session is the canonical state. Call to_workflow() to obtain a built Workflow for execution, or to_dict() to snapshot the wire format.

__init__(data=None, *, registry=None, storage_path=None)[source]¶

Parameters:

data (dict[str, Any] | None)
registry (Any | None)
storage_path (str | None)

Return type:

None

add_node(node)[source]¶

Append a node entry to the session.

node is a wire-format dict with at least name and tool identification keys for either a tool or recursive workflow node.

Return type:: None
Parameters:: node (dict[str, Any])

remove_node(name)[source]¶

Remove a node and all edges touching it.

Return type:: None
Parameters:: name (str)

add_edge(edge)[source]¶

Append one strict column or DataFrame edge record.

Return type:: None
Parameters:: edge (dict[str, Any])

remove_edge(edge_id)[source]¶

Remove an edge by its opaque id field.

Return type:: None
Parameters:: edge_id (str)

set_constant(node, field, value)[source]¶

Update a constant binding on a node.

This is a non-structural edit: the cached Workflow (if any) is mutated in place so that subsequent to_workflow() / validate() / plan() calls do not re-resolve any tool class.

Return type:

Parameters:

node (str)
field (str)
value (Any)

set_enabled(node, enabled)[source]¶

Toggle a node’s enabled flag.

Non-structural: the cached workflow is updated in place.

Return type:

Parameters:

node (str)
enabled (bool)

property nodes: dict[str, dict[str, Any]]¶

property edges: list[dict[str, Any]]¶

property errors: list[ValidationError]¶: Cached errors from the last validate() call (or empty).

property failed_nodes: dict[str, ValidationError]¶: Failed nodes from the last to_workflow() build.

to_dict()[source]¶

Return the wire-format snapshot. The returned dict is a copy.

Return type:: dict[str, Any]

to_workflow()[source]¶

Return a built Workflow, caching the result.

Uses Workflow.from_dict(partial=True, validate_only=True) so per-node failures are captured in Workflow.failed_nodes rather than raising.

Return type:: Workflow

validate()[source]¶

Return the validation errors for the current state.

Cached across non-structural edits.

Return type:: list[ValidationError]

plan()[source]¶

Return a fresh Workflow.plan() for the current state.

Return type:: dict[str, NodePlan]

classmethod from_dict(data, *, registry=None)[source]¶

Return type:

WorkflowSession

Parameters:

data (dict[str, Any])
registry (Any | None)

Workflow nodes¶

A workflow invocation embedded as a node in another workflow.

class bioimageflow.workflow_node.WorkflowNode[source]¶

Bases: Node

An independent structural snapshot of a workflow invocation.

__init__(workflow, *, name=None, bindings=None)[source]¶

Parameters:

workflow (Workflow)
name (str | None)
bindings (dict[str, Any] | None)

Return type:

None

property internal_nodes: list[Node]¶

bind_port(port_id, value)[source]¶

Bind one stable input port and apply it to every declared target.

Return type:

Parameters:

port_id (str)
value (Any)

output_name_for_id(port_id)[source]¶

Return type:: str
Parameters:: port_id (str)

Wetlands configuration¶

Orchestrator-side Wetlands environment management.

Provides a single shared EnvironmentManager instance used by both the execution engine (tool dispatch) and the tool loader (package install). Call configure_wetlands() once at the top of your script to set paths; everything else picks up the same instance automatically.

bioimageflow.env_manager.configure_wetlands(**config)[source]¶

Set Wetlands configuration for the entire process.

Must be called before any tool loading or workflow execution. Subsequent calls are ignored with a warning if the manager is already initialized.

Return type:: None
Parameters:: config (Any)

Common parameters:: wetlands_instance_path: Path for Wetlands state (logs, pixi). conda_path: Path to the pixi or micromamba installation. main_conda_environment_path: Main conda env for dep checking. debug: Enable debugpy in worker processes. Use BIOIMAGEFLOW_USE_LOCAL_CORE=1 to inject the local editable bioimageflow-core checkout into worker environments.

bioimageflow.env_manager.get_shared_environment_manager(**config)[source]¶

Return the process-wide Wetlands EnvironmentManager.

On first call, creates the manager using configuration from configure_wetlands() (merged with any config kwargs passed here). Subsequent calls return the cached instance.

Return type:: EnvironmentManager
Parameters:: config (Any)

class bioimageflow.env_manager.WetlandsEnvManager[source]¶

Bases: object

Manages Wetlands environments for ProcessingTool execution.

Creates environments lazily on first use.
Caches launched environments by name.
Auto-injects bioimageflow-core into every environment’s pip deps.
Provides dispatch helpers that route calls through the Wetlands proxy.

__init__(wetlands_instance_path=None, conda_path=None, main_conda_environment_path=None, bioimageflow_core_dependency=None, use_local_bioimageflow_core=None, **kwargs)[source]¶

Parameters:

wetlands_instance_path (Path | None)
conda_path (str | None)
main_conda_environment_path (str | None)
bioimageflow_core_dependency (Any | None)
use_local_bioimageflow_core (bool | None)
kwargs (Any)

Return type:

None

get_or_create(env_spec, max_workers=1, worker_env=None, worker_timeout=None)[source]¶

Get or create a Wetlands environment.

Wetlands validates same-name environment recipe reuse. On first creation, env.launch(max_workers=..., worker_env=..., worker_timeout=...) is called. Subsequent calls with a different max_workers or worker_timeout log a warning but do not re-launch.

This method is thread-safe and may be called concurrently.

Return type:

Parameters:

env_spec (EnvironmentSpec)
max_workers (int)
worker_env (Any | None)
worker_timeout (float | None)

submit_process_batch(env_spec, tool_file_path, tool_class_name, arguments_dicts, context_dict, max_workers=1, worker_env=None, worker_timeout=None)[source]¶

Submit a batch call via env.submit(). Returns a Task.

Return type: