Skip to main content

data_utils

Composable data utilities for steps.

Decomposes the monolithic initialise_model(data=...) flow into individually callable functions so that each concern (schema discovery, column configuration, dataloader creation, model init, inference) can be used and tested independently.

Also provides shared data-manipulation utilities that multiple steps need (e.g. get_data_for_files, convert_predictions_to_dataframe, convert_metrics_to_df, convert_ga_metrics_to_df).

Sub-modules​

inference Model inference pipeline helpers (build_column_config, create_inference_dataloader, init_model_for_inference, run_inference). dataframe DataFrame conversion and manipulation (convert_predictions_to_dataframe, get_data_for_files, convert_metrics_to_df, convert_ga_metrics_to_df, add_age_column). parsing Image/mask parsing and the shared prediction-envelope helpers (parse_mask_json, is_na_prediction, load_prediction_envelope). datasource Datasource helpers (resolve_output_path, use_default_rename_columns). enrichment Cache -> patient-grain frame for supplied patient enrichment (supplied_observations_from_cache). ga_metrics Shared GA metrics computation primitives used by both ga_calculation_with_fovea and ga_calculation_without_fovea steps (convert_nan_to_zero, get_max_ga_bscan_index, get_lesion_sizes, get_shortest_distance_from_image_centre, parse_bscan_predictions, compute_ga_metrics_for_scan). metric_frame Shared scan-metrics frame assembly used by both criteria_matching and biomarker_tabulation (assemble_scan_frame). fovea Fovea landmark extraction shared between step packages (get_central_slice_landmark_point, FOVEA_CENTRE_LANDMARK_INDEX). morphometry Pure voxel-scale geometry shared by every en-face and volumetric biomarker (VoxelScale, has_en_face_scale, has_voxel_scale, en_face_area_mm2, volume_mm3, volume_nl, label_en_face, label_volume, en_face_distance_grid). segmentation_masks One walk of a scan's segmentation cube, producing per-label en-face presence and per-GROUP voxel depth (ScanMasks, parse_bscan_masks, combine_label_mask, group_reports_volume). Depth is combined per group during the walk because a per-label voxel count cannot be combined into a group count afterwards. lesions Per-lesion extraction and diameter measurement, labelling a group's footprint into individual lesions and reporting two diameters for the cube's anisotropic sampling (RawLesion, extract_lesions, max_base_width_mm, max_feret_mm). fovea_geometry Fovea-relative geometry in millimetre space: distance, region of interest, and angular span (roi_mask, min_distance_from_reference_mm, angular_occupancy, aggregate_angular_span_deg).

Module​

Submodules​

Functions​

_sort_col_by_list​

def _sort_col_by_list(    col: pandas.core.series.Series, sort_by: list[str],) ‑> pd.Series[int | float]:

Sort helper: maps values to their index in sort_by.

add_age_column​

def add_age_column(    df: pandas.core.frame.DataFrame,    dob_col: str = "Patient's Birth Date",    age_col: str = 'Age (yrs)',) ‑> pandas.core.frame.DataFrame:

Add an age column to df based on a date-of-birth column.

If age_col already exists and is fully populated the dataframe is returned unchanged. If the column exists but contains NA / None values, this function falls through and fills those gaps from dob_col, leaving any already-populated rows untouched.

Unparseable DOB values (e.g. de-identification placeholders like "XXXX") are coerced to NaT; age is computed only for rows with a valid DOB and left as NA for the rest.

If DOB parsing raises OverflowError (e.g. DOB set to Timestamp.min as a placeholder), a timedelta-based fallback is used.

Returns The same dataframe with the extra age column.

apply_canonical_scan_metric_names​

def apply_canonical_scan_metric_names(    df: pd.DataFrame, metrics_type: type,) ‑> pandas.core.frame.DataFrame:

Rename a scan-metrics frame's columns to their canonical scan-metric names.

