experiment_controller

Coordinator for application flow between configuration, workspace/sample state, and processing workers.

This module wires together three main subsystems:

• ConfigCoordinator — in-memory working model for all config files with atomic commit and validation semantics.

• SampleManager — runtime view of the sample + Workspace2 (assets on disk), reconciled from config and updated via events.

• EventBus — decoupled, typed pub/sub used by the UI and back-end.

Responsibilities

• Boot an experiment (open/clone/new), seed defaults, run adjusters, validate, and commit configuration.

• Construct and cache worker instances (per pipeline and channel), reconcile them when channels change, and expose a simple get_worker(…) API.

• Bridge UI patches into ConfigCoordinator.submit_patch(…) so all edits are validated and persisted atomically.

• Optionally seal and snapshot the working config for reproducible runs.

• Listen to UI/domain events (e.g., channel rename/changed) and keep worker topology consistent.

Thread-safety

• ConfigCoordinator provides its own locking; reads of get_config_view() are safe against mid-edit states.

• EventBus is thread-safe and resilient to dead weakrefs.

Design Notes

• All persistence flows through ConfigCoordinator

• SampleManager keeps the Workspace2 in sync by subscribing to domain events (e.g., ChannelRenamed) and by reading the validated config view.

• Worker creation is lazy; reconciliation removes stale workers while new ones are built on first access.

class AnalysisGroupController(cfg_coordinator_factory, event_bus, exp_controller_factory)

Bases: object

Manages multiple ExperimentControllers, one per experiment root (src_dir). Also acts as a factory/router to fetch workers tied to the correct sample.

apply_patch(patch: dict[str, Any]) → None

get_config_view() → dict[str, Any]

get_sample_manager(sample_src_dir: str | Path) → SampleManager

get_worker(sample_src_dir: str | Path, pipeline: str, *, channel=None, substep=None)

infer_required_sections() → set[str]

set_group_base_dir(results_folder: str | Path)

set_groups(groups: dict[str, list[str]])

set_pipeline(pipeline: str) → None

Single entry point for pipeline changes. Updates config and orchestrator atomically.

set_progress_watcher(watcher)

set_thread_wrapper(wrapper)

property density_orchestrator: DensityGroupAnalysisOrchestrator

property group_base_dir: Path

property group_cfg_coordinator

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

class AppMode(value)

Bases: Enum

EXPERIMENT = 'experiment'

GROUP = 'group'

class ExperimentController(*, cfg_coordinator: ConfigCoordinator, sample_manager: SampleManager, proc_launcher: ClearMap.pipeline_orchestrators.processor_launcher.ProcessorLauncher | None = None, evt_bus: EventBus, use_materializers: bool = True, use_snapshots: bool = False)

Bases: BusSubscriberMixin

Orchestrates experiment lifecycle, config edits, and worker topology.

This controller is the single entrypoint for app flow. It owns the ConfigCoordinator and SampleManager, constructs per-pipeline workers, applies UI patches through the coordinator, and reacts to bus events that affect worker/channel state.

Parameters:

• cfg_coordinator (ConfigCoordinator) – Central configuration manager (working view + validation + commit).

• sample_manager (SampleManager) – Provides sample semantics and manages the Workspace2.

• proc_launcher (ProcessorLauncher, optional) – Used for sealing/snapshotting and launching processors.

• evt_bus (EventBus) – Event bus for decoupled UI/backend communication.

• use_materializers (bool, default True) – Reserved for future materialization hooks (kept for compatibility).

• use_snapshots (bool, default False) – When True, runs seal+snapshot before launching processors.

_exp_dir

Current experiment directory (None until set_experiment_dir()).

Type:

Optional[pathlib.Path]

_workers

Cached workers keyed by pipeline and (channel_key, substep).

Type:

Dict[str, Dict[Tuple[ChannelKey, Optional[str]], object]]

_factories

Factory registry: (pipeline, substep) -> factory(sm, coord, channel_key).

Type:

Dict[Tuple[str, Optional[str]], Callable]

Signals

-------

- UiChannelRenamed  -> :meth:`on_channel_renamed`

- UiChannelsChanged -> :meth:`on_channels_changed`

Published Events

----------------

- (optional) WorkspaceChanged(exp_dir=...) when the experiment directory is set.

Ensure the event class exists before enabling.

Key Operations

--------------

boot_open(exp_dir)

Load configs, seed defaults, run adjusters, validate, and commit.

boot_new(dest_dir, template_dir=None)

Create from defaults or clone template, then open.

apply_ui_patch(patch)

Validate+commit a UI patch via ConfigCoordinator.submit_patch(…).

get_worker(pipeline, channel=None, substep=None)

Lazy construction of workers. Cached until reconciliation.

reconcile_workers_after_channel_change(before, after)

Remove stale workers after channel removals; new ones remain lazy.

seal_and_snapshot

