preprocessing

Just-in-time preprocessing of cached raw model inference.

Raw inference is cached unprocessed so it can be shared across tasks; the format-specific transforms a calculation step needs are selected by interrogating the cached frame's shape (see bitfount.preprocessing.pipelines) and applied (apply_preprocessing / apply_inference_preprocessing) inside that step's own preprocessing.py.

Module

Submodules

• bitfount.preprocessing.apply - Apply preprocessing to cached raw inference frames.
• bitfount.preprocessing.detectors - Shape detectors: interrogate a cached frame to decide which pipeline it needs.
• bitfount.preprocessing.factory - Shape-based preprocessor factory shared by every consuming step.
• bitfount.preprocessing.metadata - Datasource metadata fetching for preprocessing transforms.
• bitfount.preprocessing.models - Typed config models for the preprocessing steps.
• bitfount.preprocessing.pipelines - Centralised preprocessing pipelines (ordered lists of typed transforms).

Functions

apply_inference_preprocessing

def apply_inference_preprocessing(    df: pd.DataFrame,    configs: list[PreprocessingConfig],    *,    datasource: BaseSource,    file_id_column: str = 'file_id',) ‑> pandas.core.frame.DataFrame:

Apply a preprocessing pipeline to a cached frame, wiring datasource metadata.