CST/GCC descriptor collisions become cst_*/gcc_*; fluid scalars gain the fluid_ prefix and the per-scan segmentation_volumes dict column is flattened into fluid_<label>_volume columns (absent label or non-dict cell → pd.NA) then dropped. Any other metrics type is returned unchanged. The ORIGINAL_FILENAME_METADATA_COLUMN merge key is never renamed (it is not in any rename map).

Arguments

  • df: The per-scan metrics frame produced by convert_metrics_to_df.
  • metrics_type: The metrics dataclass the frame's rows conform to.

Returns The frame with canonical column names. A metrics type needing no transform is returned unchanged (and unmutated); otherwise a copy.

assemble_scan_frame​

def assemble_scan_frame(    base: pandas.core.frame.DataFrame,    scan_frames: collections.abc.Sequence[pandas.core.frame.DataFrame],    *,    key: str = '_original_filename',) ‑> tuple[pandas.core.frame.DataFrame, dict[str, CriterionProvenance]]:

Left-merge each non-empty scan frame onto base, tagging SCAN provenance.

Shared scan-metrics merge loop used by both criteria_matching and biomarker_tabulation. Each already-built metrics frame (e.g. one per GA/fluid/CST/GCC producer) is merged in turn: colliding columns are coalesced via _drop_colliding_columns rather than suffixed, so a left-merge cannot silently orphan a criterion's target column.

Arguments

  • base: The frame being merged onto (already carries the merge key).
  • scan_frames: Already-built per-source metrics frames to merge in, in order; a frame with no rows is skipped. Each frame must hold at most one row per key value — the left-merge fans base's rows out on a duplicate key. Every current builder satisfies this (cache file_id is unique per partition; backfill_metrics keys by filename), so the precondition is not re-checked here.
  • key: The shared merge key column.

Returns (merged_df, column_provenance) — base with every non-empty scan_frames entry merged in, and a {column: CriterionProvenance.SCAN} map covering every column each merged-in frame contributed (after collision coalescing).

build_column_config​

def build_column_config(    image_prefix: str = 'Pixel Data',    selected_cols: Optional[list[str]] = None,    selected_cols_prefix: str | None = None,    schema_requirements: Any = 'empty',    batch_transforms: list[dict[str, typing.Any]] | None = None,    image_prefix_batch_transforms: list[dict[str, typing.Any]] | None = None,    auto_convert_grayscale_images: bool = True,) ‑> DataStructure:

Build a DataStructure suitable for ophthalmology inference.

This is a thin wrapper around the DataStructure constructor with sensible defaults matching the GA Trial Bronze task YAML.

Arguments

  • image_prefix: Image column prefix (default "Pixel Data").
  • selected_cols: Explicit list of selected columns. Defaults to ["Columns", "Rows"] when None.
  • selected_cols_prefix: Prefix for selected columns (default same as image_prefix).
  • schema_requirements: Schema requirement level.
  • batch_transforms: Optional batch transforms list.
  • image_prefix_batch_transforms: Optional image-specific batch transforms.
  • auto_convert_grayscale_images: Convert grayscale to RGB (default True).

Returns A DataStructure instance.

canonical_scan_metric_field_specs​

def canonical_scan_metric_field_specs(    metrics_type: type,) ‑> list[FieldSpec]:

FieldSpecs for a scan-metrics type, named to match the canonical frame.

Applies the same rename as apply_canonical_scan_metric_names to each of the type's field_specs() (so a column_criteria targeting a canonical name resolves), and for fluid appends the three flattened fluid_<label>_volume columns (which field_specs_from_dataclass skips because segmentation_volumes is a container). A type without field_specs contributes nothing rather than raising.

Arguments

  • metrics_type: The metrics dataclass a scan_metrics source declares.

Returns The canonical-named FieldSpec list (empty for an unrecognised type).

compute_ga_metrics_for_scan​

