generic_orchestrators#

Base classes for all ClearMap processing workers.

Architecture#

The orchestrator hierarchy has three layers:

Configuration + asset access

OrchestratorBase owns the ConfigCoordinator reference and exposes read-only config views, workspace asset retrieval via get() / get_path(), and progress-watcher plumbing. Everything else in pipeline_orchestrators/ inherits from it.

Single-experiment pipelines

PipelineOrchestrator adds stop/cancel logic and the wrap_in_thread interface used by the GUI.

ChannelPipelineOrchestrator further narrows the scope to a single channel: its config property automatically returns the section for channel, and patch_channel() scopes config writes to that channel without the caller having to know the full key path.

CompoundChannelPipelineOrchestrator is the equivalent for pipelines that operate on a pair (or tuple) of channels, such as colocalization.

Step tracking

ProcessorSteps is an independent ABC that tracks which pipeline steps have already produced on-disk outputs. It is composed into concrete orchestrators that expose a binarization or graph-construction pipeline so the GUI can resume from the last completed step without re-running everything.

Multi-experiment / group analyses

GroupOrchestratorBase sits outside the single-sample hierarchy. It holds a reference to an AnalysisGroupController and provides per-sample worker and sample-manager access, plus progress and threading plumbing.

Relationship to the GUI#

The GUI never calls processing code directly. Instead, each pipeline_orchestrators/ tab creates and caches one or more workers (subclasses of the classes here) via get_worker. Config edits made in the GUI flow through ConfigCoordinator and are visible to workers on the next self.config access without reconstruction.

Typical advanced-user pattern:

from ClearMap.pipeline_orchestrators.sample_info_management import build_sample_manager
from ClearMap.pipeline_orchestrators.cell_map import CellDetector

sm = build_sample_manager('/path/to/experiment')
detector = CellDetector(sm, cfg_coordinator, channel='cfos')
detector.run_cell_detection()
detector.filter_cells()
detector.atlas_align()

exception CanceledProcessing[source]#: Bases: BrokenProcessPool

class ChannelPipelineOrchestrator(coordinator: ConfigCoordinator)[source]#

Bases: PipelineOrchestrator

Tab processor that is processing a single channel.

The config is expected to have a ‘channels’ section with the channel name as key. When accessed, the config property returns the section for the current channel.

channel_cfg_view(cfg_name)[source]#: Read-only view of the config section for the current channel. Raises if no channel is set or section missing.

get(asset_type, channel=<ClearMap.pipeline_orchestrators.generic_orchestrators._CurrentChannelSentinel object>, asset_sub_type=None, **kwargs)[source]#

Retrieve a workspace asset by type and channel.

Parameters:

asset_type (str) – Logical asset type, e.g. 'stitched', 'cells', 'atlas'.
channel (str, optional) – Channel name. Defaults to 'current', which concrete subclasses resolve to their active channel.
asset_sub_type (str or None, optional) – Optional sub-type qualifier, e.g. 'raw', 'filtered'.
**kwargs – Forwarded to get().

Returns

Asset: The workspace asset object.

Raises:: ClearMapRuntimeError – If the workspace has not been initialised yet.

patch_channel(patch: dict, *, origin: str = '') → None[source]#

Submit a config patch scoped to the current channel.

The patch is automatically nested under {config_name}.channels.{channel} before being forwarded to the coordinator.

Parameters:

patch (dict) – Flat-or-nested dict of values to update for this channel.
origin (str, optional) – Human-readable description of the call site, used for audit logging. Inferred from the call stack when empty.

channel: str#

property config: Mapping[str, Any]#: Read-only view of the config section for the current channel. Raises if no channel is set or section missing.

config_name = ''#

class CompoundChannelPipelineOrchestrator(coordinator: ConfigCoordinator)[source]#

Bases: PipelineOrchestrator

Tab processor that is processing a compound channels (made of several source channels)

