Skip to main content

thickness_metrics

Shared retinal-layer thickness computation primitives.

These pure free functions are used by both the CST and GCC thickness calculation steps. Extracted here so that neither step package depends on the other, mirroring how ga_metrics.py serves both GA steps.

Ported verbatim from the legacy abstract base _BaseThicknessWorkerSide in federated/algorithms/ophthalmology/base_thickness_calculation_algorithm.py. The only change is that per-instance configuration previously read from self (desired inner/outer layers, strict-measurement flag, landmark index, etc.) is now passed as explicit parameters. Metric-specific logic (_calculate_metric, region statistics, glaucoma staging) is intentionally NOT extracted and lives in each consuming step.

Module​

Functions​

build_layer_boundaries​

def build_layer_boundaries(    parsed_predictions: Sequence[RLPrediction | None],    keep_modes: Mapping[RetinalLayer, KeepMode],) ‑> dict[RetinalLayer, numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]] | None:

Interpolate each requested layer onto a common (num_bscans, width) grid.

Each layer is interpolated over its own X-range and left NaN outside it. That restriction is not cosmetic: np.interp clamps to the endpoint value outside the data range rather than returning NaN, so without it a layer detected across half the B-scan would report a flat, invented boundary across the other half.

Restricting per layer is also what makes a pair correct without any pairwise logic. Each filled region is one contiguous interval, so the intersection of two of them is the interval where both layers were detected, and a subtraction of the two rows is NaN everywhere else.

The grid width spans every requested layer, so asking for an extra layer can widen the grid. The extra columns are NaN for the layers that do not reach them, so no NaN-aware reduction measures anything different. It can still differ in the last bits: np.nanmean sums pairwise and groups by array length, so a longer array of the same values regroups the additions. That is about 1e-16 relative, and only when the extra layer is the widest one requested.

Arguments

  • parsed_predictions: B-scan predictions in B-scan order, as returned by parse_bscan_predictions. None entries yield an all-NaN row.
  • keep_modes: The layers to extract, each mapped to the KeepMode used to collapse multivalued X positions.

Returns One (num_bscans, width) float array per requested layer, or None if no B-scan yielded a usable boundary for any requested layer.

build_thickness_map​

def build_thickness_map(    layer_pred: pd.Series,    pixel_spacing_row_mm: float,    inner_layer: RetinalLayer,    outer_layer: RetinalLayer,    strict_measurement: bool,) ‑> tuple[numpy.ndarray[typing.Any, numpy.dtype[typing.Any]], str, str] | None:

Build thickness map from layer segmentations with fallback support.

Arguments

  • layer_pred: Layer predictions for one file.
  • pixel_spacing_row_mm: Pixel spacing in row direction (mm/pixel).
  • inner_layer: Desired inner layer for the thickness measurement.
  • outer_layer: Desired outer layer for the thickness measurement.
  • strict_measurement: If True, only calculate if both desired inner and outer layers are available (no fallback).

Returns

  • None if no valid thickness map can be built.
  • Otherwise, a tuple containing:
    • thickness_map_um: of shape (num_slices, width) in micrometers.
    • inner_layer_name: Name of inner layer for thickness map.
    • outer_layer_name: Name of outer layer for thickness map.

build_thickness_map_with_boundaries​

def build_thickness_map_with_boundaries(    layer_pred: pd.Series,    pixel_spacing_row_mm: float,    inner_layer: RetinalLayer,    outer_layer: RetinalLayer,    strict_measurement: bool,    extra_layers: Mapping[RetinalLayer, KeepMode] | None = None,) ‑> ThicknessMap | None:

Build a thickness map, and any extra layer boundaries, in one pass.

The thickness itself is the absolute difference of the selected pair's boundary grids, so the NaN outside each layer's own X-range is what restricts the measurement to where both layers were detected.

extra_layers are extracted in the same walk over each B-scan's polygons and returned alongside. They take no part in selecting the measured pair and are never subject to fallback: an extra layer is requested by name, so a caller measuring it can only ever measure that layer or nothing. They do widen the shared grid when they extend further than the pair does, which appends NaN columns — see build_layer_boundaries for the one consequence that has, which is float regrouping rather than a different measurement.