def compute_ga_metrics_for_scan(    bscan_predictions: tuple[str, ...],    slice_thickness: float,    pixel_spacing_column: float,    all_segmentation_labels: dict[str, int],    ga_area_include_segmentations: list[str],    ga_area_exclude_segmentations: list[str],    n_scan_biomarker_thresholds: Mapping[str, float] | None = None,    include_raw_pathology_probabilities: bool = False,) ‑> GAMetrics:

Compute GA metrics for a single scan (without fovea).

This encapsulates the per-file computation loop body from _WorkerSide.run() (lines 268-343 in the original).

Arguments

  • bscan_predictions: The raw per-B-scan prediction JSON strings.
  • slice_thickness: The distance between B-scans in mm.
  • pixel_spacing_column: The pixel spacing along the B-scan column axis.
  • all_segmentation_labels: All segmentation label names to class index.
  • ga_area_include_segmentations: Segmentation labels to include when computing the GA area column mask.
  • ga_area_exclude_segmentations: Segmentation labels to exclude when computing the GA area column mask.
  • n_scan_biomarker_thresholds: Per-biomarker >= probability threshold used to compute n_scan_run_lengths. Keys define which N-scan biomarkers are evaluated; when None, every label in N_SCAN_BIOMARKER_LABELS defaults to 0.5.
  • include_raw_pathology_probabilities: Whether to include the raw per-B-scan pathology probability arrays on the returned metrics.

Raises

  • Exception: Propagates any exception from prediction parsing or metric computation so that the caller can handle/skip.

convert_ga_metrics_to_df​

def convert_ga_metrics_to_df(    ga_metrics: collections.abc.Mapping[str, typing.Any],    additional_pathology_prob_cols: collections.abc.Collection[str] | None = None,    additional_area_cols: collections.abc.Collection[str] | None = None,    additional_run_length_cols: collections.abc.Collection[str] | None = None,    metrics_type: type | None = None,) ‑> pandas.core.frame.DataFrame:

Convert a dict of GAMetrics objects into a dataframe.

Thin back-compat wrapper over convert_metrics_to_df; preserves the GA-specific exemplar-by-type selection between GAMetrics and GAMetricsWithFovea for expected_cols(), while always guarding record building on the base GAMetrics type. This matters when ga_metrics mixes plain GAMetrics and GAMetricsWithFovea values: the resolved exemplar type may narrow to GAMetricsWithFovea (to pick up the fovea columns), but a plain GAMetrics entry must still have its base fields recorded rather than being dropped entirely for failing an isinstance check against the narrower subclass.

Arguments

  • ga_metrics: Mapping of filename to GAMetrics instances.
  • additional_pathology_prob_cols: Additional pathology names to extract into max_<name>_probability columns.
  • additional_area_cols: Additional biomarker labels to extract into <label>_area columns.
  • additional_run_length_cols: Additional N-scan biomarker labels to extract into n_scan_run_<label> columns.
  • metrics_type: Expected type of GAMetrics instances (fallback when all None).

Returns DataFrame with each row representing GAMetrics for a single file.

convert_metrics_to_df​

def convert_metrics_to_df(    metrics: collections.abc.Mapping[str, typing.Any],    metrics_type: type[MetricsLike],    additional_pathology_prob_cols: collections.abc.Collection[str] | None = None,    additional_area_cols: collections.abc.Collection[str] | None = None,    additional_run_length_cols: collections.abc.Collection[str] | None = None,    record_isinstance_type: type | None = None,) ‑> pandas.core.frame.DataFrame:

Convert a mapping of filename to metrics objects into a wide DataFrame.

Works for any metrics dataclass that exposes to_record() and expected_cols() (e.g. GAMetrics, GAMetricsWithFovea, or the fluid-volume/CST/GCC metrics dataclasses). Values that are not instances of record_isinstance_type (defaulting to metrics_type; e.g. plain-string error reasons stored in place of a successful result) are treated as empty records rather than raising, so an error placeholder still produces a row with all metric columns unset.

