Ingestion API

Parsing

imperandi.ingest.parse.add_parse_arguments(parser, include_manifest=True, include_dry_run=True)

Add DICOM discovery, identity, archive, and resume options to a parser.

Parameters:

• parser (ArgumentParser)

• include_manifest (bool)

• include_dry_run (bool)

Return type:

None

imperandi.ingest.parse.build_effective_tags(*, default_tags, user_tags, patient_tag, study_tag, series_tag)

Combine default, requested, and identifier tags without duplicates.

Parameters:

• default_tags (list[str])

• user_tags (list[str])

• patient_tag (str)

• study_tag (str)

• series_tag (str)

Return type:

list[str]

imperandi.ingest.parse.build_global_readers(*, initial_archive_mode, tags, force, archive_max_depth)

Build selected/full header readers that can switch to archive mode.

Returns:

A selected-tag reader, a full-header reader, and their shared mutable archive-detection state.

Parameters:

• initial_archive_mode (bool)

• tags (list[str])

• force (bool)

• archive_max_depth (int)

imperandi.ingest.parse.build_parser(add_help=True, include_manifest=True, include_dry_run=True)

Build the standalone parser for DICOM metadata ingestion.

Parameters:

• add_help (bool)

• include_manifest (bool)

• include_dry_run (bool)

Return type:

ArgumentParser

imperandi.ingest.parse.choose_ids(df, root_path, id_source, patient_tag, study_tag, series_tag, scan_root_col='_scan_root', relative_path_col='_relative_path')

Compute patient_key / study_id / series_id.

Rules:

• patient_key_path = first directory under root

• study_id, series_id come from tags only

• if study/series tags missing -> “0”

• multiple files per series are expected

Parameters:

• df (DataFrame)

• root_path (Path)

• id_source (str)

• patient_tag (str)

• study_tag (str)

• series_tag (str)

• scan_root_col (str)

• relative_path_col (str)

Return type:

DataFrame

imperandi.ingest.parse.default_output_dir(root_path)

Choose the default parse output directory for an input path or glob.

Parameters:

root_path (str | None)

Return type:

Path

imperandi.ingest.parse.detect_archive_mode_by_subsample(resolved_roots, sample_size=128)

Return whether a deterministic sample of the roots contains archives.

Parameters:

• resolved_roots (list[Path])

• sample_size (int)

Return type:

bool

imperandi.ingest.parse.ensure_directory_exists(output_dir)

Create an output directory and any missing parents.

Parameters:

output_dir (Path)

imperandi.ingest.parse.extract_dicom_tags_recursive(ds, parent_key='')

Flatten all readable elements of a DICOM dataset into a dictionary.

imperandi.ingest.parse.get_dicom_path_entries(root_path, archive_max_depth=3)

Strategy: 1) Resolve root_path as directories and/or archive files. 2) Recursively discover DICOM sources, including nested archives. 3) Prefer *.dcm and fallback to header validation when needed.

Parameters:

• root_path (str | Path)

• archive_max_depth (int)

Return type:

list[dict]

imperandi.ingest.parse.get_dicom_paths(root_path)

Return discovered DICOM paths for compatibility with older callers.

imperandi.ingest.parse.main(args)

Run DICOM discovery and metadata parsing for a normalized namespace.

imperandi.ingest.parse.normalize_parse_args(args)

Resolve parse defaults and validate checkpoint settings in-place.

Named path options take precedence over positional values. Temporary parser attributes are removed from the returned namespace.

Parameters:

args (Namespace)

Return type:

Namespace

imperandi.ingest.parse.parse_arguments()

Parse and normalize arguments for the standalone parse command.

imperandi.ingest.parse.process_with_checkpoint(df_paths, read_func, checkpoint_every_rows, checkpoint_every_sec, resume, strict_resume, output_dir, final_name, read_path_col='dicom_path', *, num_workers=1, worker_config=None, transform_chunk=None, return_df=True, expected_columns=None, resume_signature=None)

