spacr.pipeline_v2

Streaming mask pipeline (v2).

Replaces the multi-copy disk chain that preprocess_generate_masks has run since day one:

originals

→ renamed + split into channel folders → orig/ backup → per-channel npy → batch npz on disk → cellpose → per-field mask npy → concatenated into merged/

…with a two-pass streaming pipeline that keeps only what the downstream measure module actually reads:

Pass 1 — assemble

walk originals, parse metadata regex, build one npy stack per field with all image channels in the C axis. Emit filename_map.csv recording every original → stack mapping.

Pass 2 — segment

stream the plate in batches of N fields, hand each batch to Cellpose, append the mask channels to the SAME stack file. Optional intermediate NPZ is memory-only (never touches disk) unless keep_npz=True.

Output — merged/ folder holds one file per field, each shape (H, W, C_image + C_mask) in uint16, plus:

channel_order.json {“channels”: […]} filename_map.csv original path, plate/well/field/…, stack idx

Public API:

from spacr.pipeline_v2 import (
    FilenameMapper, stream_originals_to_stack,
    stream_masks_from_stack, run_v2,
)

# High-level (one call):
run_v2(src_folder, channels=(0,1,2,3), model="cyto", diameter=60)

# Low-level (two passes, run each explicitly):
mapper = FilenameMapper.discover(src_folder,
                                   metadata_type="cellvoyager")
stacks = stream_originals_to_stack(src_folder, mapper, channels=(0,1,2,3))
stream_masks_from_stack(stacks, model="cyto", diameter=60)

This module is opt-in for one release cycle. Once the follow-up commit wires it as the default in spacr.core.preprocess_generate_masks() the whole disk chain above collapses to merged/ alone.

Attributes

LOG

Classes

FilenameMapper	Walks a folder of microscopy images, parses each filename's
FilenameRecord	One entry in the filename map.
StackFile	One field's on-disk stack: merged/stack_<id>.npy with

Functions

run_v2(, channel_names, model_name, ...)	Run the entire v2 pipeline against src. Convenience wrapper.
stream_masks_from_stack(, diameter, batch_fields, ...)	Batch the field stacks through Cellpose, then append the mask
stream_originals_to_stack(, channel_names, dst)	Write one merged/stack_<field>.npy per field.

Module Contents

class spacr.pipeline_v2.FilenameMapper(records: List[FilenameRecord], metadata_type: str, regex: str)

Walks a folder of microscopy images, parses each filename’s metadata via a regex, and records the mapping to a per-plate CSV.

The CSV is written next to the merged/ folder (at the plate root) so users can Excel-open filename_map.csv and see the original path of every image in the run.

Variables:

• records – list of FilenameRecord in file-system order.

• metadata_type – which regex was used ("cellvoyager" / "yokogawa" / "custom").

• regex – compiled regex pattern that matched.

by_field() → Dict[str, List[FilenameRecord]]

Group records by stack_field_id — one entry per field, with one record per channel inside.

classmethod discover(src: pathlib.Path, metadata_type: str = 'auto', custom_regex: str | None = None, exts: Sequence[str] = ('.tif', '.tiff', '.png', '.jpg', '.jpeg')) → FilenameMapper

Scan src for images + parse each name with the metadata regex. Falls back through cellvoyager → yokogawa on metadata_type="auto".

Parameters:

• src – folder to scan (not recursive; we expect images at the top level as the current spacr layout does).

• metadata_type – "auto" / "cellvoyager" / "yokogawa" / "custom". When "custom", custom_regex must be given.

• custom_regex – user-supplied regex; required for metadata_type="custom".

• exts – image file extensions to include.

Returns:

a populated FilenameMapper.

Raises:

ValueError – when no images are found or no regex fits.

field_ids() → List[str]

Return the sorted list of unique stack_field_id values.

classmethod load_csv(path: pathlib.Path) → FilenameMapper

Rehydrate a mapper from a previously-saved CSV.

save_csv(path: pathlib.Path) → pathlib.Path

Write the mapping to path as a CSV that Excel opens cleanly. One row per (original image, resulting stack slot).

metadata_type

records

regex

class spacr.pipeline_v2.FilenameRecord

One entry in the filename map.

Variables:

• original_path – absolute path to the source image on disk.

• plate – plate id parsed from the filename.

• well – well id parsed from the filename.