Arguments

  • metrics: Mapping of filename to metrics instances (or non-metrics error values, e.g. strings, which produce an empty record).
  • metrics_type: The metrics dataclass type these records should conform to; used to look up expected_cols(), and (unless record_isinstance_type is given) also for the record-building isinstance guard.
  • additional_pathology_prob_cols: Additional pathology names to extract into max_<name>_probability columns, for metrics types (such as GAMetrics) whose to_record() supports this argument.
  • additional_area_cols: Additional biomarker labels to extract into <label>_area columns, for metrics types (such as GAMetrics) whose to_record() supports this argument.
  • additional_run_length_cols: Additional N-scan biomarker labels to extract into n_scan_run_<label> columns, for metrics types (such as GAMetrics) whose to_record() supports this argument.
  • record_isinstance_type: The type used for the record-building isinstance guard, if different from metrics_type. Useful when metrics_type is a subclass (e.g. GAMetricsWithFovea) but values of the base class (e.g. GAMetrics) should still have their base fields recorded rather than being dropped entirely. Defaults to None, meaning metrics_type is used for both.

Returns DataFrame with one row per mapping entry, an ORIGINAL_FILENAME_METADATA_COLUMN column of the mapping keys, and any missing expected columns added (and logged) as empty columns.

convert_nan_to_zero​

def convert_nan_to_zero(value: Any) ‑> float:

Convert NaN values to 0.

convert_predictions_to_dataframe​

def convert_predictions_to_dataframe(predictions: Any) ‑> pandas.core.frame.DataFrame:

Convert PredictReturnType to DataFrame if necessary.

Arguments

  • predictions: The predictions from the model algorithm.

Returns The predictions as a DataFrame.

create_inference_dataloader​

def create_inference_dataloader(    datasource: FileSystemIterableSource,    schema: BitfountSchema,    datastructure: DataStructure,    batch_size: int | None = None,) ‑> Any:

Create a test-only dataloader for inference, bypassing BitfountDataBunch.

This reproduces the subset of BitfountDataBunch.__init__ that is relevant for inference:

  1. datastructure.set_training_column_split_by_semantic_type(schema)
  2. data_factory.create_dataset(... data_split=TEST, splitter=_InferenceSplitter)
  3. data_factory.create_dataloader(dataset, batch_size)

No train/validation splits are created.

Arguments

  • datasource: The datasource to iterate over.
  • schema: A BitfountSchema (should already have features populated via BitfountSchema.add_dataframe_features).
  • datastructure: A DataStructure (from build_column_config).
  • batch_size: Batch size for the dataloader.

Returns A BitfountDataLoader wrapping the test dataset.

cst_metrics_from_cache​

def cst_metrics_from_cache(    accessor: Any,) ‑> dict[str, CSTMetrics | None]:

Rebuild a filename→CSTMetrics map from a cst_calculation cache.

Typed wrapper over the shared metrics_from_cache for the CST background-phase calc cache, mapping an uncomputable file (null metrics_json) to None.

Arguments

  • accessor: A CacheAccessor for the cst_calculation cache partition.

Returns A {file_id: CSTMetrics | None} map.

extract_bscan_predictions​

def extract_bscan_predictions(ga_inferences: Any) ‑> tuple[str, ...]:

Extract ordered B-scan prediction strings from a GA inferences dict.

Arguments

  • ga_inferences: The value of the inferences_json (or inferences_json_ga) field from a merged row. Expected to be a dict keyed by B-scan index (numeric strings or ints).

Returns Tuple of prediction strings sorted by numeric B-scan index, or an empty tuple if the input is not a non-empty dict.

fluid_metrics_from_cache​

def fluid_metrics_from_cache(    accessor: Any,) ‑> dict[str, FluidVolumeMetrics | None]:

Rebuild a filename→FluidVolumeMetrics map from a fluid_calculation cache.

Typed wrapper over the shared metrics_from_cache for the fluid-volume background-phase calc cache, mapping an uncomputable file (null metrics_json) to None.