Arguments

  • layer_pred: Layer predictions for one file.
  • pixel_spacing_row_mm: Pixel spacing in row direction (mm/pixel).
  • inner_layer: Desired inner layer for the thickness measurement.
  • outer_layer: Desired outer layer for the thickness measurement.
  • strict_measurement: If True, only calculate if both desired inner and outer layers are available (no fallback).
  • extra_layers: Additional layers to extract in the same pass, each with the KeepMode to collapse multivalued X positions. A layer already in the selected pair is ignored, keeping the pair's own keep mode.

Returns A ThicknessMap, or None if nothing at all could be extracted. When no pair can be selected but extra_layers were requested, the result carries those boundaries with its three pair fields None.

dedupe_layer_points​

def dedupe_layer_points(    points: NDArray[Any], keep_mode: KeepMode = min,) ‑> numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]:

Remove duplicate Y coordinates, keeping only one Y value per X.

For retinal layer boundaries, when multiple Y values exist at the same X:

  • Inner layers (ILM, RNFL, etc.): keep minimum Y
  • Outer layers (RPE, Bruch's, etc.): keep maximum Y

Arguments

  • points: Nx2 array where each row is [x, y]
  • keep_mode: "KeepMode.MIN" to keep minimum Y, "KeepMode.MAX" to keep maximum Y

Returns Mx2 array with unique X values (M lte N)

filled_x_range​

def filled_x_range(row: NDArray[Any]) ‑> tuple[int, int] | None:

The inclusive X-range a build_layer_boundaries row was filled over.

Arguments

  • row: One B-scan's interpolated boundary row.

Returns (x_min, x_max), or None for an all-NaN row.

find_fallback_layer​

def find_fallback_layer(    preferred_layer: RetinalLayer,    available_layers: set[RetinalLayer],    search_direction: "Literal['inner', 'outer']",) ‑> RetinalLayer | None:

Find a fallback layer if the preferred layer is not available.

get_available_layers​

def get_available_layers(    bscan_prediction: RLPrediction, available_layers: set[RetinalLayer],) ‑> set[RetinalLayer]:

Collect unique layer names from a B-scan prediction into available_layers.

parse_bscan_prediction​

def parse_bscan_prediction(    bscan_prediction: Any,) ‑> RLPrediction | None:

Preprocess B-scan prediction string for JSON parsing.

parse_bscan_predictions​

def parse_bscan_predictions(    layer_pred: pd.Series,) ‑> tuple[list[RLPrediction | None], set[RetinalLayer]] | None:

Normalise every B-scan cell in a file row and collect the layers present.

One pass over the prediction columns, in B-scan order. The returned list is index-aligned with that order, with None held for a B-scan whose cell could not be normalised, so downstream indexing by B-scan stays valid.

Arguments

  • layer_pred: Layer predictions for one file.

Returns (parsed_predictions, available_layers), or None if the row has no prediction columns or no parseable layer in any of them.

parse_center_coordinates​

def parse_center_coordinates(    file_row: pd.Series,    landmark_idx: int,    center_landmark_type: str = 'landmark',    metric_name: str = '',) ‑> tuple[float, float, float] | None:

Parse center coordinates from landmark model predictions.

This follows the same approach as the GA trial calculation algorithm, extracting a specific landmark from the central slice. Assumes landmarks are contiguous triples per slice: [start, end, middle].

Arguments

  • file_row: Row containing 'central_slice' and 'landmarks' columns.
  • landmark_idx: Index of the landmark to use (0=start, 1=end, 2=middle).
  • center_landmark_type: Nature of the center landmark ("fovea"/"macula"). Used only for log text.
  • metric_name: Name of the metric being calculated. Used only for log text.

Returns Tuple of (slice, x, y) coordinates, or None if parsing fails.

parse_layer_boundaries​

def parse_layer_boundaries(    pred_data: RLPrediction | str,    keep_modes: Mapping[RetinalLayer, KeepMode],) ‑> dict[RetinalLayer, numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]]:

Extract boundary coordinates for several layers in one pass.

The instance list is walked once, so asking for four layers costs the same walk as asking for two. This is the whole reason the function exists: a step measuring two layer pairs that share a layer would otherwise walk every B-scan's polygons once per pair.

Arguments

  • pred_data: Prediction dict with key 'mask' -> 'instances'.
  • keep_modes: The layers to extract, each mapped to the KeepMode used to collapse multivalued X positions.