The config is expected to have a ‘channels’ section with the channel names as keys. When accessed, the config property returns the section for all channels.

get(asset_type, channel=<ClearMap.pipeline_orchestrators.generic_orchestrators._CurrentChannelSentinel object>, asset_sub_type=None, **kwargs)[source]#

Retrieve a workspace asset by type and channel.

Parameters:

asset_type (str) – Logical asset type, e.g. 'stitched', 'cells', 'atlas'.
channel (str, optional) – Channel name. Defaults to 'current', which concrete subclasses resolve to their active channel.
asset_sub_type (str or None, optional) – Optional sub-type qualifier, e.g. 'raw', 'filtered'.
**kwargs – Forwarded to get().

Returns

Asset: The workspace asset object.

Raises:: ClearMapRuntimeError – If the workspace has not been initialised yet.

patch_channel(patch: dict, *, origin: str = '') → None[source]#

channels: List[str]#

property config: Mapping[str, Any]#: Read-only view of the config section for the current channel. Raises if no channel is set or section missing.

config_name = ''#

class GroupOrchestratorBase(*, group_controller: AnalysisGroupController, groups: dict[str, list[str]] | None = None, channel: str | None = None)[source]#

Bases: object

Base for multi-experiment/group analyses. Depends on AnalysisGroupController (not a single ExperimentController). Provides progress + threading plumbing and per-sample worker access.

get_sample_manager_for(sample_src_dir: str | Path)[source]#

Return the SampleManager for a specific experiment directory.

Parameters:: sample_src_dir (str or Path) – Root directory of the experiment.

Returns

SampleManager

get_worker_for_sample(sample_src_dir: str | Path, pipeline: str, *, channel=None, substep=None)[source]#

Return the pipeline worker for a specific sample directory.

Parameters:

sample_src_dir (str or Path) – Root directory of the experiment.
pipeline (str) – Pipeline name, e.g. 'cell_map', 'registration'.
channel (str or None, optional) – Channel key (required for per-channel pipelines).
substep (str or None, optional) – Substep key (required for multi-substep pipelines such as vasculature).

Returns

PipelineOrchestrator: The worker for that sample + pipeline combination.

run()[source]#

Execute the group-level analysis.

Raises:: NotImplementedError – Must be implemented by concrete subclasses.

set_progress_watcher(watcher) → None[source]#

Attach a progress watcher for GUI progress-bar updates.

Parameters:: watcher (ProgressWatcher)

set_thread_wrapper(wrapper_callable: Callable) → None[source]#

Inject the GUI thread-wrapping helper (to keep the UI responsive).

Parameters:: wrapper_callable (Callable) – Typically main_window.wrap_in_thread. Falls back to synchronous execution when not set.

channel: str | None#

property groups: dict[str, list[str]]#

Mapping of group name → list of experiment source-directory paths.

Returns

dict[str, list[str]]

property results_folder: Path#

Root directory where group-level outputs are written.

Returns

Path

Raises:: ValueError – If results_folder has not been set via the group controller.

class OrchestratorBase(coordinator: ConfigCoordinator)[source]#

Bases: BusSubscriberMixin

Minimal base shared by all ClearMap processors and managers.

Owns the ConfigCoordinator reference and provides read-only config views, workspace access, and optional logging plumbing.

Parameters:: coordinator (ConfigCoordinator) – Central configuration manager. All config reads and patch submissions go through this object.

cfg_coordinator#

Type:: ConfigCoordinator

workspace#

Set by concrete subclasses after the sample is loaded.

Type:: Workspace2 or None

registration_processor#

Injected when atlas-space operations are needed.

Type:: RegistrationProcessor or None

setup_complete#

True once the subclass has finished its own setup() call.

Type:: bool

filename(*args, **kwargs)[source]#

A shortcut to get the file path from the workspace

Parameters:

args (list) – Any positional argument accepted by workspace.filename
kwargs (dict) – Any keyword argument accepted by workspace.filename