Arguments

  • accessor: A CacheAccessor for the fluid_calculation cache partition.

Returns A {file_id: FluidVolumeMetrics | None} map.

ga_metrics_from_cache​

def ga_metrics_from_cache(    accessor: Any,) ‑> dict[str, GAMetricsWithFovea | None]:

Rebuild a filename→GAMetricsWithFovea map from a ga_calculation cache.

For the v9 flows that run GA calculation in the background, a downstream interactive step receives the step's .cache accessor rather than the in-memory metrics. Thin GA-typed wrapper over the shared metrics_from_cache (which also handles schema-drift tolerance), mapping an uncomputable file (null metrics_json) to None.

Arguments

  • accessor: A CacheAccessor for the ga_calculation cache partition.

Returns A {file_id: GAMetricsWithFovea | None} map suitable for convert_ga_metrics_to_df.

gcc_metrics_from_cache​

def gcc_metrics_from_cache(    accessor: Any,) ‑> dict[str, GCCMetrics | None]:

Rebuild a filename→GCCMetrics map from a gcc_calculation cache.

Typed wrapper over the shared metrics_from_cache for the GCC background-phase calc cache, mapping an uncomputable file (null metrics_json) to None.

Arguments

  • accessor: A CacheAccessor for the gcc_calculation cache partition.

Returns A {file_id: GCCMetrics | None} map.

get_data_for_files​

def get_data_for_files(    datasource: Any,    filenames: list[str],    file_key_col: str | None = None,    use_cache: bool = True,) ‑> pandas.core.frame.DataFrame:

Retrieve data from a datasource for a given list of files.

The dataframe returned will be sorted to match the ordering of the files in filenames.

get_lesion_sizes​

def get_lesion_sizes(    num_lesions: int,    labeled_array: NDArray[Any],    slice_thickness: float,    pixel_spacing_column: float,) ‑> list[float]:

Calculate the size of each lesion in mm^2.

Arguments

  • num_lesions: Number of lesions in the image.
  • labeled_array: Numpy array of shape (num_bscans, num_cols) where each pixel is labelled with the lesion number it belongs to.
  • slice_thickness: Thickness of each B-scan in mm.
  • pixel_spacing_column: Spacing between columns in mm.

Returns List of lesion sizes in mm^2.

get_max_ga_bscan_index​

def get_max_ga_bscan_index(column_masks_arr: NDArray[Any], ga_area: float) ‑> int | None:

Return the index of the B-scan with the largest GA area.

Arguments

  • column_masks_arr: Numpy array mask of shape (num_bscans, num_cols).
  • ga_area: Total GA area in mm^2.

Returns Index of the B-scan with the largest GA area, or None if no GA.

get_missing_data_reason​

def get_missing_data_reason(    slice_thickness: float,    pixel_spacing_column: float,    original_filename: str | float,    pixel_spacing_row: float | None = &lt;object object&gt;,) ‑> str | None:

Check required per-row fields and return a skip reason if any are missing.

Arguments

  • slice_thickness: Value from the Slice Thickness metadata column. Will be float('nan') when the datasource has no value for this file.
  • pixel_spacing_column: Value from the Pixel Spacing Column metadata column. Will be float('nan') when the datasource has no value for this file.
  • original_filename: Value from the filename metadata column. Will be float('nan') when the left-join produced no match.
  • pixel_spacing_row: Optional value from the Pixel Spacing Row metadata column. Only the fluid-volume calculation requires this axis, so it defaults to the _UNSET sentinel (not checked) — GA callers pass nothing and are unaffected. When a value is supplied it is validated like the other fields, including a genuine None/NaN cell (reported as pixel_spacing_row).

Returns A "missing_data:<field>,..." string if one or more fields are NaN / null, or None if all fields are present.

get_shortest_distance_from_image_centre​