• field – field index parsed from the filename.

• channel – channel index parsed from the filename.

• time – time index parsed from the filename (defaults to 1).

• z – z-slice index parsed from the filename (defaults to 1).

• stack_field_id – the field id used in merged/stack_<X>.npy.

channel: int

field: int

original_path: str

plate: str

stack_field_id: str = ''

time: int = 1

well: str

z: int = 1

class spacr.pipeline_v2.StackFile

One field’s on-disk stack: merged/stack_<id>.npy with shape (H, W, C).

Populated by stream_originals_to_stack() before Cellpose runs (C = image channels only). After stream_masks_from_stack() the same file has additional mask channels appended.

channels: List[str]

field_id: str

path: pathlib.Path

shape: Tuple[int, int, int]

spacr.pipeline_v2.run_v2(src: pathlib.Path, channels: Sequence[int] = (0, 1, 2, 3), channel_names: Sequence[str] | None = None, model_name: str = 'cyto', channels_for_cellpose: Sequence[int] = (0, 0), diameter: float | None = None, batch_fields: int = 8, metadata_type: str = 'auto', custom_regex: str | None = None, keep_npz: bool = False, cellprob_threshold: float = 0.0, flow_threshold: float = 0.4, min_size: int = 15, resample: bool = True, postprocess_settings: Dict[str, Any] | None = None, object_type: str = 'cell') → Dict[str, Any]

Run the entire v2 pipeline against src. Convenience wrapper.

Equivalent to:

mapper = FilenameMapper.discover(src, metadata_type, custom_regex)
stacks = stream_originals_to_stack(src, mapper, channels, channel_names)
stream_masks_from_stack(stacks, model_name, channels_for_cellpose,
                        diameter, batch_fields, keep_npz=keep_npz)

Returns:

dict with mapper (FilenameMapper), stacks (list of StackFile), and dst (Path to merged/).

spacr.pipeline_v2.stream_masks_from_stack(stacks: List[StackFile], model_name: str = 'cyto', channels_for_cellpose: Sequence[int] = (0, 0), diameter: float | None = None, batch_fields: int = 8, mask_channel_name: str = 'mask', keep_npz: bool = False, npz_dir: pathlib.Path | None = None, cellprob_threshold: float = 0.0, flow_threshold: float = 0.4, min_size: int = 15, resample: bool = True, postprocess_settings: Dict[str, Any] | None = None, object_type: str = 'cell') → List[StackFile]

Batch the field stacks through Cellpose, then append the mask channel(s) to the SAME npy files.

Parameters:

• stacks – list produced by stream_originals_to_stack().

• model_name – Cellpose model to use ("cyto", "nuclei", …).

• channels_for_cellpose – Cellpose’s channels= argument — e.g. [0, 0] for grayscale, [2, 1] for green cyto + blue nucleus.

• diameter – expected object diameter in px (None → Cellpose auto).

• batch_fields – how many field stacks to load into memory at once. Larger = faster but more RAM.

• mask_channel_name – human name to record for the appended mask channel (default "mask").

• keep_npz – when True, write the intermediate memory batch as an NPZ file to npz_dir for debugging. Deleted after the batch runs unless this flag is set.

• npz_dir – where to write the (optional) intermediate NPZ files. Defaults to a scratch subfolder under the stack folder.

Returns:

the same list, with each StackFile.shape / .channels updated to reflect the appended mask channel.

spacr.pipeline_v2.stream_originals_to_stack(src: pathlib.Path, mapper: FilenameMapper, channels: Sequence[int] = (0, 1, 2, 3), channel_names: Sequence[str] | None = None, dst: pathlib.Path | None = None) → List[StackFile]

Write one merged/stack_<field>.npy per field.

Reads originals directly (no rename-into-channel-folders step), stacks the selected channels along the C axis, and writes one npy per field. Also emits a channel_order.json sidecar describing which C-index holds which channel.

Parameters:

• src – plate folder containing the original images.

• mapper – FilenameMapper produced from src.

• channels – which channel numbers (as parsed from filenames) to include, in the order they should occupy the C axis.

• channel_names – human names for those channels (must match channels length). Default: ["ch0", "ch1", …].

• dst – override the output folder; defaults to <src>/merged.

Returns:

list of StackFile, one per field written.

spacr.pipeline_v2.LOG