Skip to main content

types

Useful types for Federated Learning.

Module​

Functions​

get_task_results_directory​

def get_task_results_directory(context: ProtocolContext) ‑> pathlib.Path:

Return the path to the task results directory based on the provided context.

Classes​

AccessCheckResult​

class AccessCheckResult(*args, **kwargs):

Container for the result of the access manager check.

Variables​

AggregatorType​

class AggregatorType(*args, **kwds):

Available aggregator names from bitfount.federated.aggregator.

Ancestors​

Variables​

  • static Aggregator
  • static SecureAggregator

AlgorithmType​

class AlgorithmType(*args, **kwds):

Available algorithm names from bitfount.federated.algorithm.

Ancestors​

Variables​

  • static CSVReportAlgorithm
  • static CodeNormalisationAlgorithm
  • static EHRPatientIDExtractAlgo
  • static EHRPatientInfoDownloadAlgorithm
  • static EHRPatientQueryAlgorithm
  • static FederatedModelTraining
  • static HuggingFaceImageClassificationInference
  • static HuggingFaceImageSegmentationInference
  • static HuggingFaceImageTextGenerationInference
  • static HuggingFaceNERInference
  • static HuggingFacePerplexityEvaluation
  • static HuggingFaceTextClassificationInference
  • static HuggingFaceTextGenerationInference
  • static ImageSelectionAlgorithm
  • static LongitudinalAlgorithm
  • static MONAIBundleInference
  • static ModelEvaluation
  • static ModelInference
  • static ModelTrainingAndEvaluation
  • static PatientIDExchangeAlgorithm
  • static PrivateSqlQuery
  • static S3UploadAlgorithm
  • static S3UploadAlgorithmWithMRNTracking
  • static SqlQuery
  • static TIMMFederatedTraining
  • static TIMMFineTuning
  • static TIMMInference
  • static TrialInclusionCriteriaMatchAlgorithmEHR
  • static TrialMatchingAlgorithm

DatasourceContainer​

class DatasourceContainer(    name: str,    datasource: BaseSource,    datasource_details: PodDetailsConfig,    data_config: PodDataConfig,    schema: BitfountSchema,):

Contains a datasource and all the data related to it.

This represents a datasource configuration post-data-loading/configuration and so the data config and schema must be present.

Variables​

  • static data_config : PodDataConfig
  • static datasource : BaseSource
  • static datasource_details : PodDetailsConfig
  • static name : str
  • static schema : BitfountSchema

DatasourceContainerConfig​

class DatasourceContainerConfig(    name: str,    datasource: BaseSource,    datasource_details: PodDetailsConfig | None = None,    data_config: PodDataConfig | None = None,    schema: str | os.PathLike[str] | BitfountSchema | None = None,):

Contains a datasource and maybe some data related to it.

This represents a datasource configuration pre-data-loading/configuration and so the data config and schema are not required.

Variables​

  • static data_config : PodDataConfig | None
  • static datasource : BaseSource
  • static datasource_details : PodDetailsConfig | None
  • static name : str
  • static schema : str | os.PathLike[str] | BitfountSchema | None

HubConfig​

class HubConfig(    username: str | None,    secrets: APIKeys | RefreshableJWT | None,    session: BitfountSession | None = None,    session_info: dict[Any, Any] | None = None,):

Configuration for connecting to Bitfount Hub.

Variables​

  • static secrets : APIKeys | RefreshableJWT | None
  • static session : BitfountSession | None
  • static session_info : dict[Any, Any] | None
  • static username : str | None

HuggingFaceImageClassificationInferenceDictionary​

class HuggingFaceImageClassificationInferenceDictionary(*args, **kwargs):

Hugging Face dictionary response for image classification.

Variables​

  • static image_classification : str

InferenceLimits​

class InferenceLimits(*, limit: int, total_usage: int):

Container class for model inference usage limits.

Attributes

  • limit: The total number of inferences that can be performed.
  • total_usage: The total number of inferences performed so far.

Variables​

  • static limit : int
  • static total_usage : int
  • initial_total_usage : int - Returns the total usage that was set when this instance was created.

Static methods​