def get_shortest_distance_from_image_centre(    column_masks_arr: NDArray[Any],    labeled_array: NDArray[Any],    num_lesions: int,    slice_thickness: float,    pixel_spacing_column: float,) ‑> float:

Calculate the distance from the image centre to the nearest lesion.

Image centre is used as a proxy for the fovea.

Arguments

  • column_masks_arr: Numpy array mask of shape (num_bscans, num_cols).
  • labeled_array: Numpy array of shape (num_bscans, num_cols) where each pixel is labelled with the lesion number it belongs to.
  • num_lesions: Number of lesions in the image.
  • slice_thickness: Thickness of each B-scan in mm.
  • pixel_spacing_column: Spacing between columns in mm.

Returns Distance from the image centre to the nearest lesion in mm.

init_model_for_inference​

def init_model_for_inference(model: Any) ‑> None:

Initialise a model for inference without binding any data.

Calls model.initialise_model() with no datasource so that the model's internal create_model() and weight-loading logic runs, but no BitfountDataBunch or dataloaders are created.

After this call you can assign model.test_dl directly and run model._pl_trainer.test().

Arguments

  • model: A Bitfount model instance (already has weights loaded via deserialize).

lesion_metrics_from_cache​

def lesion_metrics_from_cache(    accessor: Any,) ‑> dict[str, LesionMetrics | None]:

Rebuild a filename→LesionMetrics map from a lesion_calculation cache.

Typed wrapper over the shared metrics_from_cache for the area background-phase calc cache, mapping an uncomputable file (null metrics_json) to None.

Arguments

  • accessor: A CacheAccessor for the lesion_calculation cache partition.

Returns A {file_id: LesionMetrics | None} map.

metrics_from_cache​

def metrics_from_cache(    cache_df: pandas.core.frame.DataFrame,    metrics_type: type[~_MetricT],    keep: collections.abc.Collection[str] | None = None,) ‑> dict[str, typing.Optional[~_MetricT]]:

Reconstruct a {file_id: metrics instance | None} map from a cache frame.

The single shared reader for every cache-backed sink (GA via ga_metrics_from_cache, plus the biomarker_tabulation CST/GCC/fluid builders). Reads each row's schemaless metrics_json blob (via metrics_json_by_file_id) and rebuilds metrics_type from it directly — the blob's keys match the dataclass's fields one-to-one (GA persists asdict minus the raw probability arrays; CST/GCC/fluid persist their to_record()). A null blob (an uncomputable file) maps to None.

Reconstruction is TOLERANT: the calc caches are task_hash-scoped and upsert-only (never pruned), so a row written by an older app version can outlive a change to its dataclass. A blob whose keys no longer match the current fields is logged and mapped to None (treated as uncomputable) rather than raising TypeError and aborting the whole step over one file.

Arguments

  • cache_df: The metric step's cache partition frame (file_id + metrics_json columns).
  • metrics_type: The metrics dataclass to reconstruct (e.g. CSTMetrics).
  • keep: If given, only these file_ids are reconstructed; others are skipped entirely (not just dropped later), so a caller that knows its selection avoids reconstructing an unbounded never-pruned partition.

Returns A {file_id: metrics_type instance | None} map, insertion-ordered by the cache frame's rows.

metrics_json_by_file_id​

def metrics_json_by_file_id(    cache_df: pandas.core.frame.DataFrame,) ‑> dict[str, dict[str, typing.Any] | None]:

Map file_id → its metrics_json dict (or None) from a cache frame.

The single reader of a scan-metrics cache partition's schemaless metrics_json blob, shared by every cache-backed sink (ga_metrics_from_cache here and the biomarker_tabulation builders). A row whose metrics_json is null (an uncomputable file the calculation skipped/failed) maps to None; the caller decides what an absent metric becomes (a reconstructed dataclass, an all-NA row, ...). An empty frame — or one without a file_id column — yields an empty map.

Arguments

  • cache_df: A scan-metrics cache partition frame (file_id + metrics_json columns), e.g. from CacheAccessor.to_dataframe().

