sample_info_management#

Sample-level metadata management, configuration synchronisation, workspace reconciliation, and channel queries.

SampleManager is the root object for a single ClearMap experiment. It owns the sample configuration (channel paths, resolutions, orientations, data types) and keeps the Workspace2 in sync with it. Every pipeline worker receives a SampleManager reference so it can resolve asset paths without knowing the experiment layout.

Responsibilities#

Configuration access: channels, data_types, get_channel_resolution(), etc. expose the sample config as typed Python values, always reflecting the latest committed state from ConfigCoordinator.
Workspace reconciliation: update_workspace() ensures the Workspace2 mirrors the current channel list — adding missing channels, updating raw-data paths, pruning deleted channels, and persisting workspace.yml.
Channel queries: get_channels_by_type(), get_channels_by_pipeline(), and get_channels_by_condition() provide filtered lookups with configurable missing_action / multiple_found_action policies ('ignore', 'warn', 'raise').
Pipeline discovery: compute_relevant_pipelines() infers which pipelines are active so the ExperimentController knows which config sections to load.
Asset access: Inherits get() and get_path().

`@adjuster_safe` marker#

Methods decorated with adjuster_safe() are safe to call from config adjusters before all channel configs are fully populated. check_protocol_coverage() verifies at module load time that every method in SampleManagerProtocol carries this marker.

Bootstrapping#

Use build_sample_manager() rather than constructing SampleManager directly:

from ClearMap.pipeline_orchestrators.sample_info_management import build_sample_manager

sm = build_sample_manager('/path/to/experiment')

print(sm.channels)                              # ['cfos', 'autofluorescence']
print(sm.get_channels_by_pipeline('CellMap'))   # ['cfos']
print(sm.get_channel_resolution('cfos'))        # (1.625, 1.625, 3.0)

raw = sm.get('raw', channel='cfos')
print(raw.is_tiled, raw.tile_grid_shape)        # True, array([3, 4])

See also

Workspace2: Asset management layer.
ExperimentController: Owns SampleManager and wires it to pipeline workers.
ClearMap.IO.assets_constants: CONTENT_TYPE_TO_PIPELINE mapping.

class SampleManager(config_coordinator: ConfigCoordinator, src_dir: Path | str | None = None)[source]#

Bases: OrchestratorBase

This class is used to manage the sample information Manage sample-level configurations and properties. Handle configurations related to the sample. Provide utility methods for checking sample properties.

static compress(assets: List[WorkspaceAsset], format: str | None = None)[source]#

static convert(assets: List[WorkspaceAsset], new_extension: str, processes: int | None = None, verbose: bool = False, **kwargs)[source]#

static decompress(assets: List[WorkspaceAsset], check: bool = True)[source]#

static plot(assets: List[WorkspaceAsset], **kwargs)[source]#

static resample(assets: List[WorkspaceAsset], x_scale: float = 1, y_scale: float = 1, z_scale: float = 1, x_resolution=None, y_resolution=None, z_resolution=None, x_shape=None, y_shape=None, z_shape=None, orientation=None, processes=None, verbose=False, **kwargs)[source]#

asset_names_to_assets(asset_names: List[str], channel: str | None = None, sample_id: str | None = None) → List[WorkspaceAsset][source]#

can_convert(channel: str) → bool[source]#

check_has_all_tiles(channel: str) → bool[source]#

Check whether all the tiles of the channel exist on disk

Parameters:: channel (str) – The channel to check

Returns

bool: True if all the tiles exist

clear_renamed_channels() → None[source]#

colocalization_pair_keys(*, oriented: bool) → list[str][source]#

colocalization_pairs() → list[tuple[str, str]][source]#

compute_relevant_pipelines() → set[str][source]#: Infer active per-sample pipelines based purely on this sample’s channels / types. Does not know/care about group/batch.

compute_required_sections() → set[str][source]#: Given this controller’s SampleManager (single-sample view), compute which config sections should exist for this experiment.

data_type(channel: str) → str[source]#

delete_resampled_files(channel: str)[source]#

get_channel_resolution(channel: str) → tuple[float, float, float][source]#

Get the resolution of the channel as defined in the sample config.

Parameters:: channel (str) – The channel to get the resolution for

Returns

tuple(float, float, float): The resolution of the channel in (x, y, z) format

get_channels_by_condition(condition: Callable, missing_action: str = 'ignore', multiple_found_action: str = 'ignore', as_list: bool = False, error_label: str = 'channel') → str | list[str][source]#

Get the channel or list of channels that satisfy a given condition. The condition is specified as a function that takes a channel config and returns a boolean. e.g. to get the channel of type ‘autofluorescence’: get_channels_by_condition(lambda cfg: cfg[‘data_type’] == ‘autofluorescence’)

Parameters:

condition (function) – A function that takes a channel config and returns a boolean.
missing_action (str) – What to do in case no matching channel is found. One of [‘warn’, ‘raise’, ‘ignore’]
multiple_found_action (str) – What to do in case multiple matching channels are found. One of [‘warn’, ‘raise’, ‘ignore’]
as_list (bool) – Whether to return the result as a list if a single channel is found.
error_label (str) – The label to use in error messages.

Returns

str | List[str]: The channel name or list of channels that match the condition.

Raises:

KeyError – If no channel is found and missing_action is ‘raise’ If multiple channels are found and multiple_found_action is ‘raise’
ValueError – If an unknown action (missing_action or multiple_found_action) is specified

get_channels_by_pipeline(pipeline_name: str, missing_action: str = 'ignore', multiple_found_action: str = 'ignore', as_list: bool = False) → str | list[str][source]#

