xpdeep_model

Define the xpdeep explainable model.

Classes:

Name	Description
XpdeepModel	Xpdeep Model class.

Functions:

Name	Description
wrapped_from_dict	Overwrite from_dict method of BuildModelPipelineInputOutput class.

Attributes:

Name	Type	Description
DROPOUT_TYPES
BATCH_NORM_TYPES

DROPOUT_TYPES = nn.Dropout | nn.Dropout2d | nn.Dropout3d | nn.AlphaDropout | nn.FeatureAlphaDropout

BATCH_NORM_TYPES = nn.BatchNorm1d | nn.BatchNorm2d | nn.BatchNorm3d | nn.SyncBatchNorm | nn.LazyBatchNorm1d | nn.LazyBatchNorm2d | nn.LazyBatchNorm3d

XpdeepModel

Xpdeep Model class.

Parameters:

Name	Type	Description	Default
feature_extraction	AbstractModule	The feature extraction model, responsible to extract the most important and coherent features prior to the task you want to achieve.	required
task_learner	AbstractModule	The task learner model is responsible to achieve your task (classification etc.), given a set of meaningful extracted features.	required
backbone	AbstractModule | None	The backbone model, having the same role as a traditional backbone model on a neural network, default None.	None
decision_graph_parameters	ModelDecisionGraphParameters	Internal parameters and architecture of the Xpdeep explainable model.	required
example_dataset	FittedParquetDataset	A FittedParquetDataset object containing a fitted schema, required to build the model. This dataset will not be used for training process but only for model definition.	required
seed	int	Seed for reproducibility.	0

Attributes:

Name	Type	Description
id	str

Methods:

Name	Description
__attrs_post_init__	Post init.
__setattr__	Set attribute.
from_torch	Build a Xpdeep model from torch model.
get_output_size	Infer the model output size without batch size as it is required to serialize a loss function.
__repr__	Represent the model.

feature_extraction: AbstractModule

task_learner: AbstractModule

decision_graph_parameters: ModelDecisionGraphParameters

example_dataset: FittedParquetDataset

backbone: AbstractModule | None = None

seed: int = 0

id: str = field(init=False)

__attrs_post_init__() -> None

Post init.

Source code in src/xpdeep/model/xpdeep_model.py

def __attrs_post_init__(self) -> None:
    """Post init."""
    with contextlib.suppress(DuplicatedRemoteObjectError, AttributeError):
        self.example_dataset.save()

    body = BuildModelPipelineInput(
        type_="BUILD_MODEL",
        feature_extraction=convert_serialized_module(self.feature_extraction.to_pydantic()),
        task_learner=convert_serialized_module(self.task_learner.to_pydantic()),
        internal_model_complexity=self.decision_graph_parameters.internal_model_complexity,
        feature_extraction_output_type=BuildModelPipelineInputFeatureExtractionOutputType(
            cast(FeatureExtractionOutputType, self.decision_graph_parameters.feature_extraction_output_type).value
        ),
        seed=self.seed,
        train_dataset_artifact_id=self.example_dataset.id,
        backbone=convert_serialized_module(self.backbone.to_pydantic()) if self.backbone is not None else None,
        graph_depth=self.decision_graph_parameters.graph_depth
        if self.decision_graph_parameters.graph_depth is not None
        else "default",
        frozen_model=self.decision_graph_parameters.frozen_model,
    )

    with ClientFactory.CURRENT.get()() as client:
        build_model_pipeline = handle_api_validation_errors(
            launch_pipeline.sync(project_id=Project.CURRENT.get().model.id, body=body, client=client),
        )

    from xpdeep.utils.pipelines import model_building  # noqa : PLC0415

    self.id = model_building.get_pipeline_result(build_model_pipeline.id).id

__setattr__(attr: str, value: object) -> None

Set attribute.

Source code in src/xpdeep/model/xpdeep_model.py

@initialized_client_verification
@initialized_project_verification
def __setattr__(self, attr: str, value: object) -> None:
    """Set attribute."""
    if hasattr(self, attr) and hasattr(self, "id"):
        message = "Updating XpdeepModel instance attributes is not allowed. Consider creating a new one."
        raise AttributeError(message)

    object.__setattr__(self, attr, value)

from_torch(example_dataset: FittedParquetDataset, feature_extraction: nn.Module, task_learner: nn.Module, decision_graph_parameters: ModelDecisionGraphParameters, backbone: nn.Module | None = None, seed: int = 0) -> Self

Build a Xpdeep model from torch model.

Source code in src/xpdeep/model/xpdeep_model.py

