Skip to main content

v1

CST/CRT calculation step v1.

Module​

Submodules​

Functions​

task_fn​

def task_fn(    datasource: BaseSource,    config: CSTCalculationConfig,    layer_predictions: CacheAccessor,    center_predictions: CacheAccessor,    filenames: list[str],    cache: CacheProtocol,    task_hash: str,    project_id: str | None = None,    run_id: str | None = None,) ‑> CSTCalculationResult:

Compute CST/CRT metrics for each file and persist them to cache.

Arguments

  • datasource: The datasource providing DICOM metadata (slice thickness, pixel spacing row/column) for each file.
  • config: CST calculation configuration.
  • layer_predictions: Cache accessor for retinal-layers model inference results (background step retinal_layers_inference.cache).
  • center_predictions: Cache accessor for fovea model inference results (background step fovea_inference.cache).
  • filenames: List of file IDs to process.
  • cache: Cache instance to persist the CST metrics into.
  • task_hash: Partition key for the cst_calculation table; a config change lands in a fresh partition.
  • project_id: Provenance only — the project that triggered this run. NOT part of the cache key: rows are keyed by (task_hash, file_id) so two projects on the same datasource share them.
  • run_id: Optional provenance run ID.

Returns CSTCalculationResult carrying the number of rows persisted and a CacheAccessor scoped to this task_hash (cst_calculation.cache).

Classes​

Config​

class Config(**data: Any):

Config for the Central Subfield Thickness (CST/CRT) calculation step.

All fields are required — defaults are provided by the YAML template.

The ilm_layer_name / rpe_layer_name fields must name valid retinal layers (RetinalLayer.from_str), replicating the validation the legacy CSTCalculationAlgorithm factory performed in its __init__.

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.

Subclasses​

Variables​

  • static cst_radius_mm : float
  • static fovea_landmark_idx : int
  • static ilm_layer_name : str
  • static model_config
  • static rpe_layer_name : str
  • static strict_measurement : bool

Result​

class Result(**data: Any):

Result of a CST/CRT calculation task.

records_stored reports how many per-file CST/CRT rows were persisted to the cst_calculation cache table. The metrics themselves are not returned in-memory: this step runs in background mode, where an in-memory result does not cross the phase boundary, so a downstream step reads the rows via the <step_name>.cache accessor (or records in the same process).

records_stored means two different things depending on which path the task took: on the compute path it is the number of rows persisted by this invocation (len(records) in persist_cst_metrics); on the read-before-compute early return, where nothing was stale, it is instead the total row count of the step's own partition (own_accessor.count() in cst_calculation_task) — there being no new rows to report, the field falls back to describing what is already there.

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.

Ancestors​

Variables​

  • static model_config
  • cache : CacheAccessor | None - Return the CacheAccessor for reading this result in a DAG pipeline.

  • records : pandas.core.frame.DataFrame - Retrieve this result's rows as a DataFrame.

    The DataFrame is fetched from the cache on each access — it is not stored in the result object, allowing lazy access to large datasets without materialising them in memory until needed.

    Returns: A pandas.DataFrame with one row per cached record.

    Raises: RuntimeError: If no accessor is attached (e.g. the result was serialised across a Prefect task boundary). In a DAG pipeline use <step_name>.cache instead.

Static methods​

from_accessor​

def from_accessor(    accessor: CacheAccessor, *, records_stored: int | None = None,) ‑> Self:

Inherited from:

CacheBackedResult.from_accessor :

Build a result carrying accessor, in one call rather than two.

_accessor is a PrivateAttr, so it cannot be passed to the constructor and every producer would otherwise construct the result and then reach in to attach the accessor. That two-step is the shape this replaces.

Arguments

  • accessor: The CacheAccessor backing <step>.cache, scoped to the partition this result stands for.
  • records_stored: Rows this result stands for. Defaults to accessor.count() — the whole partition — which is what a step that persisted nothing this run reports. A step that wrote rows passes what it wrote; see the field's docstring for why the two are not interchangeable.

Returns An instance of the calling subclass, with the accessor attached.

Methods​

model_post_init​

def model_post_init(self: BaseModel, context: Any, /) ‑> None:

Inherited from:

CacheBackedResult.model_post_init :

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Arguments

  • self: The BaseModel instance.
  • context: The context.