Skip to main content

v1

Model inference step v1.

Module​

Submodules​

Functions​

task_fn​

async def task_fn(    datasource: BaseSource,    model_resource: ModelResourceProtocol,    config: ModelInferenceConfig,    schema: BitfountSchema,    datastructure: DataStructure,    cache: CacheProtocol,    task_hash: str,    project_id: str,    run_id: str | None = None,    lifecycle_notifier: LifecycleNotifier | None = None,) ‑> ModelInferenceResult:

Run model inference on the datasource and persist results to cache.

RAW inference results are written directly to the model_inferences cache table via bulk_insert_inference_results — no postprocessing is applied here. Postprocessing is format-specific to each consumer, so it runs just-in-time inside the downstream calc step (see bitfount.preprocessing) rather than at cache time. Caching the raw output lets any task on the same datasource share these rows regardless of how it later postprocesses them. Downstream interactive steps access the rows through a CacheAccessor (<step_name>.cache) rather than receiving an in-memory DataFrame.

Arguments

  • datasource: The datasource to run inference on.
  • model_resource: Model resource used to load the model.
  • config: Model inference configuration (model ref, version, batch size).
  • schema: Datasource schema.
  • datastructure: Datasource data structure.
  • cache: Cache instance to persist inference results into.
  • task_hash: Pipeline task hash identifying this run; used as the partition key in the model_inferences table.
  • project_id: Provenance only — the project that triggered this run. NOT part of the cache key: rows are keyed by (task_hash, model_ref, model_version, file_id) so two projects on the same datasource share them.
  • run_id: The run_id for this phase of execution.
  • lifecycle_notifier: Optional notifier used to signal the batch count (NUMBER_OF_BATCHES) and per-batch ids (CURRENT_BATCH_ID) to the task initiator. Each CURRENT_BATCH_ID(n) is sent right before the n-th file chunk is inferred (mirroring the v8 worker batch loop), so the ids track actual progress. The phase-4 marker (PREPARING_DATA_BATCHES) and the batches-complete boundary are emitted once per DAG by the executor, not here. None (the default) disables signalling — e.g. background runs, which have no initiator mailbox.

Returns ModelInferenceResult with the number of records stored.

Classes​

Config​

class Config(**data: Any):

Config for model inference (fovea or GA).

batch_size is optional and defaults to DEFAULT_INFERENCE_BATCH_SIZE (1) when unset — see resolved_batch_size. The remaining fields are required and provided by the YAML template.

Note: this step caches RAW model output. Format-specific postprocessing is applied just-in-time inside the consuming calc step (see bitfount.preprocessing), so there is no postprocessors field here.

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 batch_size : int | None
  • static model_config
  • static model_ref : str
  • static model_username : str
  • static model_version : int

  • resolved_batch_size : int - Effective batch size, defaulting to 1 when batch_size is unset.

    1 is safe for every model; see DEFAULT_INFERENCE_BATCH_SIZE.

Methods​

get_model_ref​

def get_model_ref(self) ‑> ModelInferenceConfig:

Return self — the config already contains all model ref fields.

This method exists for backward compatibility with code that previously used BitfountModelRef.

Result​

class Result(**data: Any):

Result of a model inference task.

records_stored reports how many inference rows were persisted to the cache, except on the all-cached skip: with nothing left to infer the task persists nothing and reports the number of selected files instead, every one of which already has a row (see model_inference_task).

The predictions DataFrame itself is not returned directly — it must be accessed via the records property (same-process only) or, in a DAG pipeline across Prefect task boundaries, via the <step_name>.cache accessor.

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.