@classmethod
def from_torch(  # noqa: PLR0913, PLR0917
    cls,
    example_dataset: FittedParquetDataset,
    feature_extraction: nn.Module,
    task_learner: nn.Module,
    decision_graph_parameters: ModelDecisionGraphParameters,
    backbone: nn.Module | None = None,
    seed: int = 0,
) -> Self:
    """Build a Xpdeep model from torch model."""
    XpdeepModel.__assert_no_dropout_nor_batch_norm(task_learner)
    XpdeepModel.__assert_no_dropout_nor_batch_norm(feature_extraction)
    XpdeepModel.__assert_no_dropout_nor_batch_norm(backbone) if backbone is not None else None

    input_size = example_dataset.fitted_schema.input_size

    feature_extractor_device = next(iter(feature_extraction.parameters())).device
    inputs = torch.randn(2, *input_size[1:], device=feature_extractor_device)

    if feature_extractor_device != next(iter(task_learner.parameters())).device:
        msg = "Feature extractor device doe snot match task learner device."
        raise ValueError(msg)
    if backbone is not None and feature_extractor_device != next(iter(backbone.parameters())).device:
        msg = "Feature extractor device doe snot match backbone device."
        raise ValueError(msg)

    # export with batch normalization layers for instance requires more than 1 sample
    if backbone is not None:
        backbone_module = ApiModule.from_torch_module(backbone, inputs)
        inputs_feature_extraction = backbone_module(feature_extraction)
    else:
        backbone_module = None
        inputs_feature_extraction = inputs

    feature_extractor_module = ApiModule.from_torch_module(feature_extraction, inputs_feature_extraction)

    inputs_task_learner = feature_extraction(inputs_feature_extraction)
    task_learner_module = ApiModule.from_torch_module(task_learner, inputs_task_learner)

    return cls(
        feature_extraction=feature_extractor_module,
        task_learner=task_learner_module,
        backbone=backbone_module,
        decision_graph_parameters=decision_graph_parameters,
        seed=seed,
        example_dataset=example_dataset,
    )

get_output_size(schema: FittedSchema) -> tuple[int, ...]

Infer the model output size without batch size as it is required to serialize a loss function.

Source code in src/xpdeep/model/xpdeep_model.py

def get_output_size(self, schema: FittedSchema) -> tuple[int, ...]:
    """Infer the model output size without batch size as it is required to serialize a loss function."""
    x = MultiModal(
        torch.ones((5, *schema.input_size[1:]))
    )  # Batch size of one is not accepted by batch normalization layer.
    if self.backbone is not None:
        x = self.backbone(x)
    x = self.feature_extraction.forward_with_casting(x)
    x = self.task_learner.forward_with_casting(x)
    return tuple(x.shape.args[0])

__repr__() -> str

Represent the model.

Source code in src/xpdeep/model/xpdeep_model.py

def __repr__(self) -> str:
    """Represent the model."""
    model_string = repr(self.decision_graph_parameters)
    model_string += f"\n\nFeature extraction model:\n{self.feature_extraction.model}"
    model_string += f"\n\nTask learner model:\n{self.task_learner.model}"
    if self.backbone is not None:
        model_string += f"\n\nBackbone model:\n{self.backbone.model}"

    return model_string

wrapped_from_dict(cls: type[BuildModelPipelineInputOutput], src_dict: Mapping[str, object]) -> BuildModelPipelineInputOutput

Overwrite from_dict method of BuildModelPipelineInputOutput class.

Source code in src/xpdeep/model/xpdeep_model.py

@no_type_check
def wrapped_from_dict(
    cls: type[BuildModelPipelineInputOutput], src_dict: Mapping[str, object]
) -> BuildModelPipelineInputOutput:
    """Overwrite from_dict method of BuildModelPipelineInputOutput class."""
    return cls(
        type_=src_dict["type"],
        feature_extraction=SerializedModuleOutput.from_dict(src_dict["feature_extraction"]),
        task_learner=SerializedModuleOutput.from_dict(src_dict["task_learner"]),
        internal_model_complexity=src_dict["internal_model_complexity"],
        feature_extraction_output_type=BuildModelPipelineInputOutputFeatureExtractionOutputType(
            src_dict["feature_extraction_output_type"]
        ),
        seed=src_dict["seed"],
        train_dataset_artifact_id=src_dict["train_dataset_artifact_id"],
        backbone=SerializedModuleOutput.from_dict(src_dict["backbone"]) if src_dict["backbone"] is not None else None,
        graph_depth=src_dict["graph_depth"],
        frozen_model=src_dict["frozen_model"],
    )