Returns A {file_id: metrics_json_dict | None} map, insertion-ordered by the frame's rows.

parse_bscan_predictions​

def parse_bscan_predictions(    bscan_prediction_strs: tuple[str, ...],    slice_thickness: float,    pixel_spacing_column: float,    all_segmentation_labels: dict[str, int],    ga_area_include_segmentations: list[str],    ga_area_exclude_segmentations: list[str],) ‑> ParsedBScanPredictions:

Parse raw B-scan prediction strings into columnar masks and probabilities.

This function was extracted from _WorkerSide._parse_bscan_predictions in ga_trial_calculation_algorithm_base.py. The only change is that configuration values (segmentation labels) are passed as explicit parameters instead of being read from self.

Arguments

  • bscan_prediction_strs: Tuple of JSON prediction strings, one per B-scan.
  • slice_thickness: Thickness of each B-scan in mm.
  • pixel_spacing_column: Spacing between columns in mm.
  • all_segmentation_labels: Mapping of segmentation class name → index.
  • ga_area_include_segmentations: Segmentations used to include GA area.
  • ga_area_exclude_segmentations: Segmentations used to exclude GA area.

Returns A ParsedBScanPredictions containing columnar masks, class probabilities, and class areas per B-scan.

parse_mask_json​

def parse_mask_json(    json_data: Any, labels: dict[str, int],) ‑> numpy.ndarray[typing.Any, typing.Any]:

Parse segmentation mask(s) from JSON.

Arguments

  • json_data: The model output JSON data for the masks.
  • labels: The segmentation classes to generate masks for mapped to their index in the mask data.

Returns A (num_segmentations, image_height, image_width) uint8 array valued 1 where a class is present and 0 elsewhere.

resolve_output_path​

def resolve_output_path(    datasource: BaseSource,) ‑> str:

Resolve the output directory for reports.

Looks for datasource.out_path (the datasource's configured output_path kwarg, or settings.paths.cache_dir by default). Falls back to <cwd>/prefect_reports if unavailable.

run_inference​

def run_inference(    model: Any,    datasource: FileSystemIterableSource,    schema: BitfountSchema,    datastructure: DataStructure,    batch_size: int | None = None,) ‑> PredictReturnType:

End-to-end inference: init model, build dataloader, predict.

Composes init_model_for_inference, create_inference_dataloader, and the model's prediction machinery into a single call.

Arguments

  • model: A Bitfount model instance with weights loaded.
  • datasource: The datasource to run inference on.
  • schema: A populated BitfountSchema.
  • datastructure: A configured DataStructure.
  • batch_size: Batch size for inference.

Returns PredictReturnType (preds + keys).

Notes Some Hub models aggregate a whole forward batch into a single prediction (e.g. exam-level ophthalmology models). Run at a batch size

1, they emit fewer predictions than input records, tripping the "predictions vs keys" contract check. Since we cannot edit those models (they are fetched from the Hub), we run optimistically at the requested batch_size and, only if that contract fails, transparently retry the same chunk at one record per forward batch — the size at which no model can collapse per-record results. Well-behaved models never hit the retry, so they keep their throughput.

supplied_observations_from_cache​

def supplied_observations_from_cache(accessor: Any) ‑> pandas.core.frame.DataFrame:

Build the patient-grain supplied-observations frame.

The patient id is emitted under EHR_BITFOUNT_PATIENT_ID_COL so the existing patient_data merge renames and joins it unchanged.

Arguments

  • accessor: A CacheAccessor for the patient_enrichment partition.

Returns A frame with one row per patient: the bitfount patient id and that patient's list of observation entries. Empty (with both columns present) when the partition holds no rows.

use_default_rename_columns​

def use_default_rename_columns(    datasource: Any, rename_columns: collections.abc.Mapping[str, str] | None = None,) ‑> collections.abc.Mapping[str, str] | None:

Sets the default columns to include based on the datasource.