Returns

str: The file path as a string

get(asset_type, channel='current', asset_sub_type=None, **kwargs)[source]#

Retrieve a workspace asset by type and channel.

Parameters:

asset_type (str) – Logical asset type, e.g. 'stitched', 'cells', 'atlas'.
channel (str, optional) – Channel name. Defaults to 'current', which concrete subclasses resolve to their active channel.
asset_sub_type (str or None, optional) – Optional sub-type qualifier, e.g. 'raw', 'filtered'.
**kwargs – Forwarded to get().

Returns

Asset: The workspace asset object.

Raises:: ClearMapRuntimeError – If the workspace has not been initialised yet.

get_alignment_ref_channel_reg_cfg() → Mapping[str, Any][source]#

Return the registration config for the alignment reference channel.

Delegates to self.registration_processor.ref_channel_cfg.

Returns

Mapping[str, Any]: Read-only view of the reference channel’s registration section.

Raises:: ValueError – If registration_processor has not been injected.

get_path(asset_type, channel='current', asset_sub_type=None, **kwargs)[source]#

Shortcut that returns the filesystem path of a workspace asset.

Parameters:

asset_type (str) – Logical asset type.
channel (str, optional) – Channel name.
asset_sub_type (str or None, optional) – Optional sub-type qualifier.
**kwargs – Forwarded to get().

Returns

Path: The path of the requested asset.

reload_config()[source]#: Deprecated no-op: configs are managed in-memory by the coordinator.

setup_if_needed()[source]#

cfg_coordinator: ConfigCoordinator#

property config: Mapping[str, Mapping[str, Any]]#

Always-fresh, read-only snapshot of this processor’s config section.

The section is identified by config_name.

Returns

MappingProxyType: Immutable view; request again to get the latest values after a patch.

config_name = ''#

property machine_config#

Read-only view of the machine (global params) config section (verbosity, start_folder, font_size…).

Returns

Mapping[str, Any]

property registration_config#

Read-only view of the registration config section.

Returns

Mapping[str, Any]

registration_processor: 'RegistrationProcessor' | None#

setup_complete: bool#

property verbose#

True when the machine verbosity is set to 'debug'.

Returns

bool

workspace: Workspace2 | None#

class PipelineOrchestrator(coordinator: ConfigCoordinator)[source]#

Bases: OrchestratorBase

Generic tab processor class. This class is inherited by all pipeline_orchestrators in ClearMap

Base methods:

config access and mutation,
asset retrieval through workspace
progress watcher handling
process stopping
run (to be implemented in child classes)

cfg_coordinator#

The configuration coordinator to manage the configuration files.

Type:: ConfigCoordinator

stopped#

Flag to indicate if the process has been stopped.

Type:: bool

progress_watcher#

The progress watcher to monitor the progress of the processing.

Type:: ProgressWatcher or None

workspace#

The workspace to manage the assets.

Type:: Workspace or None

prepare_watcher_for_substep(counter_size, pattern, title, increment_main=False)[source]#

Prepare the progress watcher for the coming processing step. The watcher will in turn signal changes to the progress bar

Arguments

counter_size: int: The progress bar maximum
pattern: str or re.Pattern or (str, re.Pattern) or None: The string to search for in the log to signal an increment of 1
title: str: The title of the step for the progress bar
increment_main: bool: Whether a new step should be added to the main progress bar

run()[source]#

Execute the full processing pipeline for this orchestrator.

Raises:: NotImplementedError – Must be implemented by every concrete subclass.

set_progress_watcher(watcher)[source]#

Attach a progress watcher that drives progress-bar updates.

Parameters:: watcher (ProgressWatcher) – Watcher instance whose increment, increment_main_progress and main_step_name interface will be used.

set_watcher_step(step_name)[source]#

Update the displayed step name in the progress dialog.

Parameters:: step_name (str) – Human-readable name of the current processing step.