Apply DICOM header reads in chunked mode with append-only checkpointing.

Parameters:

• df_paths (DataFrame)

• checkpoint_every_rows (int)

• checkpoint_every_sec (int)

• resume (bool)

• strict_resume (bool)

• output_dir (Path)

• final_name (str)

• read_path_col (str)

• num_workers (int)

• worker_config (dict | None)

• return_df (bool)

• expected_columns (list[str] | None)

• resume_signature (dict | None)

imperandi.ingest.parse.read_dicom_header(source, *, force=False, archive_aware=False, archive_max_depth=3)

Parameters:

• archive_aware (bool)

• archive_max_depth (int)

imperandi.ingest.parse.read_dicom_header_archive_aware(source, *, force=False, archive_max_depth=3)

imperandi.ingest.parse.read_dicom_header_selected(source, *, tags, force=False, archive_aware=False, archive_max_depth=3)

Parameters:

• tags (list[str])

• archive_aware (bool)

• archive_max_depth (int)

imperandi.ingest.parse.read_dicom_header_selected_archive_aware(source, *, tags, force=False, archive_max_depth=3)

Parameters:

tags (list[str])

imperandi.ingest.parse.read_dicom_header_selected_standard(source, *, tags, force=False)

Read selected tags from a DICOM header. Returns pd.Series with one key per requested tag.

Parameters:

tags (list[str])

imperandi.ingest.parse.read_dicom_header_standard(source, *, force=False)

Read header once: recursively extract all DICOM tags into columns. Returns pd.Series with all tags flattened.

imperandi.ingest.parse.read_dicom_header_with_force(fp, force)

imperandi.ingest.parse.resolve_root_paths(root_path)

Resolve a directory, archive, or glob into deterministic input roots.

A missing path resolves to the current working directory. Glob results are de-duplicated and restricted to directories and supported archives.

Parameters:

root_path (str | Path | None)

Return type:

list[Path]

imperandi.ingest.parse.write_dicom_tags_snapshot(*, df, output_path, sample_size, seed, series_col='SeriesInstanceUID', read_path_col='_read_path', read_full_func=None)

Write recursive tags from a deterministic source sample as NDJSON.

Returns:

The number of snapshot records written.

Parameters:

• df (DataFrame)

• output_path (Path)

• sample_size (int)

• seed (int)

• series_col (str)

• read_path_col (str)

Return type:

int

Cleaning

imperandi.ingest.clean.add_clean_arguments(parser, include_manifest=True, include_csv_path=True, include_csv_path_out=True, include_dry_run=True)

Add metadata-cleaning paths, thresholds, and manifest options.

Parameters:

• parser (ArgumentParser)

• include_manifest (bool)

• include_csv_path (bool)

• include_csv_path_out (bool)

• include_dry_run (bool)

Return type:

None

imperandi.ingest.clean.add_date(df)

imperandi.ingest.clean.add_time(df)

imperandi.ingest.clean.build_parser(add_help=True, include_manifest=True)

Build the standalone parser for metadata cleaning.

Parameters:

• add_help (bool)

• include_manifest (bool)

Return type:

ArgumentParser

imperandi.ingest.clean.calculate_volume_length(df)

Compute reconstructed volume length in millimetres from slice geometry.

imperandi.ingest.clean.clean_and_save_data(csv_path, csv_path_out, csv_dict_path, manifest, volume_length_min_mm, volume_length_max_mm)

Run the complete metadata-curation pipeline and write its CSV output.

Parameters:

• csv_path – One or more parsed metadata CSV paths.

• csv_path_out – Destination for the curated volume table.

• csv_dict_path – Optional DICOM tag dictionary used by cleaning.

• manifest – Loaded dataset configuration and hook definitions.

• volume_length_min_mm – Inclusive minimum reconstructed length.

• volume_length_max_mm – Inclusive maximum reconstructed length.

The function writes csv_path_out when provided and otherwise performs the same transformations and logging without persisting a table.