Convenience for consuming steps: given the pipeline a step has selected (by interrogating the frame's shape — see bitfount.preprocessing.detectors), applies it with a datasource-backed fetcher for any transform that needs extra columns (e.g. Columns/Rows for fovea scaling).

Arguments

• df: A cached-inference DataFrame from a CacheAccessor.
• configs: The ordered preprocessing-step configs to apply.
• datasource: Datasource used to fetch any columns_required metadata.
• file_id_column: Name of the file-id column on df.

Returns The frame with its packed dict column preprocessed. Returns df unchanged when it is empty or configs is empty.

apply_preprocessing

def apply_preprocessing(    df: pd.DataFrame,    configs: list[PreprocessingConfig],    *,    fetch_metadata: MetadataFetcher | None = None,    file_id_column: str = 'file_id',    inferences_column: str = 'inferences_json',) ‑> pandas.core.frame.DataFrame:

Apply configs to the packed dict column of df.

Arguments

• df: A cached-inference DataFrame (e.g. from CacheAccessor) with a file-id column and a packed dict column.
• configs: Ordered preprocessing-step configs (a pipeline from bitfount.preprocessing.pipelines).
• fetch_metadata: Callback supplying any columns_required datasource columns (e.g. Columns/Rows for fovea scaling). Required only when configs declare columns_required.
• file_id_column: Name of the file-id column on df.
• inferences_column: Name of the packed dict column on df.

Returns A copy of df with inferences_column replaced by the preprocessed (re-packed) dict. Returns df unchanged when it is empty or configs is empty.

build_preprocessor

def build_preprocessor(    candidates: Sequence[ShapeCandidate], *, step_name: str,) ‑> collections.abc.Callable[..., pandas.core.frame.DataFrame]:

Build a step's preprocess from its ordered shape candidates.

Arguments

• candidates: Ordered (shape_detector, pipeline) pairs. The first detector that matches the frame wins; its pipeline is applied.
• step_name: The consuming step's name, used in the no-match warning.

Returns preprocess(df, *, datasource): applies the first matching candidate's pipeline, or returns df unchanged when it is empty or no candidate matches (passthrough + warn).

fetch_postprocessor_metadata

def fetch_postprocessor_metadata(    datasource: BaseSource, keys: list[str], columns: list[str],) ‑> pandas.core.frame.DataFrame | None:

Get datasource rows used to supply postprocessor metadata columns.

For a FileSystemIterableSource this passes skip_non_tabular_data=True so only the tabular metadata (e.g. Columns/Rows for the fovea landmark scaling) is loaded rather than full scan payloads, and degrades gracefully to None rather than failing the consuming step.

Arguments

• datasource: The datasource to read metadata rows from.
• keys: The file keys (ORIGINAL_FILENAME_METADATA_COLUMN values) to fetch rows for.
• columns: The metadata columns required by the postprocessors (used for logging context only; the merge is performed by the caller).

Returns A metadata DataFrame aligned to keys, or None when the datasource cannot supply it.

matches_fovea_v7_output

def matches_fovea_v7_output(    df: pd.DataFrame, inferences_column: str = 'inferences_json',) ‑> bool:

Whether df holds raw fovea v7 landmark output.

Fingerprint: an integer-"0" wide column whose value is a dict carrying the v7 landmark fields (fovea_pit_coordinates).

matches_pathology_output

def matches_pathology_output(df: pd.DataFrame) ‑> bool:

Whether df holds pathology model output (segmentation-per-B-scan).

Fingerprint: a wide column matching Pixel_Data_\d+_prediction whose element dict carries a classes key OR a cnv_probability key. Both are absent from retinal-layers output, which shares the wide-column shape — so this value-structure test is what disambiguates the two (see matches_retinal_layers_output).

matches_retinal_layers_output

def matches_retinal_layers_output(df: pd.DataFrame) ‑> bool:

Whether df holds retinal-layers model output (layer segmentation).

Fingerprint: a wide column matching Pixel_Data_\d+_prediction whose element dict carries instances — either at the top level (element['instances'], Heidelberg) or nested under mask (element['mask']['instances'], Altris) — AND carries NEITHER classes NOR cnv_probability. The classes/cnv exclusion is the primary disambiguator from pathology output (see matches_pathology_output).

wide_columns

def wide_columns(    df: pd.DataFrame, inferences_column: str = 'inferences_json',) ‑> set[str]:

Return the wide column names encoded in the packed inference frame.

Each row's inferences_json is a dict whose keys are the wide prediction columns (see apply_preprocessing). Returns the keys of the first non-null dict, or an empty set if the frame carries no inference dicts.

Classes

ExtractionSpec

class ExtractionSpec(**data: Any):

One json_extract_to_columns extraction.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables

• static model_config

• static source_path : str

• static target_column : str

FieldMapping

class FieldMapping(**data: Any):

One json_restructure field move (dot-notation paths).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables

• static model_config

• static source_path : str

• static target_path : str

JSONExtractToColumnsConfig

class JSONExtractToColumnsConfig(**data: Any):

Extract JSON dict fields into new DataFrame columns.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables

• static drop_source : bool

• static extractions : list[ExtractionSpec]

• static model_config

• static source_column : str

• static type : Literal['json_extract_to_columns']

JSONKeyRenameConfig

class JSONKeyRenameConfig(**data: Any):

Rename keys within JSON dict columns.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables

• static column_patterns : list[str]

• static key_mappings : list[KeyMapping]

• static model_config

• static recursive : bool

• static type : Literal['json_key_rename']

JSONRestructureConfig

class JSONRestructureConfig(**data: Any):

Move fields between levels of a JSON dict column.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables

• static column_patterns : list[str]

• static field_mappings : list[FieldMapping]

• static keep_original : bool

• static model_config

• static type : Literal['json_restructure']

KeyMapping

class KeyMapping(**data: Any):

One json_key_rename key rename.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables

• static model_config

• static source_key : str

• static target_key : str

ScaleLandmarkCoordinatesConfig

class ScaleLandmarkCoordinatesConfig(**data: Any):

Scale landmark coordinates from model-input to target image space.

columns_required is not a postprocessor constructor kwarg — it is read by get_postprocessor_metadata_columns to fetch the target dimension columns (e.g. Columns/Rows) before the transform runs, then dropped.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables

• static columns_required : list[str]

• static drop_target_dimension_columns : bool

• static model_config

• static source_column : str

• static source_height : float

• static source_width : float

• static swap_xy : bool

• static target_column : str | None

• static target_height_column : str

• static target_width_column : str

• static type : Literal['scale_landmark_coordinates']

StringToJSONConfig

class StringToJSONConfig(**data: Any):

Convert string columns containing JSON into dict objects.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables

• static column_patterns : list[str]

• static model_config

• static type : Literal['string_to_json']

ZipFlattenCoordinatesConfig

class ZipFlattenCoordinatesConfig(**data: Any):

Zip slice indices with coordinate groups into flat triples.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables

• static coordinates_path : str

• static drop_source : bool

• static model_config

• static slice_indices_path : str

• static source_column : str

• static target_column : str

• static type : Literal['zip_flatten_coordinates']

---