setup(sample_manager: SampleManager | None = None)[source]#

Attach a sample manager and mark this processor as ready.

Parameters:: sample_manager (SampleManager, optional) – If provided, replaces the currently stored sample manager. When None, the previously stored instance is reused.
Raises:: ValueError – If the config section identified by config_name is absent.

setup_if_needed()[source]#: Warning

This assumes self.channel is set if needed! and self.registration_processor is set if needed!

stop_process()[source]#

Request cancellation of the current processing step.

Sets stopped to True and attempts to shut down any running executor or subprocess attached to the workspace. Raises CanceledProcessing when a subprocess is terminated.

update_watcher_main_progress(val=1)[source]#

Increment the main (top-level) progress bar by val.

Parameters:: val (int, optional) – Number of main steps completed. Default is 1.

update_watcher_progress(val)[source]#

Increment the sub-step progress bar by val.

Parameters:: val (int) – Number of units to add.

progress_watcher: 'ProgressWatcher' | None#

sample_manager: 'SampleManager' | None#

stopped: bool#

class ProcessorSteps(workspace: ClearMap.IO.workspace2.Workspace2, channel: str | Sequence[str] = '', step_order_provider: Callable[[], tuple[str, ...]] | None = None)[source]#

Bases: ABC

Ordered sequence of processing steps with disk-level asset tracking.

Step order is resolved fresh on every access via an injected provider callable, so GUI-driven reordering takes effect immediately without processor reconstruction.

Parameters:

workspace (Workspace2) – The workspace that owns the on-disk assets.
channel (str or Sequence[str]) – Channel name(s) this step sequence operates on.
step_order_provider (Callable[[], tuple[str, ...]] or None) – Optional callable that returns the current step order. When provided it is called on every access to steps, allowing the GUI to reorder steps without reconstructing the processor. Falls back to _default_steps when None.

abstractmethod asset_from_step_name(step_name: str) → ClearMap.IO.workspace_asset.Asset[source]#

Return the on-disk asset that corresponds to step_name.

Parameters:: step_name (str) – A member of steps.

Returns

Asset: The asset associated with that step.

Raises:: NotImplementedError – Must be implemented by every concrete subclass.

get_asset(step: str, *, step_back: bool = False, n_before: int = 0) → ClearMap.IO.workspace_asset.Asset[source]#

Return the Asset for step, optionally walking back if absent and step_back is True.

Parameters:

step (str) – Target step name.
step_back (bool) – If True and the asset does not exist, walk backwards through all preceding steps until one is found.
n_before (int) – If > 0, resolve to the step n_before positions earlier before checking existence.

Returns

Asset: The asset corresponding to the step name.

get_next_steps(step_name: str) → list[str][source]#

Return all steps that follow step_name in the pipeline.

Parameters:: step_name (str) – A member of steps.

Returns

list of str: Steps after step_name, in execution order. Empty list when step_name is the last step.

remove_next_steps_files(target_step_name: str) → None[source]#

Delete on-disk outputs for all steps that follow target_step_name.

Useful when a step is re-run and downstream outputs are therefore stale. Warns before each deletion.

Parameters:: target_step_name (str) – The step after which all existing outputs will be deleted.

step_exists(step_name: str) → bool[source]#

Check whether the output asset for step_name exists on disk.

Parameters:: step_name (str) – A member of steps.

Returns

bool: True if the corresponding asset exists.

channel: str | Sequence[str]#

property existing_steps: list[str]#

Steps whose output asset already exists on disk.

Returns

list of str: Subset of steps for which step_exists() is True, in step order.

property last_step: str | None#

The last step whose output exists on disk, or None.

Returns

str or None: Last element of existing_steps, or None when no step has been run yet.

property steps: tuple[str, ...]#

Ordered step names, resolved fresh on every access.

Returns

tuple of str: Step names in execution order, either from the injected provider or from _default_steps.

workspace: ClearMap.IO.workspace2.Workspace2#