linumpy.io.slice_config#

Shared helpers for reading, writing and stamping slice_config.csv.

slice_config.csv is the single per-slice trace file threaded through the reconstruction pipeline. Each stage that makes a per-slice decision (quality assessment, rehoming correction, auto-exclusion, missing-slice interpolation, …) stamps its flag columns via this module and hands the enriched file to the next stage.

Only pipeline-decision columns live here; raw metrics belong in the pipeline report and per-stage diagnostics JSON.

Concurrency model#

This module does not implement any file locking. Safe concurrent use depends on the upstream Nextflow pipeline’s channel discipline:

  • Every process receives slice_config.csv as an immutable input staged into its own work directory. Nothing reads and writes the same file at the same time.

  • Per-slice stages (interpolation, pairwise registration, …) emit per-slice fragment files (slice_z{NN}_manifest.csv). Those fragments are collected and merged sequentially in a single downstream process (finalise_interpolation), so the CSV writer always runs on a single worker.

  • Stamping helpers (stamp() / merge_fragments()) always produce a new CSV at slice_config_out rather than updating in place, so a reader on the old version is never in a torn state.

If you ever need to call these helpers outside of Nextflow (e.g. ad-hoc scripts running in parallel), make sure each writer targets a distinct output path; otherwise the last writer wins.

Attributes#

CANONICAL_COLUMNS

TRUE_STRINGS

FALSE_STRINGS

Functions#

normalize_slice_id(slice_id)

Return slice_id as a two-digit zero-padded string ("01", "17").

read(path)

Read slice_config.csv; return slice_id -> row with normalized ids.

read_header(path)

Return the header row of path (empty list if file has no header).

write(path, rows[, extra_columns])

Atomically write rows to path.

stamp(path_in, path_out, slice_id, **flags)

Stamp a single slice: read path_in, update slice_id with flags, write to path_out.

stamp_many(path_in, path_out, updates)

Stamp multiple slices at once.

merge_fragments(path_in, fragment_paths, path_out[, ...])

Merge per-slice CSV fragments into path_in and write to path_out.

filter_slices_to_use(path)

Return the set of slice IDs whose use column is truthy.

get_flag(row, column[, default])

Return a boolean flag from a config row (default when absent/empty).

is_interpolated(path, slice_id)

Return True if slice_id is flagged as interpolated in path.

force_skip_slices(path)

Return slice IDs that stacking should treat as motor-only (force-skip their pairwise transforms).

Module Contents#

linumpy.io.slice_config.CANONICAL_COLUMNS: list[str] = ['slice_id', 'use', 'exclude_reason', 'quality_score', 'galvo_confidence', 'galvo_fix', 'notes',...[source]#
linumpy.io.slice_config.TRUE_STRINGS[source]#
linumpy.io.slice_config.FALSE_STRINGS[source]#
linumpy.io.slice_config.normalize_slice_id(slice_id)[source]#

Return slice_id as a two-digit zero-padded string ("01", "17").

Accepts int / str / float (“1.0”) inputs. Falls back to str(slice_id).strip() for non-numeric ids.

Parameters:

slice_id (object)

Return type:

str

linumpy.io.slice_config.read(path)[source]#

Read slice_config.csv; return slice_id -> row with normalized ids.

Raises FileNotFoundError if the file does not exist. Row values are kept as strings (CSV native); use get_flag() for bool coercion.

Parameters:

path (pathlib.Path)

Return type:

collections.OrderedDict[str, dict[str, str]]

linumpy.io.slice_config.read_header(path)[source]#

Return the header row of path (empty list if file has no header).

Parameters:

path (pathlib.Path)

Return type:

list[str]

linumpy.io.slice_config.write(path, rows, extra_columns=())[source]#

Atomically write rows to path.

  • The header always starts with CANONICAL_COLUMNS (in that order); any extra columns come after. Missing canonical columns are emitted empty.

  • Rows are sorted by slice_id.

  • slice_id is normalised to a 2-digit string.

Parameters:
Return type:

None

linumpy.io.slice_config.stamp(path_in, path_out, slice_id, **flags)[source]#

Stamp a single slice: read path_in, update slice_id with flags, write to path_out.

New slice rows are appended with use=false when the row is absent.

Parameters:
Return type:

None

linumpy.io.slice_config.stamp_many(path_in, path_out, updates)[source]#

Stamp multiple slices at once.

updates maps slice_id -> {column: value}. Unknown slices are appended with use=false unless the caller supplies a use key.

Parameters:
Return type:

None

linumpy.io.slice_config.merge_fragments(path_in, fragment_paths, path_out, column_map=None)[source]#

Merge per-slice CSV fragments into path_in and write to path_out.

Each fragment is a small CSV with at least a slice_id column. Columns from the fragment are stamped onto the matching slice row, renamed via column_map if provided ({fragment_col: target_col}).

Fragments that reference slices absent from the base config add new rows (use=false).

Parameters:
Return type:

None

linumpy.io.slice_config.filter_slices_to_use(path)[source]#

Return the set of slice IDs whose use column is truthy.

When slice_config.csv is missing this raises FileNotFoundError – callers should guard on path.exists() or pass an optional path themselves.

Parameters:

path (pathlib.Path)

Return type:

set[str]

linumpy.io.slice_config.get_flag(row, column, default=False)[source]#

Return a boolean flag from a config row (default when absent/empty).

Parameters:
Return type:

bool

linumpy.io.slice_config.is_interpolated(path, slice_id)[source]#

Return True if slice_id is flagged as interpolated in path.

Parameters:
Return type:

bool

linumpy.io.slice_config.force_skip_slices(path)[source]#

Return slice IDs that stacking should treat as motor-only (force-skip their pairwise transforms).

A slice is force-skipped when it is explicitly excluded (use=false) or was flagged by auto-exclude (auto_excluded=true).

Parameters:

path (pathlib.Path)

Return type:

set[str]