Get the channels that are relevant for a given pipeline

Parameters:

pipeline_name (str) – The name of the pipeline
missing_action (str) – What to do if no channel is found ‘ignore’ : ignore and return empty list ‘warn’ : warn and return empty list ‘raise’ : raise an error
multiple_found_action (str) – What to do if multiple channels are found ‘ignore’ : ignore and return all channels ‘warn’ : warn and return all channels ‘raise’ : raise an error
as_list (bool) – Whether to return the result as a list if a single channel is found.

Returns

List[str]: The channels that are relevant for the pipeline

Raises:

KeyError – If no channel is found and missing_action is ‘raise’ If multiple channels are found and multiple_found_action is ‘raise’
ValueError – If an unknown action (missing_action or multiple_found_action) is specified

get_channels_by_type(channel_type: str, missing_action: str = 'warn', multiple_found_action: str = 'ignore', as_list: bool = False) → str | list[str][source]#

Get the channel or list of channels that are of a given type.

Parameters:

channel_type (str) – Type of the channel as defined in asset_constants
missing_action (str) – What to do in case the channel specified is not found. One of [‘warn’, ‘raise’, ‘ignore’]
multiple_found_action (str) – What to do in case multiple matching channels are found. One of [‘warn’, ‘raise’, ‘ignore’]
as_list (bool) – Whether to return the result as a list if a single channel is found.

Returns

str | List[str]: The channel name or list of channels that match the type.

Raises:

KeyError – If no channel is found and missing_action is ‘raise’ If multiple channels are found and multiple_found_action is ‘raise’
ValueError – If an unknown action (missing_action or multiple_found_action) is specified

get_instance_keys_by_pipeline(pipeline_name: str, *, instance_kind: str, oriented: bool = False) → list[str][source]#

get_stitchable_channels() → list[str][source]#

has_npy(channel: str | None = None) → bool[source]#

Check if the channel is in npy format

Parameters:: channel (str) – The channel to check

Returns

bool: True if the raw channel is in npy format

has_tiles(channel: str | None = None) → bool[source]#

infer_channel_index_from_name(path: Path | str) → int | None[source]#

Extract channel index from typical microscopy filenames.

Extracts Cxx from filenames like _C00.ome.tif -> 0.

Parameters:: path (Path or str) – Microscopy image filename.

Returns

int or None: Channel index extracted from Cxx pattern, or None if not found.

Examples

>>> infer_channel_index_from_name("image_C03.ome.tif")
3

is_tiled(channel) → bool[source]#

needs_registering(registration_processor: RegistrationProcessor) → bool[source]#

patch_channel(channel, patch: dict)[source]#

rename_channels_in_workspace(names_map: Dict[str, str])[source]#

resampled_shape(channel: str) → tuple[int, int, int] | None[source]#

save_workspace()[source]#

set_channel_expression(channel: str, expression: str | tag_expression.Expression)[source]#

set_channel_resolution(channel: str, resolution: tuple[float, float, float])[source]#

set_renamed_channels(mapping: dict[str, str]) → None[source]#

set_resource_type_to_folder(new_mapping: dict, *, migrate: bool = False, dry_run: bool = False) → dict[str, tuple[Path, Path]][source]#

Update the workspace’s resource_type_to_folder layout.

If dry_run=True:
- compute and return the migration plan,
- DO NOT move files,
- DO NOT change workspace or persist anything.
If dry_run=False:
- apply layout to the workspace (optionally migrating),
- update SampleManager.resource_type_to_folder,
- persist the workspace.

setup(src_dir: Path | str | None = None)[source]#

Setup the sample manager with the given configs.

Parameters:: src_dir (str | Path | None) – The source directory of the sample

stitched_shape(channel: str) → tuple[int, int, int][source]#

update_workspace()[source]#

use_npy(channel: str) → bool[source]#

z_only(channel) → bool[source]#

Check if the channel is z only (no x or y tiles)

Parameters:: channel (str) – The channel to check

Returns

bool: True if the channel is z only

property alignment_reference_channel: str | None#

property autofluorescence_is_tiled: bool#

Check if the autofluorescence channel is tiled (has x and y tiles) .. rubric:: Returns

bool: True if the autofluorescence channel is tiled

property channels: list[str]#

property channels_to_convert: list[str]#

property channels_to_detect: list[str]#

config_name = 'sample'#

property data_types: list[str]#

property is_colocalization_compatible: bool#

property pipeline_ready_channels: list[str]#: Channels with a meaningful data_type (excludes undefined/unconfigured).

property prefix: str | None#

Get the prefix to use for the files

Returns

str: The prefix to use, None to not use any

property relevant_pipelines: list[str]#

All the pipelines relevant to any of the sample channels

Returns

List[str]: The relevant pipeline names

property renamed_channels: dict[str, str]#

property stitchable_channels: list[str]#

adjuster_safe(fn)[source]#: No-op marker: ‘this method tolerates incomplete channel configs’. This is meant to label SampleManager methods that are used in config adjusters (SampleManagerProtocol) and that can be safely called even if some channels have incomplete configs (e.g. missing path).

build_sample_manager(src_dir='', bus: EventBus | None = None)[source]#

check_protocol_coverage(cls, protocol_cls)[source]#

Check if all the decorated methods of SampleManager (meant to be used in SampleManagerProtocol) are present and decorated in the given class.

Raises TypeError if any protocol member lacks the marker.

sample_info_management#

Responsibilities#

@adjuster_safe marker#

Bootstrapping#

`@adjuster_safe` marker#