functions
Fluid-volume calculation functions (ported from the v8 algorithm).
These are pure free-function ports of the fluid-volume math that lived on
_WorkerSide in
bitfount.federated.algorithms.ophthalmology.fluid_volume_calculation_algorithm.
The legacy algorithm file is left untouched (additive port); the math here is
preserved verbatim (voxel formula, scipy.ndimage.label usage, zeroed-metrics
early returns, and rounding).
The low-level convert_nan_to_zero and parse_mask_json primitives are reused
from steps.data_utils rather than re-ported. The distance helper is fluid's
own 3-axis variant (image centre as fovea proxy) and is deliberately not the
2-axis GA helper.
Module
Functions
compute_fluid_metrics_for_scan
def compute_fluid_metrics_for_scan( bscan_predictions: tuple[str, ...], slice_thickness: float, pixel_spacing_column: float, pixel_spacing_row: float, fluid_volume_include_segmentations: list[str],) ‑> FluidVolumeMetrics:Compute fluid-volume metrics for a single scan.
This encapsulates the per-file computation loop body (steps 6b-6n) from
_WorkerSide.run(), parallel to compute_ga_metrics_with_fovea_for_scan.
Raises
Exception: Propagates any exception from prediction parsing or metric computation so that the caller can handle/skip.
get_lesion_volumes
def get_lesion_volumes( labeled_array: NDArray[Any], num_lesions: int, voxel_volume: float,) ‑> list[float]:Calculate the volume of each lesion in nL.
Ported verbatim from _WorkerSide._get_lesion_volumes.
Arguments
labeled_array: Array of shape (num_bscans, num_rows, num_cols) where each voxel is labelled with the lesion number it belongs to.num_lesions: Number of lesions in the image.voxel_volume: Volume of a single voxel in nL.
Returns List of lesion volumes in nL.
get_max_bscan_index
def get_max_bscan_index(mask: NDArray[Any], total_volume: float) ‑> int | None:Return the index of the B-scan with the largest fluid volume.
Ported verbatim from _WorkerSide._get_max_bscan_index.
Arguments
mask: Binary mask of fluid across B-scans (shape: num_bscans, num_rows, num_cols).total_volume: Total fluid volume across all B-scans (used to check if any fluid is present).
Returns Index of the B-scan with the largest fluid volume, or None if no fluid.
get_shortest_distance_from_image_centre
def get_shortest_distance_from_image_centre( mask: NDArray[Any], voxel_sizes: tuple[float, float, float],) ‑> float:Distance from the image centre to the nearest lesion voxel (3-axis).
Ported verbatim from _WorkerSide._get_shortest_distance_from_image_centre.
Image centre is used as a proxy for the fovea. This is the fluid-specific
3-axis variant and is distinct from the GA 2-axis helper.
Arguments
mask: Array of shape (num_bscans, num_rows, num_cols) where each voxel is 1 if part of a lesion, 0 otherwise.voxel_sizes: Per-axis voxel sizes in mm, in the same axis order asmask: (slice_thickness, pixel_spacing_row, pixel_spacing_column).
Returns Distance from the image centre to the nearest lesion voxel in mm, or NaN when the mask has no lesion voxels.
parse_bscan_fluid_predictions_with_cnv
def parse_bscan_fluid_predictions_with_cnv( bscan_prediction_strs: tuple[str, ...], fluid_volume_include_segmentations: list[str],) ‑> tuple[dict[str, numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]], list[float]]:Convert per-B-scan predictions to fluid masks and CNV probabilities.
Ported verbatim from
_WorkerSide._parse_bscan_fluid_predictions_with_cnv. Reads the top-level
cnv_probability, requires a "mask" key (frames without one are skipped),
and reads mask.instances / mask.metadata (via parse_mask_json).
Arguments
bscan_prediction_strs: Tuple of predictions for a single scan's B-scans.fluid_volume_include_segmentations: Segmentation labels to build masks for. Falls back toFLUID_VOLUME_INCLUDE_SEGMENTATION_LABELSwhen empty.
Returns A tuple of:
- dict mapping each included segmentation label to a stacked binary
mask of shape (num_bscans, num_rows, num_cols), or a
(0, 0, 0)array when that label had no B-scan masks. - list of CNV probabilities (one per parsed B-scan).