from_access_check_result​

def from_access_check_result(    access_check_result: AccessCheckResult,) ‑> dict[str, InferenceLimits]:

Construct model name -> inference usage limits from access check results.

If a model usage limit is undefined or not present, this indicates that there is no usage limit, and so we do not add this to the constructed dict.

Returns Dictionary of full model name (e.g. "some-model-owner/some-model:12" to the model inference usage limits for that model as detailed in access_check_results).

MinimalDatasourceConfig​

class MinimalDatasourceConfig(    datasource_cls_name: str,    name: str,    datasource_args: _JSONDict,    file_system_filters: FileSystemFilterConfig | None,    data_split: DataSplitConfig | None,    is_reconnection: bool = False,):

Minimal serializable configuration required for creating a datasource.

Variables​

  • static data_split : DataSplitConfig | None
  • static datasource_args : _JSONDict
  • static datasource_cls_name : str
  • static file_system_filters : FileSystemFilterConfig | None
  • static is_reconnection : bool
  • static name : str

MinimalSchemaGenerationConfig​

class MinimalSchemaGenerationConfig(    datasource_name: str,    description: str | None,    column_descriptions: Mapping[str, Mapping[str, str]] | Mapping[str, str] | None,    ignore_cols: list[str] | None,    force_stypes: "MutableMapping[Literal['categorical', 'continuous', 'image', 'text', 'image_prefix'], list[str]] | None",):

Minimal serializable configuration required for creating a schema.

Variables​

  • static datasource_name : str
  • static description : str | None
  • static ignore_cols : list[str] | None

MinimalSchemaUploadConfig​

class MinimalSchemaUploadConfig(    public_metadata: PodPublicMetadata,    access_manager_public_key: RSAPublicKey,    pod_public_key: RSAPublicKey,):

Minimal serializable configuration required for uploading a schema.

Variables​

  • static access_manager_public_key : RSAPublicKey
  • static pod_public_key : RSAPublicKey
  • static public_metadata : PodPublicMetadata

ModelURLs​

class ModelURLs(    *, model_download_url: _S3PresignedURL, model_weights_url: _S3PresignedURL | None,):

Container class for model download URLs from the authorisation checker.

Attributes

  • model_download_url: URL for downloading the model code.
  • model_weights_url: URL for downloading the model weights.

Variables​

  • static model_download_url : bitfount.types._S3PresignedURL
  • static model_weights_url : Optional[bitfount.types._S3PresignedURL]

Static methods​

from_access_check_result​

def from_access_check_result(    access_check_result: AccessCheckResult,) ‑> dict[str, ModelURLs]:

Construct model name -> model download URLs from access check results.

Returns Dictionary of full model name/id (e.g. "some-model-owner/some-model:12" to the model download URL and/or model weights URL for that model as detailed in access_check_results).

ProtocolContext​

class ProtocolContext(    *,    task_id: str,    task_context: TaskContext,    inference_limits: dict[str, InferenceLimits] = {},    model_urls: dict[str, ModelURLs] = {},    project_id: str | None = None,    test_run: bool = False,):

Details needed for the protocol at runtime.

Attributes

  • inference_limits: A mapping of model name (full name, including owner and version) to the inference limits information for that model. e.g. {"some-model-owner/some-model:12": {"limit": 90, "total_usage": 10}}
  • model_urls: A mapping of model name (full name, including owner and version) to any URLs needed to download the model/weights in the context of the task.
  • task_context: Which context (modeller or worker) the task is running in.
  • project_id: Optional. The ID of the project this task is part of.
  • task_id: The ID of the task this context is for.
  • test_run: Whether this is a test run.

Variables​

  • static project_id : str | None
  • static task_id : str
  • static test_run : bool

Methods​

get_project_results_dir​

def get_project_results_dir(self) ‑> pathlib.Path:

Get the directory where project results are stored.

If TASK_RESULTS_DIR is set, that is used as the base. Otherwise, the base will be OUTPUT_DIR/"task-results".

Within that directory, returns the subdirectory named after the project ID. This is the directory containing all task runs for the project.

Returns Path to the project results directory. The directory is created if it doesn't exist.

