spacr.ports

Typed input/output ports: what each spaCR module consumes and produces.

Every pipeline module in spaCR reads a folder laid out by the module before it and writes a folder the module after it expects. Until now that contract lived only in the code that happened to open the path, so “can Measure run here?” could only be answered by starting Measure and watching it fail twenty minutes in.

This module makes the contract data. Each module declares:

• what it consumes — a Port per input, carrying a kind ("merged-arrays", "measurements-db", "crops", …), the path it lives at relative to the project root, and a shape/table contract;

• what it produces — the same, for its outputs.

check_ready() then answers before a run starts whether a module can run, and when it cannot, says why and what to do about it. The answer is a list of spacr.validate.Problem — the same type the settings pre-flight returns — so a caller can print both through spacr.validate.format_report().

Path conventions are not invented here. They are the ones already in the code: <root>/merged/*.npy and <root>/measurements/measurements.db (spacr.core.preprocess_generate_masks()), the “src may already be the merged folder” hop (spacr.crops._looks_like_experiment_root, spacr.validate._resolve_merged_dir), the src / orig / consolidated search order for raw images (spacr.validate._scan_raw_images), <root>/data/**/*_png (spacr.measure), <root>/model/... (spacr.deep_spacr), <root>/results/... (spacr.ml). Module keys are spacr.validate.APP_FUNCTIONS keys, so check_ready("measure", s) and validate_settings(s, "measure") speak the same language.

Public API

Port / ShapeContract / ModulePorts

The declarations.

PORTS, module_ports, register_module_ports, known_modules

The registry and its extension seam.

project_root, resolve_port, declared_inputs, declared_outputs

Path resolution.

check_ready, format_readiness, describe_ports

“Can module X run, and if not, why not?”

producers_of, consumers_of, next_modules, upstream_modules

The module graph, for auto-chaining and “continue to next step”.

The module imports only the standard library plus two dependency-light spaCR modules (spacr.validate, spacr.resume). No numpy, torch, pandas or Cellpose: a readiness check has to cost less than the run it is protecting.

Exceptions

UnknownModule

No ports are declared for the requested module key.

Classes

ModulePorts	Everything one module reads and writes.
Port	One thing a module consumes or produces.
Readiness	Whether a module can run, and why not when it cannot.
ResolvedPort	A Port bound to a project root and looked up on disk.
ShapeContract	What an array at a port has to look like.

Functions

check_ready(→ Readiness)	Answer "can module run here?" before anything is loaded.
consumers_of(→ Tuple[str, ...])	Return the modules that consume kind, sorted.
declared_inputs(→ Tuple[ResolvedPort, ...])	Return this module's consumed ports, resolved against the project.
declared_outputs(→ Tuple[ResolvedPort, ...])	Return this module's produced ports, resolved against the project.
describe_ports(→ str)	Render one module's declared contract as text.
format_readiness(→ str)	Render a Readiness as a block of text for a user.
known_modules(→ Tuple[str, ...])	Return every module key with declared ports, sorted.
module_ports(→ ModulePorts)	Return the port declaration for module.
next_modules(→ Tuple[str, ...])	Return the modules that can run on what module produces.
port_problems(→ Tuple[spacr.validate.Problem, ...])	Validate one input port after resolving it against root.
producers_of(→ Tuple[str, ...])	Return the modules that produce kind, sorted.
project_root(→ str)	Return the absolute project root a run works in.
register_module_ports(→ ModulePorts)	Add or replace one module's port declaration.
resolve_port(→ ResolvedPort)	Bind port to root and look it up on disk.
upstream_modules(→ Tuple[str, ...])	Return the modules that produce what module requires.

Module Contents

exception spacr.ports.UnknownModule

Bases: KeyError

No ports are declared for the requested module key.

Initialize self. See help(type(self)) for accurate signature.

class spacr.ports.ModulePorts

Everything one module reads and writes.

Parameters:

• key – the module key, as in spacr.validate.APP_FUNCTIONS.

• consumes – inputs.

• produces – outputs.

• summary – one line describing the module, for describe_ports().

port(role: str) → Port

Return the consumed or produced port with role.

Parameters:

role – the port’s role name.

Raises:

KeyError – when this module declares no such role.

class spacr.ports.Port

One thing a module consumes or produces.

Parameters:

• kind – the vocabulary term, e.g. MERGED_ARRAYS. Ports match across modules by kind, which is what makes auto-chaining possible.