imperandi.ingest.clean.clean_pixel_spacing(df)

imperandi.ingest.clean.clean_scan_size(df)

imperandi.ingest.clean.compute_acquisition_order(df)

imperandi.ingest.clean.compute_visit_order(df)

imperandi.ingest.clean.correct_volume_ids(df, z_tolerance=0.001)

Merge “pseudo-volumes” (multiple volume_id values) that actually belong to the same volume, but do it robustly when DICOM tags/columns are missing.

Strategy: - If volume_id missing -> return df unchanged. - Group by the maximum available columns from a preferred list.

If none available -> fallback to grouping by patient_key, study_id, series_id (subset that exists).

• Determine z positions using the best available source:

  1. ImagePositionPatient (z component)

  2. SliceLocation

  If neither usable -> skip that group.

• If spacing between sorted z positions is consistent (within tolerance) -> merge volume_ids.

imperandi.ingest.clean.drop_irrelevant_dicom_tags(df)

imperandi.ingest.clean.filter_by_acquisition_plane(df, angle_thresh_deg=10.0)

imperandi.ingest.clean.filter_ct_modality(df)

Keep CT Image Storage rows when modality tags are available.

imperandi.ingest.clean.filter_image_type(df)

imperandi.ingest.clean.filter_volumes_by_size(df, min_length_mm, max_length_mm)

Keep volumes within inclusive length bounds, retaining missing lengths.

imperandi.ingest.clean.generate_volume_id(df)

Add a deterministic identifier for each candidate imaging volume.

imperandi.ingest.clean.group_volumes(df)

Aggregate instance-level metadata into one row per volume identifier.

imperandi.ingest.clean.load_data(csv_path)

Load and concatenate metadata CSVs, dropping empty helper columns.

imperandi.ingest.clean.map_series_description(df, csv_tag_dict)

imperandi.ingest.clean.normalize_clean_args(args)

Resolve clean input/output paths and legacy threshold aliases in-place.

Parameters:

args (Namespace)

Return type:

Namespace

imperandi.ingest.clean.parse_arguments()

Parse and normalize arguments for the standalone clean command.

imperandi.ingest.clean.read_csv_with_valid_columns(file)

Read only recognized identifier and DICOM metadata columns from CSV.

imperandi.ingest.clean.remove_mpr(df)

imperandi.ingest.clean.remove_other_organs_description(df)

imperandi.ingest.clean.remove_pet_ct(df)

imperandi.ingest.clean.remove_scouts_localizers(df)

imperandi.ingest.clean.reorder_columns(df)

imperandi.ingest.clean.reorder_rows(df)

imperandi.ingest.clean.uniform_string(s)

Manifest hooks

imperandi.ingest.apply_hook_manifests.apply_derived_columns(df, manifest, *, hook_resolver=<function resolve_hook>)

Apply manifest-defined hooks that derive columns from existing values.

Each hook result is expanded as a mapping. Existing columns are preserved by the default missing_only join mode or replaced by overwrite.

Parameters:

• df (DataFrame) – Metadata table to enrich.

• manifest (dict) – Loaded dataset manifest.

• hook_resolver (Callable[[dict], Callable | None]) – Callable that resolves a hook configuration.

Returns:

The enriched table.

Return type:

DataFrame

imperandi.ingest.apply_hook_manifests.apply_id_standardization(df, manifest, *, hook_resolver=<function resolve_hook>, logger=None)

Standardize patient_key using the hook configured by a manifest.

The original value is retained in _patient_key_raw. If a non-empty raw key produces an empty standardized key, patient_key_std_failed marks the affected row.

Parameters:

• df (DataFrame) – Metadata table to update.

• manifest (dict) – Loaded dataset manifest.

• hook_resolver (Callable[[dict], Callable | None]) – Callable that resolves a hook configuration.

• logger (Logger | None) – Optional logger for standardization warnings.

Returns:

The updated metadata table.

Return type:

DataFrame