Raises

  • ValueError: If project_id is not set on this context.

get_task_results_dir​

def get_task_results_dir(self) ‑> pathlib.Path:

Get the directory where task results should be stored for this task run.

If TASK_RESULTS_DIR is set, that is used as the base. Otherwise, the base will be OUTPUT_DIR/"task-results".

Within that directory, create a subdirectories named after the project ID, and task ID, if provided.

ProtocolType​

class ProtocolType(*args, **kwds):

Available protocol names from bitfount.federated.protocol.

Ancestors​

Variables​

  • static EHRNERProtocol
  • static EHRScreeningProtocol
  • static FederatedAveraging
  • static GenericBiomarkerProtocolGranite
  • static InferenceAndCSVReport
  • static InferenceAndCSVReportWithAggregateReporting
  • static InferenceAndImageOutput
  • static InferenceAndReturnCSVReport
  • static InstrumentedInferenceAndCSVReport
  • static NextGenSearchProtocol
  • static PatientTrialMatchingProtocol
  • static ResultsOnly

SMARTBackendEHRConfig​

class SMARTBackendEHRConfig(    client_id: str,    base_url: str,    scopes: list[str],    provider: EHRProvider,    key_pair: Optional[SMARTBackendKeyPair] = None,    list_resource_ids: Optional[list[str]] = None,):

EHR connection for SMART Backend Services (system access, no user browser).

When key_pair is omitted, the pod loads or generates keys under its storage path at startup. When set, the given PEM pair is used and must match client_id registered with the EHR.

smart_backend_auth is set by Pod after the Hub exists: config.smart_backend_auth = SMARTBackendAuth(config, hub). Until then it is None.

smart_backend_refresh_handler is a single RefreshableJWTHandler created by the Pod for token caching across liveness checks and EHR sessions; it is omitted when the config is copied for worker subprocesses.

Variables​

  • static base_url : str
  • static client_id : str
  • static key_pair : Optional[bitfount.federated.types.SMARTBackendKeyPair]
  • static list_resource_ids : list[str] | None
  • static provider : Literal['nextgen enterprise', 'nextech intellechartpro r4', 'smarthealthit r4', 'epic r4', 'modmed r4', 'generic r4']
  • static scopes : list[str]

SMARTStandaloneEHRConfig​

class SMARTStandaloneEHRConfig(    base_url: str, provider: EHRProvider, list_resource_ids: Optional[list[str]] = None,):

Configuration for EHR details.

Variables​

  • static base_url : str
  • static list_resource_ids : list[str] | None
  • static provider : Literal['nextgen enterprise', 'nextech intellechartpro r4', 'smarthealthit r4', 'epic r4', 'modmed r4', 'generic r4']

SerializedAggregator​

class SerializedAggregator(*args, **kwargs):

Serialized representation of an aggregator.

Variables​

  • static class_name : str

SerializedAlgorithm​

class SerializedAlgorithm(*args, **kwargs):

Serialized representation of an algorithm.

Variables​

  • static class_name : str

SerializedDataStructure​

class SerializedDataStructure(*args, **kwargs):

Serialized representation of a data structure.

Variables​

  • static compatible_datasources : list[str]
  • static schema_requirements : dict[str, typing.Any]
  • static table : str | dict[str, str]
  • static task_filters : list[bitfount.federated.types.SerializedTaskFilter]

SerializedModel​

class SerializedModel(*args, **kwargs):

Serialized representation of a model.

Variables​

SerializedProtocol​

class SerializedProtocol(*args, **kwargs):

Serialized representation of a protocol.

Variables​

  • static class_name : str
  • static primary_results_path : str

TaskContext​

class TaskContext(*args, **kwds):

Describes the context (modeller or worker) in which the task is running.

This is used for models where the model differs depending on if it is on the modeller-side or worker-side of the federated process. It is also used for batched execution.

Ancestors​

Variables​

  • static MODELLER
  • static WORKER

TextGenerationDictionary​

class TextGenerationDictionary(*args, **kwargs):

Hugging Face dictionary response for text generation.

Variables​

  • static generated_text : str