Processing API

DICOM conversion

imperandi.process.convert.add_convert_arguments(parser, include_manifest=True, include_dry_run=True)

Add conversion paths, archive controls, and resume options to a parser.

Parameters:

• parser (ArgumentParser)

• include_manifest (bool)

• include_dry_run (bool)

imperandi.process.convert.build_parser(add_help=True)

Build the standalone DICOM-to-NIfTI conversion parser.

Parameters:

add_help (bool)

Return type:

ArgumentParser

imperandi.process.convert.convert_dicom_to_nifti_parallel(df, output_dir, show_progress, num_workers, *, on_result=None)

Convert multiple DICOM volumes to NIfTI in parallel using multiprocessing.

Parameters:

• df (pd.DataFrame) – DataFrame containing DICOM metadata.

• output_dir (str) – Directory to save the NIfTI files.

• show_progress (bool) – Whether to display progress using tqdm.

• num_workers (int) – Number of parallel processes to use.

Returns:

The updated DataFrame with NIfTI paths and a DataFrame with any errors encountered.

Return type:

tuple

imperandi.process.convert.convert_list_str_to_list(cell)

Convert a string representation of a list to an actual list using literal_eval.

Parameters:

cell (str) – String that represents a list.

Returns:

If the string can be converted to a list, return the list, else return the original value.

Return type:

list or original value

imperandi.process.convert.main(args)

Main function to convert DICOM series to NIfTI files in parallel and save the results to CSV.

Parameters:

args (argparse.Namespace) – Parsed command-line arguments.

imperandi.process.convert.materialize_archive_dicom_paths(df, archive_session)

Replace archive:// DICOM paths with local materialized paths. Returns updated dataframe and an error dataframe for rows that could not be materialized.

Parameters:

• df (DataFrame)

• archive_session (ArchiveSession)

Return type:

tuple[DataFrame, DataFrame]

imperandi.process.convert.normalize_convert_args(args)

Resolve conversion paths and validate input CSVs in-place.

Raises:

• FileNotFoundError – If an input CSV does not exist.

• ValueError – If an input is not CSV or no output directory is supplied.

Parameters:

args (Namespace)

Return type:

Namespace

imperandi.process.convert.parse_arguments()

Parse and normalize arguments for the standalone convert command.

imperandi.process.convert.process_single_volume(k, row, output_dir, verbose, return_status=False)

Convert a single DICOM series to a NIfTI file, saving the result to the specified output directory.

Parameters:

• k (int) – Index of the current volume being processed.

• row (pd.Series) – Metadata for the current DICOM series.

• output_dir (str) – Directory to save the NIfTI files.

• verbose (bool) – If true, configure verbose logging for worker setup.

Returns:

• default (return_status=False): (index, export_path, error_row)

• with status (return_status=True): (index, export_path, error_row, status), where status is “converted”, “skipped”, or “failed”.

Return type:

tuple

Segmentation

segment.py

Batch‑process a list of 3‑D volumes to obtain masks with a configurable segmentation backend (default: TotalSegmentator v2).

The module supports both CLI and library usage:

1. reads a CSV containing a nifti_path column,

2. spawns a multiprocessing pool (spawn context – required for PyTorch + CUDA),

3. runs config‑driven segmentation tasks per volume,

4. optionally merges / cleans masks, and

5. writes updated CSVs with output paths and a separate error CSV.

class imperandi.process.segment.TotalSegmentatorBackend

Bases: object

Thin wrapper for TotalSegmentator to keep dependency optional.

run(*, input_path, output_dir, task, **kwargs)

Parameters:

• input_path (Path)

• output_dir (Path)

• task (str)

• kwargs (Any)

Return type:

None

imperandi.process.segment.add_segment_arguments(parser, include_manifest=True, include_dry_run=True)

Add segmentation, multiprocessing, and resume options to a parser.

Parameters:

• parser (ArgumentParser)

• include_manifest (bool)

• include_dry_run (bool)

Return type:

None

imperandi.process.segment.build_output_column_map(tasks)

Map each unique logical task output to its mask_* CSV column.

Parameters:

tasks (List[Dict[str, Any]])

Return type:

Dict[str, str]

imperandi.process.segment.build_output_fetch_map(tasks)

Map logical task outputs to filenames produced by the backend.

Parameters:

tasks (List[Dict[str, Any]])

Return type:

Dict[str, str]

imperandi.process.segment.build_parser(add_help=True)

Build the standalone batch-segmentation parser.

Parameters:

add_help (bool)

Return type:

ArgumentParser

imperandi.process.segment.clean_and_merge_masks(dir_path, mask_files, *, output_name, radius_mm=5.0, verbose=False, close=True, fill_holes=True, largest_cc=True)

Merge masks and optionally apply morphological cleanup.

Parameters:

• dir_path (Path)

• mask_files (List[str])

• output_name (str)

• radius_mm (float)

• verbose (bool)

• close (bool)

• fill_holes (bool)

• largest_cc (bool)

Return type:

bool

imperandi.process.segment.compute_struct_elem(zooms, radius_mm=5.0)

Create a spherical structuring element with radius_mm in real units.

Parameters:

• zooms (Tuple[float, ...])

• radius_mm (float)

Return type:

ndarray

imperandi.process.segment.infer_task_fetch_outputs(task)

Map logical output keys to backend-produced filenames to fetch.

Parameters:

task (Dict[str, Any])

Return type:

Dict[str, str]

imperandi.process.segment.infer_task_outputs(task)

Infer normalized logical output names from one segmentation task.

Parameters:

task (Dict[str, Any])

Return type:

List[str]

imperandi.process.segment.load_nifti(path)

Return image data, affine matrix and voxel sizes (zoom).

Parameters:

path (Path)

Return type:

Tuple[ndarray, ndarray, Tuple[float, …]]

imperandi.process.segment.load_segmentation_config(manifest_arg, *, base_path)

Load segmentation config from manifest, falling back to generic manifest.

Parameters:

• manifest_arg (str | None)

• base_path (Path)

Return type:

Dict[str, Any]

imperandi.process.segment.main(args)

Run manifest-configured segmentation over every eligible cohort row.

Parameters:

args (Namespace)

Return type:

None

imperandi.process.segment.normalize_segment_args(args)

Resolve segmentation input, output, and error paths in-place.

Parameters:

args (Namespace)

Return type:

Namespace

imperandi.process.segment.prefetch_totalsegmentator_models(tasks_config)

Download required TotalSegmentator weights before multiprocessing.

Parameters:

tasks_config (Dict[str, Any])

Return type:

None

imperandi.process.segment.process_single_volume(idx, row, tasks_config, *, verbose, force, backend=None)

Return (idx, output_dir|None, error_msg|None, warning_msg|None, outputs|None).

Parameters:

• idx (int)

• row (Dict[str, Any])

• tasks_config (Dict[str, Any])

• verbose (bool)

• force (bool)

• backend (TotalSegmentatorBackend | None)

Return type:

Tuple[int, str | None, str | None, str | None, Dict[str, str] | None]

imperandi.process.segment.resolve_merge_outputs(postprocess, tasks, *, output_to_column=None)

Resolve post-processing merge keys to known logical task outputs.

Raises:

ValueError – If merge keys are absent or refer to unknown mask columns.

Parameters:

• postprocess (Dict[str, Any])

• tasks (List[Dict[str, Any]])

• output_to_column (Dict[str, str] | None)

Return type:

List[str]

imperandi.process.segment.save_nifti(data, affine, out_path, *, dtype=<class 'numpy.uint8'>)

Write data to out_path as a NIfTI‑1 file with dtype.

Parameters:

• data (ndarray)

• affine (ndarray)

• out_path (Path)

Return type:

None

imperandi.process.segment.segment_volume(nifti_path, output_dir, tasks_config, *, verbose=False, force=False, backend=None, resolved_output_to_fetch=None)

Run segmentation tasks and optional post‐processing.

Parameters:

• nifti_path (Path)

• output_dir (Path)

• tasks_config (Dict[str, Any])

• verbose (bool)

• force (bool)

• backend (TotalSegmentatorBackend | None)

• resolved_output_to_fetch (Dict[str, str] | None)

Return type:

List[str]