Returns One Nx2 array of [x, y] boundary points per requested layer, in the order requested. A requested layer with too few points gets an empty (0, 2) array, so every requested key is always present. An empty dict if parsing raises.

parse_layer_prediction​

def parse_layer_prediction(    pred_data: RLPrediction | str, inner_layer_name: str, outer_layer_name: str,) ‑> dict[str, numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]]:

Parse layer segmentation prediction and extract boundary coordinates.

Handles multiple polygon instances per layer (combines their points). A two-layer wrapper over parse_layer_boundaries, kept because callers hold layer names selected by select_measurement_layers rather than enum members.

Arguments

  • pred_data: Prediction dict with key 'mask' -> 'instances'.
  • inner_layer_name: Name of the inner layer to extract (e.g., "ILM").
  • outer_layer_name: Name of the outer layer to extract (e.g., "RPE Layer").

Returns Dictionary mapping layer names to lists of Nx2 arrays. Each array has rows of [x, y] coordinates representing the boundary. Format: {'ILM': [array([[x0, y0], [x1, y1], ...])], 'RPE Layer': [...]} Only returns the two requested layers.

select_measurement_layers​

def select_measurement_layers(    available_layers: set[RetinalLayer],    inner_layer: RetinalLayer,    outer_layer: RetinalLayer,    strict_measurement: bool,) ‑> tuple[str | None, str | None]:

Select which layers to use for thickness measurement with fallback logic.

Arguments

  • available_layers: Set of available layer names.
  • inner_layer: Desired inner layer for the thickness measurement.
  • outer_layer: Desired outer layer for the thickness measurement.
  • strict_measurement: If True, only use the desired layers (no fallback).

Returns Tuple of (inner_layer_name, outer_layer_name) or (None, None) if no suitable pair found.

split_points​

def split_points(    points: NDArray[Any],) ‑> list[numpy.ndarray[typing.Any, numpy.dtype[typing.Any]]]:

Split flat points array into Nx2 coordinate array with [x, y] per row.

Arguments

  • points: Flat array [x0, y0, x1, y1, ..., xN, yN]

Returns List containing single Nx2 array where each row is [x, y]. Returns empty list if insufficient points.

Classes​

KeepMode​

class KeepMode(*args, **kwds):

Which Y to keep where a polyline is multivalued at one X.

"min" keeps the shallowest and "max" the deepest, Y increasing with depth. The convention is a role, not a property of the layer: the inner boundary of a pair keeps "min" and the outer keeps "max", so the pair reports the largest separation available. Callers state it per layer, because a layer that is inner in one pair can be outer in another.

Ancestors​

Variables​

  • static MAX
  • static MIN

RLInstance​

class RLInstance(*args, **kwargs):

Typed Dict for Retinal Layer instance.

Variables​

  • static classId : int
  • static className : str
  • static points : list[float] | list[int]
  • static probability : float
  • static type : Literal['polygon']

RLMask​

class RLMask(*args, **kwargs):

Typed Dict for Retinal Layer mask.

Variables​

RLPrediction​

class RLPrediction(*args, **kwargs):

Typed Dict for Retinal Layer prediction.

Variables​

ThicknessMap​

class ThicknessMap(    thickness_um: NDArray[Any] | None,    inner_layer_name: str | None,    outer_layer_name: str | None,    boundaries: dict[RetinalLayer, NDArray[Any]],):

A thickness map, with the boundary grids it was derived from.

The boundaries are handed back so a caller measuring a second layer pair on the same volume — cst_calculation v2 measures EZ to RPE alongside ILM to RPE — gets them from the same pass over the polygons.

The three pair fields are all-or-nothing: they are None together when no pair could be selected, which is a result carrying only the layers the caller requested by name. That case exists so failing to measure one pair does not withhold a second pair that was never at issue.

Attributes

  • thickness_um: (num_bscans, width) separation of the selected pair, or None if no pair was selected.
  • inner_layer_name: The inner layer actually measured, after fallback, or None if no pair was selected.
  • outer_layer_name: The outer layer actually measured, after fallback, or None if no pair was selected.
  • boundaries: Every requested layer's (num_bscans, width) boundary grid, the selected pair included.

Variables​

  • static inner_layer_name : str | None
  • static outer_layer_name : str | None