Ensure committed config; optionally write an immutable snapshot.

Invariants

----------

- All config mutations go through `ConfigCoordinator.submit(...)` or

submit_patch(…); direct edits to files are disallowed.

- `SampleManager` derives runtime state from committed config and bus events.

- Workers never mutate global config directly; they request patches through

controller-provided paths.

apply_ui_patch(patch: Dict[str, Any]) → None

Single write entrypoint: Params call this with a patch (persisted fields only). We merge it, derive dependent sections, validate, and atomically commit.

boot_new(dest_dir: Path | None = None, template_dir: Path | None = None) → None

Create a new experiment from defaults or clone a template, then open it.

boot_open(exp_dir: Path) → None

Open an existing experiment directory: load all configs, build working model, materialize/validate/commit (to normalize), and build tabs.

channel_snapshot()

Single “truth” event: current channel list + registration partner defaults, derived centrally via the registration worker.

clone_from(template_dir: Path, dest_dir: Path) → None

Explicit clone API (alias); then open the new experiment.

ensure_config_present(name: str) → Path | None

Ensure the named config file exists in the current experiment directory, copying from defaults if needed. Returns the path or None if no default exists for that name.

Parameters:

name (str) – Logical config name, e.g. “sample”, “registration”, etc.

Returns

Path | None

The path to the ensured config file, or None if no default exists.

get_config_view() → Dict[str, Any]

Dict-like snapshot the UI can read to populate widgets. Do not expose repository paths here—UI shouldn’t hit disk.

get_or_init_sample_id()

get_worker(pipeline: str, channel: str | Tuple[str, ...] | None = None, substep: str | None = None)

get_workers(pipeline: str, channels: Iterable[str | Tuple[str, ...] | None], substep: str | None = None)

launch_processor(processor_class, **opts)

Typical run: seal snapshot, then launch processor with config-dir=<snapshot>.

on_channel_renamed(evt: UiChannelRenamed) → None

Apply a rename everywhere (workers, config), then publish one snapshot event.

on_channels_changed(evt: UiChannelsChanged) → None

Handle add/remove; seed/drop per-channel sections, purge artifacts, then publish.

read_sample_version() → str

reconcile_workers(pipeline: str, desired_channels: Iterable[str | Tuple[str, ...] | None] | None, substep: str | None = None, keep_global: bool = True) → Dict[str | Tuple[str, ...] | None, object]

Ensure all desired workers exist and remove obsolete ones for this (pipeline, substep). desired_channels=None means ‘global-only’ (i.e., {None} if keep_global).

reconcile_workers_after_channel_change(before: list[str], after: list[str]) → None

Prune workers that reference removed channels by delegating to reconcile_workers. Creation of workers for new channels is left lazy (on first use).

register_worker_factory(pipeline: str, factory: Callable, substep: str | None = None, scope: str = 'global') → None

Register a factory function that creates workers for a given pipeline (and optionally substep). The factory will be called with (sample_manager, cfg_coordinator, channel_key).

Parameters:

• pipeline (str) – Name of the processing pipeline (e.g., “registration”, “cell_map”).

• factory (Callable) – Factory function to create the worker. (built with the ctor of the worker)

• substep (Optional[str], default None) – Optional substep identifier within the pipeline.

• scope (str, default "global") – Scope of the worker: “global”, “per_channel”, or “per_pair”.

rename_workers_channels(old_chan: str | Tuple[str, ...], new_chan: str | Tuple[str, ...])

sample_path_exists() → bool

seal_and_snapshot() → Path

Ensure the on-disk config is fully materialized & valid; take an immutable snapshot directory used by pipeline_orchestrators for reproducibility.

set_experiment_dir(exp_dir: str | Path) → None

set_sample_id(sample_id: str) → None

set_workers_progress_watcher(watcher)

Attach watcher to SampleManager and every existing worker.

upgrade_configs(from_version: str, to_version: str) → None

Convert the config to the current ClearMap version in the current exp_dir. Reload all configs into the working model afterwards.

Parameters:

• from_version (str) – The version the config is currently in.

• to_version (str) – The version to convert the config to.

worker_is_ready(worker_name)

property exp_dir: Path | None

property hydrating: bool

swap_old_channel(ch_key: str | Tuple[str, ...] | None, old_chan: str | Tuple[str, ...], new_chan: str | Tuple[str, ...]) → str | Tuple[str, ...] | None

Swap old_chan by new_chan in ch_key if present. If ch_key is a simple channel (str) or None -> str.replace() or None If ch_key is a tuple -> replace old_chan by new_chan in the tuple :param ch_key: ChannelKey to update :type ch_key: ChannelKey :param old_chan: Channel name to replace :type old_chan: RealChannelKey :param new_chan: Channel name to use as replacement :type new_chan: RealChannelKey

Returns

ChannelKey:

new_ch_key with old_chan replaced by new_chan, or None if key was None