• role – short name, unique within a module, e.g. "merged". It labels the problem messages and the registered artifact.

• path – location relative to the project root. "" is the root.

• pattern – glob applied inside path. Empty means path names a single file or folder that must exist. ** is honoured, and | separates alternatives, all of which are searched.

• required – False for an output that may legitimately be absent (masks/ after cleanup) or an input a module can do without.

• min_count – how many matches pattern must yield.

• extensions – when set, matches are kept only if their lowercased name ends with one of these.

• shape – array contract, for .npy ports.

• tables – SQLite tables that must exist and hold at least one row. Only meaningful for MEASUREMENTS_DB ports.

• description – one line, shown by describe_ports().

relative() → str

Return the port’s location relative to the project root, for display.

class spacr.ports.Readiness

Whether a module can run, and why not when it cannot.

Parameters:

• module – the canonical module key that was checked.

• root – the project root it was checked against.

• ok – True when nothing blocking was found. bool(readiness) is the same answer.

• problems – spacr.validate.Problem instances, errors and warnings mixed, with the port role in Problem.setting.

• satisfied – roles of the inputs that were found and accepted.

• inputs – artifact ids backing the satisfied inputs, when the check was given a registry.

__bool__() → bool

True when the module can run.

__str__() → str

The full report; see format_readiness().

property errors: Tuple[spacr.validate.Problem, ...]

The blocking problems.

property reason: str

One human-readable line saying why the answer is what it is.

property warnings: Tuple[spacr.validate.Problem, ...]

The non-blocking problems.

class spacr.ports.ResolvedPort

A Port bound to a project root and looked up on disk.

Parameters:

• port – the declaration.

• root – the project root it was resolved against.

• target – absolute path of the file or folder the port names.

• paths – absolute paths the port’s pattern matched, sorted. Empty for a pattern-less port.

• exists – whether the port is present at all.

• count – number of matches; 1 for a pattern-less port that exists.

property kind: str

The port’s kind, for convenience.

property location: str

the file, or the folder.

Type:

The single path this port stands for

property role: str

The port’s role, for convenience.

class spacr.ports.ShapeContract

What an array at a port has to look like.

Checked from the .npy header alone (see spacr.resume.read_npy_header()), so validating a thousand 100-megabyte fields stays cheap — and it catches truncation, which np.load only discovers after allocating the whole array.

Parameters:

• ndim – required number of axes, or None for any.

• min_planes – smallest acceptable length of the last axis. A merged array needs at least one image plane and one mask plane.

• dtype – required numpy dtype string such as "uint16", or "" for any.

describe() → str

Return a one-line human description of the contract.

spacr.ports.check_ready(module: str, settings: str | Mapping[str, Any] | None = None, *, root: str = '', registry: Any = None, sample: int = 3) → Readiness

Answer “can module run here?” before anything is loaded.

Every declared input port is resolved against the project root and looked up on disk: present, plentiful enough, the right array shape, and — for a measurements database — carrying the tables the module reads. A missing optional port produces a warning instead of an error.

When a spacr.artifacts.Registry is supplied, each satisfied input is also matched against the registry, so the ids of the artifacts being consumed come back in Readiness.inputs and an input the registry reports as stale is added as a warning.

Parameters:

• module – module key or alias, as in spacr.validate.APP_FUNCTIONS.

• settings – the settings dict about to be run, or a source path.

• root – explicit project root, overriding settings.

• registry – optional spacr.artifacts.Registry, for provenance and staleness.

• sample – how many arrays per port to check the shape of. Reading a .npy header is cheap but not free, and a bad run is bad from its first field.

Returns:

a Readiness; bool(result) is the yes/no answer and Readiness.reason is the sentence to show a user.

Raises:

UnknownModule – when nothing is declared for module.

spacr.ports.consumers_of(kind: str, *, required_only: bool = False) → Tuple[str, ...]

Return the modules that consume kind, sorted.

Parameters:

• kind – a vocabulary term such as MEASUREMENTS_DB.

• required_only – only modules that need it, not those that can optionally use it.

spacr.ports.declared_inputs(module: str, settings: str | Mapping[str, Any] | None = None, *, root: str = '') → Tuple[ResolvedPort, ...]

Return this module’s consumed ports, resolved against the project.

Parameters:

• module – module key or alias.

• settings – settings dict or source path, used to derive the root.

