types
Useful types for Federated Learning.
Module
Functions
get_task_results_directory
def get_task_results_directory(context: Optional[ProtocolContext]) ‑> pathlib.Path:
Return the path to the task results directory based on the provided context.
If context is provided, use the derived task results directory from this. Otherwise, default to the configured OUTPUT_DIR.
Classes
AccessCheckResult
class AccessCheckResult(*args, **kwargs):
Container for the result of the access manager check.
Ancestors
- builtins.dict
Variables
- static
limits : Optional[list[_AMResourceUsageAndLimitJSON]]
- static
response_type : _PodResponseType
AggregatorType
class AggregatorType( value, names=None, *, module=None, qualname=None, type=None, start=1,):
Available aggregator names from bitfount.federated.aggregator
.
AlgorithmType
class AlgorithmType(value, names=None, *, module=None, qualname=None, type=None, start=1):
Available algorithm names from bitfount.federated.algorithm
.
Variables
- static
BscanImageAndMaskGenerationAlgorithm
- static
CSVReportAlgorithm
- static
CSVReportGeneratorAlgorithm
- static
CSVReportGeneratorOphthalmologyAlgorithm
- static
EHRPatientQueryAlgorithm
- static
ETDRSAlgorithm
- static
FederatedModelTraining
- static
FluidVolumeCalculationAlgorithm
- static
FoveaCoordinatesAlgorithm
- static
GATrialCalculationAlgorithm
- static
GATrialCalculationAlgorithmAmethyst
- static
GATrialCalculationAlgorithmBronze
- static
GATrialCalculationAlgorithmCharcoal
- static
GATrialCalculationAlgorithmJade
- static
GATrialPDFGeneratorAlgorithm
- static
GATrialPDFGeneratorAlgorithmAmethyst
- static
GATrialPDFGeneratorAlgorithmJade
- static
HuggingFaceImageClassificationInference
- static
HuggingFaceImageSegmentationInference
- static
HuggingFacePerplexityEvaluation
- static
HuggingFaceTextClassificationInference
- static
HuggingFaceTextGenerationInference
- static
ImageSelectionAlgorithm
- static
ModelEvaluation
- static
ModelInference
- static
ModelTrainingAndEvaluation
- static
NextGenPatientInfoDownloadAlgorithm
- static
PrivateSqlQuery
- static
RecordFilterAlgorithm
- static
ReduceCSVAlgorithmCharcoal
- static
S3UploadAlgorithm
- static
SqlQuery
- static
TIMMFineTuning
- static
TIMMInference
- static
TrialInclusionCriteriaMatchAlgorithm
- static
TrialInclusionCriteriaMatchAlgorithmAmethyst
- static
TrialInclusionCriteriaMatchAlgorithmBronze
- static
TrialInclusionCriteriaMatchAlgorithmCharcoal
- static
TrialInclusionCriteriaMatchAlgorithmJade
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: Optional[PodDetailsConfig] = None, data_config: Optional[PodDataConfig] = None, schema: Optional[Union[str, os.PathLike, BitfountSchema]] = 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 : Optional[PodDataConfig]
- static
datasource : BaseSource
- static
datasource_details : Optional[PodDetailsConfig]
- static
name : str
- static
schema : Optional[Union[str, os.PathLike, BitfountSchema]]
EHRConfig
class EHRConfig( base_url: str, provider: EHRProvider, enterprise_url: Optional[str] = None, smart_on_fhir_url: Optional[str] = None, smart_on_fhir_resource_server_url: Optional[str] = None,):
Configuration for EHR details.
Ancestors
- bitfount.types.UsedForConfigSchemas
Variables
- static
base_url : str
- static
enterprise_url : Optional[str]
- static
provider : Literal['nextgen enterprise', 'nextech intellechartpro r4', 'smarthealthit r4', 'epic r4']
- static
smart_on_fhir_resource_server_url : Optional[str]
- static
smart_on_fhir_url : Optional[str]
HubConfig
class HubConfig( username: Optional[str], secrets: Optional[Union[APIKeys, ExternallyManagedJWT]], session: Optional[BitfountSession] = None, session_info: Optional[dict] = None,):
Configuration for connecting to Bitfount Hub.
Variables
- static
secrets : Optional[Union[APIKeys, ExternallyManagedJWT]]
- static
session : Optional[BitfountSession]
- static
session_info : Optional[dict]
- static
username : Optional[str]
HuggingFaceImageClassificationInferenceDictionary
class HuggingFaceImageClassificationInferenceDictionary(*args, **kwargs):
Hugging Face dictionary response for image classification.
Ancestors
- builtins.dict
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:
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: Optional[FileSystemFilterConfig], data_split: Optional[DataSplitConfig],):
Minimal serializable configuration required for creating a datasource.
Variables
- static
data_split : Optional[DataSplitConfig]
- static
datasource_args : _JSONDict
- static
datasource_cls_name : str
- static
file_system_filters : Optional[FileSystemFilterConfig]
- static
name : str
MinimalSchemaGenerationConfig
class MinimalSchemaGenerationConfig( datasource_name: str, description: Optional[str], column_descriptions: Optional[Union[Mapping[str, Mapping[str, str]], Mapping[str, str]]], ignore_cols: Optional[list[str]], force_stypes: "Optional[MutableMapping[Literal['categorical', 'continuous', 'image', 'text', 'image_prefix'], list[str]]]",):
Minimal serializable configuration required for creating a schema.
Variables
- static
column_descriptions : Union[Mapping[str, Mapping[str, str]], Mapping[str, str], ForwardRef(None)]
- static
datasource_name : str
- static
description : Optional[str]
- static
force_stypes : Optional[MutableMapping[Literal['categorical', 'continuous', 'image', 'text', 'image_prefix'], list[str]]]
- static
ignore_cols : Optional[list]
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:
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( *, inference_limits: dict[str, InferenceLimits] = {}, model_urls: dict[str, ModelURLs] = {}, task_context: Optional[TaskContext] = None, project_id: Optional[str] = None, task_id: Optional[str] = None,):
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
: Optional. 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
: Optional. The ID of the task this context is for.
Variables
- static
inference_limits : dict
- static
model_urls : dict
- static
project_id : Optional[str]
- static
task_context : Optional[TaskContext]
- static
task_id : Optional[str]
Methods
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(value, names=None, *, module=None, qualname=None, type=None, start=1):
Available protocol names from bitfount.federated.protocol
.
Variables
- static
BasicOCTProtocol
- static
DataExtractionProtocolCharcoal
- static
FederatedAveraging
- static
FluidVolumeScreeningProtocol
- static
GAScreeningProtocol
- static
GAScreeningProtocolAmethyst
- static
GAScreeningProtocolBronze
- static
GAScreeningProtocolBronzeWithEHR
- static
GAScreeningProtocolCharcoal
- static
GAScreeningProtocolJade
- static
InSiteInsightsProtocol
- static
InferenceAndCSVReport
- static
InferenceAndImageOutput
- static
InferenceAndReturnCSVReport
- static
InstrumentedInferenceAndCSVReport
- static
NextGenSearchProtocol
- static
ResultsOnly
- static
RetinalDiseaseProtocolCobalt
SerializedAggregator
class SerializedAggregator(*args, **kwargs):
Serialized representation of an aggregator.
Ancestors
- builtins.dict
Variables
- static
class_name : str
SerializedAlgorithm
class SerializedAlgorithm(*args, **kwargs):
Serialized representation of an algorithm.
Ancestors
- builtins.dict
Variables
- static
class_name : str
- static
datastructure : typing_extensions.NotRequired[SerializedDataStructure]
- static
model : typing_extensions.NotRequired[SerializedModel]
SerializedDataStructure
class SerializedDataStructure(*args, **kwargs):
Serialized representation of a data structure.
Ancestors
- builtins.dict
Variables
- static
compatible_datasources : typing_extensions.NotRequired[list[str]]
- static
schema_requirements : typing_extensions.NotRequired[dict[str, typing.Any]]
- static
table : typing_extensions.NotRequired[typing.Union[str, dict[str, str]]]
SerializedModel
class SerializedModel(*args, **kwargs):
Serialized representation of a model.
Ancestors
- builtins.dict
Variables
- static
class_name : str
- static
datastructure : NotRequired[SerializedDataStructure]
- static
hub : NotRequired[Optional[BitfountHub]]
- static
schema : NotRequired[_StrAnyDict]
SerializedProtocol
class SerializedProtocol(*args, **kwargs):
Serialized representation of a protocol.
Ancestors
- builtins.dict
Variables
- static
aggregator : typing_extensions.NotRequired[SerializedAggregator]
- static
algorithm : Union[SerializedAlgorithm, list[SerializedAlgorithm]]
- static
class_name : str
- static
primary_results_path : typing_extensions.NotRequired[str]
TaskContext
class TaskContext(value, names=None, *, module=None, qualname=None, type=None, start=1):
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.
TextGenerationDictionary
class TextGenerationDictionary(*args, **kwargs):
Hugging Face dictionary response for text generation.
Ancestors
- builtins.dict
Variables
- static
generated_text : str