• root – explicit project root, overriding settings.

Raises:

UnknownModule – when nothing is declared for module.

spacr.ports.declared_outputs(module: str, settings: str | Mapping[str, Any] | None = None, *, root: str = '') → Tuple[ResolvedPort, ...]

Return this module’s produced ports, resolved against the project.

What a finished run should have written — the list spacr.artifacts.register_run_outputs() walks.

Parameters:

• module – module key or alias.

• settings – settings dict or source path, used to derive the root.

• root – explicit project root, overriding settings.

Raises:

UnknownModule – when nothing is declared for module.

spacr.ports.describe_ports(module: str) → str

Render one module’s declared contract as text.

Parameters:

module – module key or alias.

Returns:

a multi-line description of what it consumes and produces.

Raises:

UnknownModule – when nothing is declared for module.

spacr.ports.format_readiness(readiness: Readiness) → str

Render a Readiness as a block of text for a user.

Parameters:

readiness – the result of check_ready().

Returns:

a multi-line report; errors first, each with its fix line.

spacr.ports.known_modules() → Tuple[str, ...]

Return every module key with declared ports, sorted.

spacr.ports.module_ports(module: str) → ModulePorts

Return the port declaration for module.

Accepts every alias spacr.validate.APP_ALIASES accepts, so "measure_crop" and "measure" resolve to the same declaration.

Parameters:

module – module key or alias.

Returns:

the ModulePorts declared for it.

Raises:

UnknownModule – when nothing is declared for it.

spacr.ports.next_modules(module: str) → Tuple[str, ...]

Return the modules that can run on what module produces.

The answer to “this run finished — what would you like to do next?”.

Parameters:

module – module key or alias.

Raises:

UnknownModule – when nothing is declared for it.

spacr.ports.port_problems(port: Port, root: str, *, sample: int = 3) → Tuple[spacr.validate.Problem, ...]

Validate one input port after resolving it against root.

This is the per-port validation used by check_ready() and is suitable for callers that have a Port without a module key. It returns the same missing-input, count, shape, and table problems as the full readiness check.

Parameters:

• port – the declaration.

• root – absolute project root to resolve it against.

• sample – how many arrays to check the shape of; 0 skips the check.

Returns:

spacr.validate.Problem instances, possibly empty.

spacr.ports.producers_of(kind: str) → Tuple[str, ...]

Return the modules that produce kind, sorted.

Parameters:

kind – a vocabulary term such as MERGED_ARRAYS.

spacr.ports.project_root(settings_or_src: str | Mapping[str, Any] | None, module: str = '') → str

Return the absolute project root a run works in.

Reuses the conventions already in the code rather than inventing one:

• a list of sources means several plates, and the first names the first project (spacr.core.preprocess_generate_masks() loops over settings['src']);

• a src that already ends in merged names the merged folder, not the project — the same hop spacr.crops._looks_like_experiment_root and spacr.validate._resolve_merged_dir make;

• modules whose folder is not src are looked up in ROOT_KEYS and spacr.validate.ALT_SRC_KEYS.

Parameters:

• settings_or_src – a settings dict, a path, or None.

• module – module key or alias, used only to pick the settings key.

Returns:

an absolute path, or "" when no source is set.

spacr.ports.register_module_ports(ports: ModulePorts, *, overwrite: bool = False) → ModulePorts

Add or replace one module’s port declaration.

The seam a plugin — or a module written after this one — uses to join the graph, so PORTS never has to be edited by hand.

Parameters:

• ports – the declaration. Its key is lowercased before storage.

• overwrite – allow replacing an existing declaration. Off by default, so two plugins claiming one key is an error rather than a silent last-one-wins.

Returns:

the stored declaration, for chaining.

Raises:

ValueError – when the key is empty, when it is already declared and overwrite is False, or when two of its ports share a role.

spacr.ports.resolve_port(port: Port, root: str) → ResolvedPort

Bind port to root and look it up on disk.

Parameters:

• port – the declaration.

• root – absolute project root.

Returns:

a ResolvedPort. A missing path is not an exception — absence is an answer, and check_ready() is what judges it.

spacr.ports.upstream_modules(module: str) → Tuple[str, ...]

Return the modules that produce what module requires.

The answer to “Measure needs merged arrays — who makes those?”.

Parameters:

module – module key or alias.

Raises:

UnknownModule – when nothing is declared for it.