tasks

AnomalibDetection(config, module_function, checkpoint_path=None, run_test=True, report=True)

Bases: Generic[AnomalyDataModuleT], LightningTask[AnomalyDataModuleT]

Anomaly Detection Task.

Parameters:

• config (DictConfig) –

  The experiment configuration

• module_function (DictConfig) –

  The function that instantiates the module and model

• checkpoint_path (str | None, default: None ) –

  The path to the checkpoint to load the model from. Defaults to None.

• run_test (bool, default: True ) –

  Whether to run the test after training. Defaults to False.

• report (bool, default: True ) –

  Whether to report the results. Defaults to False.

Source code in quadra/tasks/anomaly.py

def __init__(
    self,
    config: DictConfig,
    module_function: DictConfig,
    checkpoint_path: str | None = None,
    run_test: bool = True,
    report: bool = True,
):
    super().__init__(
        config=config,
        checkpoint_path=checkpoint_path,
        run_test=run_test,
        report=report,
    )
    self._module: AnomalyModule
    self.module_function = module_function
    self.export_folder = "deployment_model"
    self.report_path = ""
    self.test_results: list[dict] | None = None

module: AnomalyModule property writable

Get the module.

export()

Export model for production.

Source code in quadra/tasks/anomaly.py

def export(self) -> None:
    """Export model for production."""
    if self.config.trainer.get("fast_dev_run"):
        log.warning("Skipping export since fast_dev_run is enabled")
        return

    model = self.module.model

    input_shapes = self.config.export.input_shapes

    half_precision = "16" in self.trainer.precision

    model_json, export_paths = export_model(
        config=self.config,
        model=model,
        export_folder=self.export_folder,
        half_precision=half_precision,
        input_shapes=input_shapes,
        idx_to_class={0: "good", 1: "defect"},
    )

    if len(export_paths) == 0:
        return

    model_json["image_threshold"] = np.round(self.module.image_threshold.value.item(), 3)
    model_json["pixel_threshold"] = np.round(self.module.pixel_threshold.value.item(), 3)
    model_json["anomaly_method"] = self.config.model.model.name

    with open(os.path.join(self.export_folder, "model.json"), "w") as f:
        json.dump(model_json, f, cls=utils.HydraEncoder)

generate_report()

Generate a report for the task and try to upload artifacts.

Source code in quadra/tasks/anomaly.py

def generate_report(self):
    """Generate a report for the task and try to upload artifacts."""
    self._generate_report()
    self._upload_artifacts()

prepare()

Prepare the task.

Source code in quadra/tasks/anomaly.py

def prepare(self) -> None:
    """Prepare the task."""
    super().prepare()
    self.module = self.config.model
    self.module.model = ModelSignatureWrapper(self.module.model)

test()

Lightning test.

Source code in quadra/tasks/anomaly.py

def test(self) -> Any:
    """Lightning test."""
    self.test_results = super().test()
    return self.test_results

Classification(config, output, checkpoint_path=None, lr_multiplier=None, gradcam=False, report=False, run_test=False)

Bases: Generic[ClassificationDataModuleT], LightningTask[ClassificationDataModuleT]

Classification Task.

Parameters:

• config (DictConfig) –

  The experiment configuration

• output (DictConfig) –

  The otuput configuration.

• gradcam (bool, default: False ) –

  Whether to compute gradcams

• checkpoint_path (str | None, default: None ) –

  The path to the checkpoint to load the model from. Defaults to None.

• lr_multiplier (float | None, default: None ) –

  The multiplier for the backbone learning rate. Defaults to None.

• output (DictConfig) –

  The ouput configuration (under task config). It contains the bool "example" to generate figs of discordant/concordant predictions.

• report (bool, default: False ) –

  Whether to generate a report containing the results after test phase

• run_test (bool, default: False ) –

  Whether to run the test phase.

Source code in quadra/tasks/classification.py

def __init__(
    self,
    config: DictConfig,
    output: DictConfig,
    checkpoint_path: str | None = None,
    lr_multiplier: float | None = None,
    gradcam: bool = False,
    report: bool = False,
    run_test: bool = False,
):
    super().__init__(
        config=config,
        checkpoint_path=checkpoint_path,
        run_test=run_test,
        report=report,
    )
    self.output = output
    self.gradcam = gradcam
    self._lr_multiplier = lr_multiplier
    self._pre_classifier: nn.Module
    self._classifier: nn.Module
    self._model: nn.Module
    self._optimizer: torch.optim.Optimizer
    self._scheduler: torch.optim.lr_scheduler._LRScheduler
    self.model_json: dict[str, Any] | None = None
    self.export_folder: str = "deployment_model"
    self.deploy_info_file: str = "model.json"
    self.report_confmat: pd.DataFrame
    self.best_model_path: str | None = None

len_train_dataloader: int property

Get the length of the train dataloader.

optimizer: torch.optim.Optimizer property writable

Get the optimizer.

scheduler: torch.optim.lr_scheduler._LRScheduler property writable

Get the scheduler.

export()

Generate deployment models for the task.

Source code in quadra/tasks/classification.py

def export(self) -> None:
    """Generate deployment models for the task."""
    if self.datamodule.class_to_idx is None:
        log.warning(
            "No `class_to_idx` found in the datamodule, class information will not be saved in the model.json"
        )
        idx_to_class = {}
    else:
        idx_to_class = {v: k for k, v in self.datamodule.class_to_idx.items()}

    # Get best model!
    if self.best_model_path is not None:
        log.info("Saving deployment model for %s checkpoint", self.best_model_path)

        module = self.module.__class__.load_from_checkpoint(
            self.best_model_path,
            model=self.module.model,
            optimizer=self.optimizer,
            lr_scheduler=self.scheduler,
            criterion=self.module.criterion,
            gradcam=False,
        )
    else:
        log.warning("No checkpoint callback found in the trainer, exporting the last model weights")
        module = self.module

    input_shapes = self.config.export.input_shapes

    # TODO: What happens if we have 64 precision?
    half_precision = "16" in self.trainer.precision

    self.model_json, export_paths = export_model(
        config=self.config,
        model=module.model,
        export_folder=self.export_folder,
        half_precision=half_precision,
        input_shapes=input_shapes,
        idx_to_class=idx_to_class,
    )

    if len(export_paths) == 0:
        return

    with open(os.path.join(self.export_folder, self.deploy_info_file), "w") as f:
        json.dump(self.model_json, f)

freeze_layers_by_name(freeze_parameters_name)

Freeze layers specified in freeze_parameters_name.

Parameters:

• freeze_parameters_name (list[str]) –

  Layers that will be frozen during training.

Source code in quadra/tasks/classification.py

def freeze_layers_by_name(self, freeze_parameters_name: list[str]):
    """Freeze layers specified in freeze_parameters_name.

    Args:
        freeze_parameters_name: Layers that will be frozen during training.

    """
    count_frozen = 0
    for name, param in self.model.named_parameters():
        if any(x in name.split(".")[1] for x in freeze_parameters_name):
            log.debug("Freezing layer %s", name)
            param.requires_grad = False

        if not param.requires_grad:
            count_frozen += 1

    log.info("Frozen %d parameters", count_frozen)

freeze_parameters_by_index(freeze_parameters_index)

Freeze parameters specified in freeze_parameters_name.

Parameters:

• freeze_parameters_index (list[int]) –

  Indices of parameters that will be frozen during training.

Source code in quadra/tasks/classification.py

def freeze_parameters_by_index(self, freeze_parameters_index: list[int]):
    """Freeze parameters specified in freeze_parameters_name.

    Args:
        freeze_parameters_index: Indices of parameters that will be frozen during training.

    """
    if getattr(self.config.backbone, "freeze_parameters_name", None) is not None:
        log.warning(
            "Please be aware that some of the model's parameters have already been frozen using \
            the specified freeze_parameters_name. You are combining these two actions."
        )
    count_frozen = 0
    for i, (name, param) in enumerate(self.model.named_parameters()):
        if i in freeze_parameters_index:
            log.debug("Freezing layer %s", name)
            param.requires_grad = False

        if not param.requires_grad:
            count_frozen += 1

    log.info("Frozen %d parameters", count_frozen)

generate_report()

Generate a report for the task.

Source code in quadra/tasks/classification.py

def generate_report(self) -> None:
    """Generate a report for the task."""
    if self.datamodule.class_to_idx is None:
        log.warning("No `class_to_idx` found in the datamodule, report will not be generated")
        return

    if isinstance(self.datamodule, MultilabelClassificationDataModule):
        log.warning("Report generation is not supported for multilabel classification tasks at the moment.")
        return

    log.info("Generating report!")
    if not self.run_test or self.config.trainer.get("fast_dev_run"):
        self.datamodule.setup(stage="test")

    # Deepcopy to remove the inference mode from gradients causing issues when loading checkpoints
    # TODO: Why deepcopy of module model removes ModelSignatureWrapper?
    self.module.model.instance = deepcopy(self.module.model.instance)
    if "16" in self.trainer.precision:
        log.warning("Gradcam is currently not supported with half precision, it will be disabled")
        self.module.gradcam = False
        self.gradcam = False

    predictions_outputs = self.trainer.predict(
        model=self.module, datamodule=self.datamodule, ckpt_path=self.best_model_path
    )
    if not predictions_outputs:
        log.warning("There is no prediction to generate the report. Skipping report generation.")
        return
    all_outputs = [x[0] for x in predictions_outputs]
    all_probs = [x[2] for x in predictions_outputs]
    if not all_outputs or not all_probs:
        log.warning("There is no prediction to generate the report. Skipping report generation.")
        return
    all_outputs = [item for sublist in all_outputs for item in sublist]
    all_probs = [item for sublist in all_probs for item in sublist]
    all_targets = [target.tolist() for im, target in self.datamodule.test_dataloader()]
    all_targets = [item for sublist in all_targets for item in sublist]

    if self.module.gradcam:
        grayscale_cams = [x[1] for x in predictions_outputs]
        grayscale_cams = [item for sublist in grayscale_cams for item in sublist]
        grayscale_cams = np.stack(grayscale_cams)  # N x H x W
    else:
        grayscale_cams = None

    # creating confusion matrix
    idx_to_class = {v: k for k, v in self.datamodule.class_to_idx.items()}
    _, self.report_confmat, accuracy = get_results(
        test_labels=all_targets,
        pred_labels=all_outputs,
        idx_to_labels=idx_to_class,
    )
    output_folder_test = "test"
    test_dataloader = self.datamodule.test_dataloader()
    test_dataset = cast(ImageClassificationListDataset, test_dataloader.dataset)
    self.res = pd.DataFrame(
        {
            "sample": list(test_dataset.x),
            "real_label": all_targets,
            "pred_label": all_outputs,
            "probability": all_probs,
        }
    )
    os.makedirs(output_folder_test, exist_ok=True)
    save_classification_result(
        results=self.res,
        output_folder=output_folder_test,
        confmat=self.report_confmat,
        accuracy=accuracy,
        test_dataloader=self.datamodule.test_dataloader(),
        config=self.config,
        output=self.output,
        grayscale_cams=grayscale_cams,
    )

    if len(self.logger) > 0:
        mflow_logger = get_mlflow_logger(trainer=self.trainer)
        tensorboard_logger = utils.get_tensorboard_logger(trainer=self.trainer)
        artifacts = glob.glob(os.path.join(output_folder_test, "**/*"), recursive=True)
        if self.config.core.get("upload_artifacts") and len(artifacts) > 0:
            if mflow_logger is not None:
                log.info("Uploading artifacts to MLFlow")
                for a in artifacts:
                    if os.path.isdir(a):
                        continue

                    dirname = Path(a).parent.name
                    mflow_logger.experiment.log_artifact(
                        run_id=mflow_logger.run_id,
                        local_path=a,
                        artifact_path=os.path.join("classification_output", dirname),
                    )
            if tensorboard_logger is not None:
                log.info("Uploading artifacts to Tensorboard")
                for a in artifacts:
                    if os.path.isdir(a):
                        continue

                    ext = os.path.splitext(a)[1].lower()

                    if ext in [".png", ".jpg", ".jpeg", ".bmp", ".tiff", ".tif", ".gif"]:
                        try:
                            img = cv2.imread(a)
                            img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
                        except cv2.error:
                            log.info("Could not upload artifact image %s", a)
                            continue
                        output_path = os.path.sep.join(a.split(os.path.sep)[-2:])
                        tensorboard_logger.experiment.add_image(output_path, img, 0, dataformats="HWC")
                    else:
                        utils.upload_file_tensorboard(a, tensorboard_logger)

module(module_config)

Set the module of the model.

Source code in quadra/tasks/classification.py

@LightningTask.module.setter
def module(self, module_config):  # noqa: F811
    """Set the module of the model."""
    module = hydra.utils.instantiate(
        module_config,
        model=self.model,
        optimizer=self.optimizer,
        lr_scheduler=self.scheduler,
        gradcam=self.gradcam,
    )
    if self.checkpoint_path is not None:
        log.info("Loading model from lightning checkpoint: %s", self.checkpoint_path)
        module = module.__class__.load_from_checkpoint(
            self.checkpoint_path,
            model=self.model,
            optimizer=self.optimizer,
            lr_scheduler=self.scheduler,
            criterion=module.criterion,
            gradcam=self.gradcam,
        )
    self._module = module

prepare()

Prepare the experiment.

Source code in quadra/tasks/classification.py

def prepare(self) -> None:
    """Prepare the experiment."""
    super().prepare()
    self.model = self.config.model
    self.optimizer = self.config.optimizer
    self.scheduler = self.config.scheduler
    self.module = self.config.model.module

test()

Test the model.

Source code in quadra/tasks/classification.py

def test(self) -> None:
    """Test the model."""
    if not self.config.trainer.get("fast_dev_run"):
        log.info("Starting testing!")
        self.trainer.test(datamodule=self.datamodule, model=self.module, ckpt_path=self.best_model_path)

train()

Train the model.

Source code in quadra/tasks/classification.py

def train(self):
    """Train the model."""
    super().train()
    if (
        self.trainer.checkpoint_callback is not None
        and hasattr(self.trainer.checkpoint_callback, "best_model_path")
        and self.trainer.checkpoint_callback.best_model_path is not None
        and len(self.trainer.checkpoint_callback.best_model_path) > 0
    ):
        self.best_model_path = self.trainer.checkpoint_callback.best_model_path
        log.info("Loading best epoch weights...")

ClassificationEvaluation(config, output, model_path, report=True, gradcam=False, device=None)

Bases: Evaluation[ClassificationDataModuleT]

Perform a test on an imported Classification pytorch model.

Parameters:

• config (DictConfig) –

  Task configuration

• output (DictConfig) –

  Configuration for the output

• model_path (str) –

  Path to pytorch .pt model file

• report (bool, default: True ) –

  Whether to generate the report of the predictions

• gradcam (bool, default: False ) –

  Whether to compute gradcams

• device (str | None, default: None ) –

  Device to use for evaluation. If None, the device is automatically determined

Source code in quadra/tasks/classification.py

def __init__(
    self,
    config: DictConfig,
    output: DictConfig,
    model_path: str,
    report: bool = True,
    gradcam: bool = False,
    device: str | None = None,
):
    super().__init__(config=config, model_path=model_path, device=device)
    self.report_path = "test_output"
    self.output = output
    self.report = report
    self.gradcam = gradcam
    self.cam: GradCAM

deployment_model: BaseEvaluationModel property writable

Deployment model.

execute()

Execute the evaluation.

Source code in quadra/tasks/classification.py

def execute(self) -> None:
    """Execute the evaluation."""
    self.prepare()
    self.test()
    if self.report:
        self.generate_report()
    self.finalize()

generate_report()

Generate a report for the task.

Source code in quadra/tasks/classification.py

def generate_report(self) -> None:
    """Generate a report for the task."""
    log.info("Generating report!")
    os.makedirs(self.report_path, exist_ok=True)

    save_classification_result(
        results=self.metadata["test_results"],
        output_folder=self.report_path,
        confmat=self.metadata["test_confusion_matrix"],
        accuracy=self.metadata["test_accuracy"],
        test_dataloader=self.datamodule.test_dataloader(),
        config=self.config,
        output=self.output,
        grayscale_cams=self.metadata["grayscale_cams"],
    )

get_classifier(model_config)

Instantiate the classifier from the config.

Source code in quadra/tasks/classification.py

def get_classifier(self, model_config: DictConfig) -> nn.Module:
    """Instantiate the classifier from the config."""
    if "classifier" in model_config:
        log.info("Instantiating classifier <%s>", model_config.classifier["_target_"])
        return hydra.utils.instantiate(
            model_config.classifier, out_features=len(self.model_data["classes"]), _convert_="partial"
        )

    raise ValueError("A `classifier` definition must be specified in the config")

get_pre_classifier(model_config)

Instantiate the pre-classifier from the config.

Source code in quadra/tasks/classification.py

def get_pre_classifier(self, model_config: DictConfig) -> nn.Module:
    """Instantiate the pre-classifier from the config."""
    if "pre_classifier" in model_config and model_config.pre_classifier is not None:
        log.info("Instantiating pre_classifier <%s>", model_config.pre_classifier["_target_"])
        pre_classifier = hydra.utils.instantiate(model_config.pre_classifier, _convert_="partial")
    else:
        log.info("No pre-classifier found in config: instantiate a torch.nn.Identity instead")
        pre_classifier = nn.Identity()

    return pre_classifier

get_torch_model(model_config)

Instantiate the torch model from the config.

Source code in quadra/tasks/classification.py

def get_torch_model(self, model_config: DictConfig) -> nn.Module:
    """Instantiate the torch model from the config."""
    pre_classifier = self.get_pre_classifier(model_config)
    classifier = self.get_classifier(model_config)
    log.info("Instantiating backbone <%s>", model_config.model["_target_"])

    return hydra.utils.instantiate(
        model_config.model, classifier=classifier, pre_classifier=pre_classifier, _convert_="partial"
    )

prepare()

Prepare the evaluation.

Source code in quadra/tasks/classification.py

def prepare(self) -> None:
    """Prepare the evaluation."""
    super().prepare()
    self.datamodule = self.config.datamodule
    self.datamodule.class_to_idx = {v: int(k) for k, v in self.model_data["classes"].items()}
    self.datamodule.num_classes = len(self.datamodule.class_to_idx)

    # prepare_data() must be explicitly called because there is no training
    self.datamodule.prepare_data()
    self.datamodule.setup(stage="test")

prepare_gradcam()

Initializing gradcam for the predictions.

Source code in quadra/tasks/classification.py

def prepare_gradcam(self) -> None:
    """Initializing gradcam for the predictions."""
    if not hasattr(self.deployment_model.model, "features_extractor"):
        log.warning("Gradcam not implemented for this backbone, it will not be computed")
        self.gradcam = False
        return

    if isinstance(self.deployment_model.model.features_extractor, timm.models.resnet.ResNet):
        target_layers = [cast(BaseNetworkBuilder, self.deployment_model.model).features_extractor.layer4[-1]]
        self.cam = GradCAM(
            model=self.deployment_model.model,
            target_layers=target_layers,
        )
        for p in self.deployment_model.model.features_extractor.layer4[-1].parameters():
            p.requires_grad = True
    elif is_vision_transformer(cast(BaseNetworkBuilder, self.deployment_model.model).features_extractor):
        self.grad_rollout = VitAttentionGradRollout(cast(nn.Module, self.deployment_model.model))
    else:
        log.warning("Gradcam not implemented for this backbone, it will not be computed")
        self.gradcam = False

test()

Perform test.

Source code in quadra/tasks/classification.py

@automatic_datamodule_batch_size(batch_size_attribute_name="batch_size")
def test(self) -> None:
    """Perform test."""
    log.info("Running test")
    test_dataloader = self.datamodule.test_dataloader()

    image_labels = []
    probabilities = []
    predicted_classes = []
    grayscale_cams_list = []

    if self.gradcam:
        self.prepare_gradcam()

    with torch.set_grad_enabled(self.gradcam):
        for batch_item in tqdm(test_dataloader):
            im, target = batch_item
            im = im.to(device=self.device, dtype=self.deployment_model.model_dtype).detach()

            if self.gradcam:
                # When gradcam is used we need to remove gradients
                outputs = self.deployment_model(im).detach()
            else:
                outputs = self.deployment_model(im)

            probs = torch.softmax(outputs, dim=1)
            preds = torch.max(probs, dim=1).indices

            probabilities.append(probs.tolist())
            predicted_classes.append(preds.tolist())
            image_labels.extend(target.tolist())
            if self.gradcam and hasattr(self.deployment_model.model, "features_extractor"):
                with torch.inference_mode(False):
                    im = im.clone()
                    if isinstance(self.deployment_model.model.features_extractor, timm.models.resnet.ResNet):
                        grayscale_cam = self.cam(input_tensor=im, targets=None)
                        grayscale_cams_list.append(torch.from_numpy(grayscale_cam))
                    elif is_vision_transformer(
                        cast(BaseNetworkBuilder, self.deployment_model.model).features_extractor
                    ):
                        grayscale_cam_low_res = self.grad_rollout(input_tensor=im, targets_list=preds.tolist())
                        orig_shape = grayscale_cam_low_res.shape
                        new_shape = (orig_shape[0], im.shape[2], im.shape[3])
                        zoom_factors = tuple(np.array(new_shape) / np.array(orig_shape))
                        grayscale_cam = ndimage.zoom(grayscale_cam_low_res, zoom_factors, order=1)
                        grayscale_cams_list.append(torch.from_numpy(grayscale_cam))

    grayscale_cams: torch.Tensor | None = None
    if self.gradcam:
        grayscale_cams = torch.cat(grayscale_cams_list, dim=0)

    predicted_classes = [item for sublist in predicted_classes for item in sublist]
    probabilities = [max(item) for sublist in probabilities for item in sublist]
    if self.datamodule.class_to_idx is not None:
        idx_to_class = {v: k for k, v in self.datamodule.class_to_idx.items()}

    _, pd_cm, test_accuracy = get_results(
        test_labels=image_labels,
        pred_labels=predicted_classes,
        idx_to_labels=idx_to_class,
    )

    res = pd.DataFrame(
        {
            "sample": list(test_dataloader.dataset.x),  # type: ignore[attr-defined]
            "real_label": image_labels,
            "pred_label": predicted_classes,
            "probability": probabilities,
        }
    )

    log.info("Avg classification accuracy: %s", test_accuracy)

    self.res = pd.DataFrame(
        {
            "sample": list(test_dataloader.dataset.x),  # type: ignore[attr-defined]
            "real_label": image_labels,
            "pred_label": predicted_classes,
            "probability": probabilities,
        }
    )

    # save results
    self.metadata["test_confusion_matrix"] = pd_cm
    self.metadata["test_accuracy"] = test_accuracy
    self.metadata["predictions"] = predicted_classes
    self.metadata["test_results"] = res
    self.metadata["probabilities"] = probabilities
    self.metadata["test_labels"] = image_labels
    self.metadata["grayscale_cams"] = grayscale_cams

Evaluation(config, model_path, device=None)

Bases: Generic[DataModuleT], Task[DataModuleT]

Base Evaluation Task with deployment models.

Parameters:

• config (DictConfig) –

  The experiment configuration

• model_path (str) –

  The model path.

• device (str | None, default: None ) –

  Device to use for evaluation. If None, the device is automatically determined.

Source code in quadra/tasks/base.py

def __init__(
    self,
    config: DictConfig,
    model_path: str,
    device: str | None = None,
):
    super().__init__(config=config)

    if device is None:
        self.device = utils.get_device()
    else:
        self.device = device

    self.config = config
    self.model_data: dict[str, Any]
    self.model_path = model_path
    self._deployment_model: BaseEvaluationModel
    self.deployment_model_type: str
    self.model_info_filename = "model.json"
    self.report_path = ""
    self.metadata = {"report_files": []}

deployment_model: BaseEvaluationModel property writable

Deployment model.

prepare()

Prepare the evaluation.

Source code in quadra/tasks/base.py

def prepare(self) -> None:
    """Prepare the evaluation."""
    with open(os.path.join(Path(self.model_path).parent, self.model_info_filename)) as f:
        self.model_data = json.load(f)

    if not isinstance(self.model_data, dict):
        raise ValueError("Model info file is not a valid json")

    for input_size in self.model_data["input_size"]:
        if len(input_size) != 3:
            continue

        # Adjust the transform for 2D models (CxHxW)
        # We assume that each input size has the same height and width
        if input_size[1] != self.config.transforms.input_height:
            log.warning(
                f"Input height of the model ({input_size[1]}) is different from the one specified "
                + f"in the config ({self.config.transforms.input_height}). Fixing the config."
            )
            self.config.transforms.input_height = input_size[1]

        if input_size[2] != self.config.transforms.input_width:
            log.warning(
                f"Input width of the model ({input_size[2]}) is different from the one specified "
                + f"in the config ({self.config.transforms.input_width}). Fixing the config."
            )
            self.config.transforms.input_width = input_size[2]

    self.deployment_model = self.model_path  # type: ignore[assignment]

LightningTask(config, checkpoint_path=None, run_test=False, report=False)

Bases: Generic[DataModuleT], Task[DataModuleT]

Base Experiment Task.

Parameters:

• config (DictConfig) –

  The experiment configuration

• checkpoint_path (str | None, default: None ) –

  The path to the checkpoint to load the model from. Defaults to None.

• run_test (bool, default: False ) –

  Whether to run the test after training. Defaults to False.

• report (bool, default: False ) –

  Whether to generate a report. Defaults to False.

Source code in quadra/tasks/base.py

def __init__(
    self,
    config: DictConfig,
    checkpoint_path: str | None = None,
    run_test: bool = False,
    report: bool = False,
):
    super().__init__(config=config)
    self.checkpoint_path = checkpoint_path
    self.run_test = run_test
    self.report = report
    self._module: LightningModule
    self._devices: int | list[int]
    self._callbacks: list[Callback]
    self._logger: list[Logger]
    self._trainer: Trainer

callbacks: list[Callback] property writable

List[Callback]: The callbacks.

devices: int | list[int] property writable

List[int]: The devices ids.

logger: list[Logger] property writable

List[Logger]: The loggers.

module: LightningModule property writable

trainer: Trainer property writable

add_callback(callback)

Add a callback to the trainer.

Parameters:

• callback (Callback) –

  The callback to add

Source code in quadra/tasks/base.py

def add_callback(self, callback: Callback):
    """Add a callback to the trainer.

    Args:
        callback: The callback to add
    """
    if hasattr(self.trainer, "callbacks") and isinstance(self.trainer.callbacks, list):
        self.trainer.callbacks.append(callback)

execute()

Execute the experiment and all the steps.

Source code in quadra/tasks/base.py

def execute(self) -> None:
    """Execute the experiment and all the steps."""
    self.prepare()
    self.train()
    if self.run_test:
        self.test()
    if self.config.export is not None and len(self.config.export.types) > 0:
        self.export()
    if self.report:
        self.generate_report()
    self.finalize()

finalize()

Finalize the experiment.

Source code in quadra/tasks/base.py

def finalize(self) -> None:
    """Finalize the experiment."""
    super().finalize()
    utils.finish(
        config=self.config,
        module=self.module,
        datamodule=self.datamodule,
        trainer=self.trainer,
        callbacks=self.callbacks,
        logger=self.logger,
        export_folder=self.export_folder,
    )

    if (
        not self.config.trainer.get("fast_dev_run")
        and self.trainer.checkpoint_callback is not None
        and hasattr(self.trainer.checkpoint_callback, "best_model_path")
    ):
        log.info("Best model ckpt: %s", self.trainer.checkpoint_callback.best_model_path)

prepare()

Prepare the experiment.

Source code in quadra/tasks/base.py

def prepare(self) -> None:
    """Prepare the experiment."""
    super().prepare()

    # First setup loggers since some callbacks might need logger setup correctly.
    if "logger" in self.config:
        self.logger = self.config.logger

    if "callbacks" in self.config:
        self.callbacks = self.config.callbacks

    self.devices = self.config.trainer.devices
    self.trainer = self.config.trainer

test()

Test the model.

Source code in quadra/tasks/base.py

def test(self) -> Any:
    """Test the model."""
    log.info("Starting testing!")

    best_model = None
    if (
        self.trainer.checkpoint_callback is not None
        and hasattr(self.trainer.checkpoint_callback, "best_model_path")
        and self.trainer.checkpoint_callback.best_model_path is not None
        and len(self.trainer.checkpoint_callback.best_model_path) > 0
    ):
        best_model = self.trainer.checkpoint_callback.best_model_path

    if best_model is None:
        log.warning(
            "No best checkpoint model found, using last weights for test, this might lead to worse results, "
            "consider using a checkpoint callback."
        )

    return self.trainer.test(model=self.module, datamodule=self.datamodule, ckpt_path=best_model)

train()

Train the model.

Source code in quadra/tasks/base.py

def train(self) -> None:
    """Train the model."""
    log.info("Starting training!")
    utils.log_hyperparameters(
        config=self.config,
        model=self.module,
        trainer=self.trainer,
    )

    self.trainer.fit(model=self.module, datamodule=self.datamodule)

PatchSklearnClassification(config, output, device, automatic_batch_size, half_precision=False)

Bases: Task[PatchSklearnClassificationDataModule]

Patch classification using torch backbone for feature extraction and sklearn to learn a linear classifier.

Parameters:

• config (DictConfig) –

  The experiment configuration

• device (str) –

  The device to use

• output (DictConfig) –

  Dictionary defining which kind of outputs to generate. Defaults to None.

• automatic_batch_size (DictConfig) –

  Whether to automatically find the largest batch size that fits in memory.

• half_precision (bool, default: False ) –

  Whether to use half precision.

Source code in quadra/tasks/patch.py

def __init__(
    self,
    config: DictConfig,
    output: DictConfig,
    device: str,
    automatic_batch_size: DictConfig,
    half_precision: bool = False,
):
    super().__init__(config=config)
    self.device: str = device
    self.output: DictConfig = output
    self.return_polygon: bool = True
    self.reconstruction_results: dict[str, Any]
    self._backbone: ModelSignatureWrapper
    self._trainer: SklearnClassificationTrainer
    self._model: ClassifierMixin
    self.metadata: dict[str, Any] = {
        "test_confusion_matrix": [],
        "test_accuracy": [],
        "test_results": [],
        "test_labels": [],
    }
    self.export_folder: str = "deployment_model"
    self.automatic_batch_size = automatic_batch_size
    self.half_precision = half_precision

backbone: ModelSignatureWrapper property writable

model: ClassifierMixin property writable

sklearn.base.ClassifierMixin: The model.

trainer: SklearnClassificationTrainer property writable

execute()

Execute the experiment and all the steps.

Source code in quadra/tasks/patch.py

def execute(self) -> None:
    """Execute the experiment and all the steps."""
    self.prepare()
    self.train()
    if self.output.report:
        self.generate_report()
    if self.config.export is not None and len(self.config.export.types) > 0:
        self.export()
    self.finalize()

export()

Generate deployment model for the task.

Source code in quadra/tasks/patch.py

def export(self) -> None:
    """Generate deployment model for the task."""
    input_shapes = self.config.export.input_shapes

    idx_to_class = {v: k for k, v in self.datamodule.class_to_idx.items()}

    model_json, export_paths = export_model(
        config=self.config,
        model=self.backbone,
        export_folder=self.export_folder,
        half_precision=self.half_precision,
        input_shapes=input_shapes,
        idx_to_class=idx_to_class,
        pytorch_model_type="backbone",
    )

    if len(export_paths) > 0:
        dataset_info = self.datamodule.info

        horizontal_patches = dataset_info.patch_number[1] if dataset_info.patch_number is not None else None
        vertical_patches = dataset_info.patch_number[0] if dataset_info.patch_number is not None else None
        patch_height = dataset_info.patch_size[0] if dataset_info.patch_size is not None else None
        patch_width = dataset_info.patch_size[1] if dataset_info.patch_size is not None else None
        overlap = dataset_info.overlap

        model_json.update(
            {
                "horizontal_patches": horizontal_patches,
                "vertical_patches": vertical_patches,
                "patch_height": patch_height,
                "patch_width": patch_width,
                "overlap": overlap,
                "reconstruction_method": self.output.reconstruction_method,
                "class_to_skip": self.datamodule.class_to_skip_training,
            }
        )

        with open(os.path.join(self.export_folder, "model.json"), "w") as f:
            json.dump(model_json, f, cls=utils.HydraEncoder)

    dump(self.model, os.path.join(self.export_folder, "classifier.joblib"))

generate_report()

Generate the report for the task.

Source code in quadra/tasks/patch.py

def generate_report(self) -> None:
    """Generate the report for the task."""
    log.info("Generating report!")
    os.makedirs(self.output.folder, exist_ok=True)

    c_matrix = self.metadata["test_confusion_matrix"]
    idx_to_class = {v: k for k, v in self.datamodule.class_to_idx.items()}

    datamodule: PatchSklearnClassificationDataModule = self.datamodule
    val_img_info: list[PatchDatasetFileFormat] = datamodule.info.val_files
    for img_info in val_img_info:
        if not os.path.isabs(img_info.image_path):
            img_info.image_path = os.path.join(datamodule.data_path, img_info.image_path)
        if img_info.mask_path is not None and not os.path.isabs(img_info.mask_path):
            img_info.mask_path = os.path.join(datamodule.data_path, img_info.mask_path)

    false_region_bad, false_region_good, true_region_bad, reconstructions = compute_patch_metrics(
        test_img_info=val_img_info,
        test_results=self.metadata["test_results"],
        patch_num_h=datamodule.info.patch_number[0] if datamodule.info.patch_number is not None else None,
        patch_num_w=datamodule.info.patch_number[1] if datamodule.info.patch_number is not None else None,
        patch_h=datamodule.info.patch_size[0] if datamodule.info.patch_size is not None else None,
        patch_w=datamodule.info.patch_size[1] if datamodule.info.patch_size is not None else None,
        overlap=datamodule.info.overlap,
        idx_to_class=idx_to_class,
        return_polygon=self.return_polygon,
        patch_reconstruction_method=self.output.reconstruction_method,
        annotated_good=datamodule.info.annotated_good,
    )

    self.reconstruction_results = {
        "false_region_bad": false_region_bad,
        "false_region_good": false_region_good,
        "true_region_bad": true_region_bad,
        "reconstructions": reconstructions,
        "reconstructions_type": "polygon" if self.return_polygon else "rle",
        "patch_reconstruction_method": self.output.reconstruction_method,
    }

    with open("reconstruction_results.json", "w") as f:
        json.dump(
            self.reconstruction_results,
            f,
            cls=RleEncoder,
        )

    if hasattr(self.datamodule, "class_to_skip_training") and self.datamodule.class_to_skip_training is not None:
        ignore_classes = [self.datamodule.class_to_idx[x] for x in self.datamodule.class_to_skip_training]
    else:
        ignore_classes = None
    val_dataloader = self.datamodule.val_dataloader()
    save_classification_result(
        results=self.metadata["test_results"],
        output_folder=self.output.folder,
        confusion_matrix=c_matrix,
        accuracy=self.metadata["test_accuracy"],
        test_dataloader=val_dataloader,
        config=self.config,
        output=self.output,
        reconstructions=reconstructions,
        ignore_classes=ignore_classes,
    )

prepare()

Prepare the experiment.

Source code in quadra/tasks/patch.py

def prepare(self) -> None:
    """Prepare the experiment."""
    self.datamodule = self.config.datamodule
    self.backbone = self.config.backbone
    self.model = self.config.model

    if not self.automatic_batch_size.disable and self.device != "cpu":
        self.datamodule.batch_size = automatic_batch_size_computation(
            datamodule=self.datamodule,
            backbone=self.backbone,
            starting_batch_size=self.automatic_batch_size.starting_batch_size,
        )

    self.trainer = self.config.trainer

train()

Train the model.

Source code in quadra/tasks/patch.py

def train(self) -> None:
    """Train the model."""
    log.info("Starting training...!")
    # prepare_data() must be explicitly called if the task does not include a lightining training
    self.datamodule.prepare_data()
    self.datamodule.setup(stage="fit")
    class_to_keep = None
    if hasattr(self.datamodule, "class_to_skip_training") and self.datamodule.class_to_skip_training is not None:
        class_to_keep = [x for x in self.datamodule.class_to_idx if x not in self.datamodule.class_to_skip_training]

    self.model = self.config.model
    self.trainer.change_classifier(self.model)
    train_dataloader = self.datamodule.train_dataloader()
    val_dataloader = self.datamodule.val_dataloader()
    train_dataset = cast(PatchSklearnClassificationTrainDataset, train_dataloader.dataset)
    self.trainer.fit(train_dataloader=train_dataloader)
    _, pd_cm, accuracy, res, _ = self.trainer.test(
        test_dataloader=val_dataloader,
        class_to_keep=class_to_keep,
        idx_to_class=train_dataset.idx_to_class,
        predict_proba=True,
    )

    # save results
    self.metadata["test_confusion_matrix"] = pd_cm
    self.metadata["test_accuracy"] = accuracy
    self.metadata["test_results"] = res
    self.metadata["test_labels"] = [
        train_dataset.idx_to_class[i] if i != -1 else "N/A" for i in res["real_label"].unique().tolist()
    ]

PatchSklearnTestClassification(config, output, model_path, device='cpu')

Bases: Evaluation[PatchSklearnClassificationDataModule]

Perform a test of an already trained classification model.

Parameters:

• config (DictConfig) –

  The experiment configuration

• output (DictConfig) –

  where to save resultss

• model_path (str) –

  path to trained model from PatchSklearnClassification task.

• device (str, default: 'cpu' ) –

  the device where to run the model (cuda or cpu). Defaults to 'cpu'.

Source code in quadra/tasks/patch.py

def __init__(
    self,
    config: DictConfig,
    output: DictConfig,
    model_path: str,
    device: str = "cpu",
):
    super().__init__(config=config, model_path=model_path, device=device)
    self.output = output
    self._backbone: BaseEvaluationModel
    self._classifier: ClassifierMixin
    self.class_to_idx: dict[str, int]
    self.idx_to_class: dict[int, str]
    self.metadata: dict[str, Any] = {
        "test_confusion_matrix": None,
        "test_accuracy": None,
        "test_results": None,
        "test_labels": None,
    }
    self.class_to_skip: list[str] = []
    self.reconstruction_results: dict[str, Any]
    self.return_polygon: bool = True

backbone: BaseEvaluationModel property writable

classifier: ClassifierMixin property writable

deployment_model property writable

Deployment model.

trainer: SklearnClassificationTrainer property writable

execute()

Execute the experiment and all the steps.

Source code in quadra/tasks/patch.py

def execute(self) -> None:
    """Execute the experiment and all the steps."""
    self.prepare()
    self.test()
    if self.output.report:
        self.generate_report()
    self.finalize()

generate_report()

Generate a report for the task.

Source code in quadra/tasks/patch.py

def generate_report(self) -> None:
    """Generate a report for the task."""
    log.info("Generating report!")
    os.makedirs(self.output.folder, exist_ok=True)

    c_matrix = self.metadata["test_confusion_matrix"]
    idx_to_class = {v: k for k, v in self.datamodule.class_to_idx.items()}

    datamodule: PatchSklearnClassificationDataModule = self.datamodule
    test_img_info = datamodule.info.test_files
    for img_info in test_img_info:
        if not os.path.isabs(img_info.image_path):
            img_info.image_path = os.path.join(datamodule.data_path, img_info.image_path)
        if img_info.mask_path is not None and not os.path.isabs(img_info.mask_path):
            img_info.mask_path = os.path.join(datamodule.data_path, img_info.mask_path)

    false_region_bad, false_region_good, true_region_bad, reconstructions = compute_patch_metrics(
        test_img_info=test_img_info,
        test_results=self.metadata["test_results"],
        patch_num_h=datamodule.info.patch_number[0] if datamodule.info.patch_number is not None else None,
        patch_num_w=datamodule.info.patch_number[1] if datamodule.info.patch_number is not None else None,
        patch_h=datamodule.info.patch_size[0] if datamodule.info.patch_size is not None else None,
        patch_w=datamodule.info.patch_size[1] if datamodule.info.patch_size is not None else None,
        overlap=datamodule.info.overlap,
        idx_to_class=idx_to_class,
        return_polygon=self.return_polygon,
        patch_reconstruction_method=self.output.reconstruction_method,
        annotated_good=datamodule.info.annotated_good,
    )

    self.reconstruction_results = {
        "false_region_bad": false_region_bad,
        "false_region_good": false_region_good,
        "true_region_bad": true_region_bad,
        "reconstructions": reconstructions,
        "reconstructions_type": "polygon" if self.return_polygon else "rle",
        "patch_reconstruction_method": self.output.reconstruction_method,
    }

    with open("reconstruction_results.json", "w") as f:
        json.dump(
            self.reconstruction_results,
            f,
            cls=RleEncoder,
        )

    if self.class_to_skip is not None:
        ignore_classes = [datamodule.class_to_idx[x] for x in self.class_to_skip]
    else:
        ignore_classes = None
    test_dataloader = self.datamodule.test_dataloader()
    save_classification_result(
        results=self.metadata["test_results"],
        output_folder=self.output.folder,
        confusion_matrix=c_matrix,
        accuracy=self.metadata["test_accuracy"],
        test_dataloader=test_dataloader,
        config=self.config,
        output=self.output,
        reconstructions=reconstructions,
        ignore_classes=ignore_classes,
    )

prepare()

Prepare the experiment.

Source code in quadra/tasks/patch.py

def prepare(self) -> None:
    """Prepare the experiment."""
    super().prepare()

    idx_to_class = {}
    class_to_idx = {}
    for k, v in self.model_data["classes"].items():
        idx_to_class[int(k)] = v
        class_to_idx[v] = int(k)

    self.idx_to_class = idx_to_class
    self.class_to_idx = class_to_idx
    self.config.datamodule.class_to_idx = class_to_idx

    self.datamodule = self.config.datamodule
    # Configure trainer
    self.trainer = self.config.trainer

    # prepare_data() must be explicitly called because there is no lightning training
    self.datamodule.prepare_data()
    self.datamodule.setup(stage="test")

test()

Run the test.

Source code in quadra/tasks/patch.py

@automatic_datamodule_batch_size(batch_size_attribute_name="batch_size")
def test(self) -> None:
    """Run the test."""
    test_dataloader = self.datamodule.test_dataloader()

    self.class_to_skip = self.model_data["class_to_skip"] if hasattr(self.model_data, "class_to_skip") else None
    class_to_keep = None

    if self.class_to_skip is not None:
        class_to_keep = [x for x in self.datamodule.class_to_idx if x not in self.class_to_skip]
    _, pd_cm, accuracy, res, _ = self.trainer.test(
        test_dataloader=test_dataloader,
        idx_to_class=self.idx_to_class,
        predict_proba=True,
        class_to_keep=class_to_keep,
    )

    # save results
    self.metadata["test_confusion_matrix"] = pd_cm
    self.metadata["test_accuracy"] = accuracy
    self.metadata["test_results"] = res
    self.metadata["test_labels"] = [
        self.idx_to_class[i] if i != -1 else "N/A" for i in res["real_label"].unique().tolist()
    ]

PlaceholderTask

Bases: Task

Placeholder task.

execute()

Execute the task and all the steps.

Source code in quadra/tasks/base.py

def execute(self) -> None:
    """Execute the task and all the steps."""
    log.info("Running Placeholder Task.")
    log.info("Quadra Version: %s", str(get_version()))
    log.info("If you are reading this, it means that library is installed correctly!")

SSL(config, run_test=False, report=False, checkpoint_path=None)

Bases: LightningTask

SSL Task.

Parameters:

• config (DictConfig) –

  The experiment configuration

• checkpoint_path (str | None, default: None ) –

  The path to the checkpoint to load the model from Defaults to None

• report (bool, default: False ) –

  Whether to create the report

• run_test (bool, default: False ) –

  Whether to run final test

Source code in quadra/tasks/ssl.py

def __init__(
    self,
    config: DictConfig,
    run_test: bool = False,
    report: bool = False,
    checkpoint_path: str | None = None,
):
    super().__init__(
        config=config,
        checkpoint_path=checkpoint_path,
        run_test=run_test,
        report=report,
    )
    self._backbone: nn.Module
    self._optimizer: torch.optim.Optimizer
    self._lr_scheduler: torch.optim.lr_scheduler._LRScheduler
    self.export_folder = "deployment_model"

optimizer: torch.optim.Optimizer property writable

Get the optimizer.

scheduler: torch.optim.lr_scheduler._LRScheduler property writable

Get the scheduler.

export()

Deploy a model ready for production.

Source code in quadra/tasks/ssl.py

def export(self) -> None:
    """Deploy a model ready for production."""
    half_precision = "16" in self.trainer.precision

    input_shapes = self.config.export.input_shapes

    model_json, export_paths = export_model(
        config=self.config,
        model=self.module.model,
        export_folder=self.export_folder,
        half_precision=half_precision,
        input_shapes=input_shapes,
        idx_to_class=None,
    )

    if len(export_paths) == 0:
        return

    with open(os.path.join(self.export_folder, "model.json"), "w") as f:
        json.dump(model_json, f, cls=utils.HydraEncoder)

learnable_parameters()

Get the learnable parameters.

Source code in quadra/tasks/ssl.py

def learnable_parameters(self) -> list[nn.Parameter]:
    """Get the learnable parameters."""
    raise NotImplementedError("This method must be implemented by the subclass")

test()

Test the model.

Source code in quadra/tasks/ssl.py

def test(self) -> None:
    """Test the model."""
    if self.run_test and not self.config.trainer.get("fast_dev_run"):
        log.info("Starting testing!")
        log.info("Using last epoch's weights for testing.")
        self.trainer.test(datamodule=self.datamodule, model=self.module, ckpt_path=None)

Segmentation(config, num_viz_samples=5, checkpoint_path=None, run_test=False, evaluate=None, report=False)

Bases: Generic[SegmentationDataModuleT], LightningTask[SegmentationDataModuleT]

Task for segmentation.

Parameters:

• config (DictConfig) –

  Config object

• num_viz_samples (int, default: 5 ) –

  Number of samples to visualize. Defaults to 5.

• checkpoint_path (str | None, default: None ) –

  Path to the checkpoint to load the model from. Defaults to None.

• run_test (bool, default: False ) –

  If True, run test after training. Defaults to False.

• evaluate (DictConfig | None, default: None ) –

  Dict with evaluation parameters. Defaults to None.

• report (bool, default: False ) –

  If True, create report after training. Defaults to False.

Source code in quadra/tasks/segmentation.py

def __init__(
    self,
    config: DictConfig,
    num_viz_samples: int = 5,
    checkpoint_path: str | None = None,
    run_test: bool = False,
    evaluate: DictConfig | None = None,
    report: bool = False,
):
    super().__init__(
        config=config,
        checkpoint_path=checkpoint_path,
        run_test=run_test,
        report=report,
    )
    self.evaluate = evaluate
    self.num_viz_samples = num_viz_samples
    self.export_folder: str = "deployment_model"
    self.exported_model_path: str | None = None
    if self.evaluate and any(self.evaluate.values()):
        if (
            self.config.export is None
            or len(self.config.export.types) == 0
            or "torchscript" not in self.config.export.types
        ):
            log.info(
                "Evaluation is enabled, but training does not export a deployment model. Automatically export the "
                "model as torchscript."
            )
            if self.config.export is None:
                self.config.export = DictConfig({"types": ["torchscript"]})
            else:
                self.config.export.types.append("torchscript")

        if not self.report:
            log.info("Evaluation is enabled, but reporting is disabled. Enabling reporting automatically.")
            self.report = True

module: SegmentationModel property writable

Get the module.

export()

Generate a deployment model for the task.

Source code in quadra/tasks/segmentation.py

def export(self) -> None:
    """Generate a deployment model for the task."""
    log.info("Exporting model ready for deployment")

    # Get best model!
    if (
        self.trainer.checkpoint_callback is not None
        and hasattr(self.trainer.checkpoint_callback, "best_model_path")
        and self.trainer.checkpoint_callback.best_model_path is not None
        and len(self.trainer.checkpoint_callback.best_model_path) > 0
    ):
        best_model_path = self.trainer.checkpoint_callback.best_model_path
        log.info("Loaded best model from %s", best_model_path)

        module = self.module.__class__.load_from_checkpoint(
            best_model_path,
            model=self.module.model,
            loss_fun=None,
            optimizer=self.module.optimizer,
            lr_scheduler=self.module.schedulers,
        )
    else:
        log.warning("No checkpoint callback found in the trainer, exporting the last model weights")
        module = self.module

    if "idx_to_class" not in self.config.datamodule:
        log.info("No idx_to_class key")
        idx_to_class = {0: "good", 1: "bad"}  # TODO: Why is this the default value?
    else:
        log.info("idx_to_class is present")
        idx_to_class = self.config.datamodule.idx_to_class

    if self.config.export is None:
        raise ValueError(
            "No export type specified. This should not happen, please check if you have set "
            "the export_type or assign it to a default value."
        )

    half_precision = "16" in self.trainer.precision

    input_shapes = self.config.export.input_shapes

    model_json, export_paths = export_model(
        config=self.config,
        model=module.model,
        export_folder=self.export_folder,
        half_precision=half_precision,
        input_shapes=input_shapes,
        idx_to_class=idx_to_class,
    )

    if len(export_paths) == 0:
        return

    # Pick one model for evaluation, it should be independent of the export type as the model is wrapped
    self.exported_model_path = next(iter(export_paths.values()))

    with open(os.path.join(self.export_folder, "model.json"), "w") as f:
        json.dump(model_json, f, cls=utils.HydraEncoder)

generate_report()

Generate a report for the task.

Source code in quadra/tasks/segmentation.py

def generate_report(self) -> None:
    """Generate a report for the task."""
    if self.evaluate is not None:
        log.info("Generating evaluation report!")
        eval_tasks: list[SegmentationEvaluation] = []
        if self.evaluate.analysis:
            if self.exported_model_path is None:
                raise ValueError(
                    "Exported model path is not set yet but the task tries to do an analysis evaluation"
                )
            eval_task = SegmentationAnalysisEvaluation(
                config=self.config,
                model_path=self.exported_model_path,
            )
            eval_tasks.append(eval_task)
        for task in eval_tasks:
            task.execute()

        if len(self.logger) > 0:
            mflow_logger = get_mlflow_logger(trainer=self.trainer)
            tensorboard_logger = utils.get_tensorboard_logger(trainer=self.trainer)

            if mflow_logger is not None and self.config.core.get("upload_artifacts"):
                for task in eval_tasks:
                    for file in task.metadata["report_files"]:
                        mflow_logger.experiment.log_artifact(
                            run_id=mflow_logger.run_id, local_path=file, artifact_path=task.report_path
                        )

            if tensorboard_logger is not None and self.config.core.get("upload_artifacts"):
                for task in eval_tasks:
                    for file in task.metadata["report_files"]:
                        ext = os.path.splitext(file)[1].lower()

                        if ext in [".png", ".jpg", ".jpeg", ".bmp", ".tiff", ".tif", ".gif"]:
                            try:
                                img = cv2.imread(file)
                                img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
                            except cv2.error:
                                log.info("Could not upload artifact image %s", file)
                                continue

                            tensorboard_logger.experiment.add_image(
                                os.path.basename(file), img, 0, dataformats="HWC"
                            )
                        else:
                            utils.upload_file_tensorboard(file, tensorboard_logger)

prepare()

Prepare the task.

Source code in quadra/tasks/segmentation.py

def prepare(self) -> None:
    """Prepare the task."""
    super().prepare()
    self.module = self.config.model

SegmentationAnalysisEvaluation(config, model_path, device=None)

Bases: SegmentationEvaluation

Segmentation Analysis Evaluation Task Args: config: The experiment configuration model_path: The model path. device: Device to use for evaluation. If None, the device is automatically determined.

Source code in quadra/tasks/segmentation.py

def __init__(
    self,
    config: DictConfig,
    model_path: str,
    device: str | None = None,
):
    super().__init__(config=config, model_path=model_path, device=device)
    self.test_output: dict[str, Any] = {}

generate_report()

Generate a report.

Source code in quadra/tasks/segmentation.py

def generate_report(self) -> None:
    """Generate a report."""
    log.info("Generating analysis report")

    for stage, output in self.test_output.items():
        image_mean = OmegaConf.to_container(self.config.transforms.mean)
        if not isinstance(image_mean, list) or any(not isinstance(x, (int, float)) for x in image_mean):
            raise ValueError("Image mean is not a list of float or integer values, please check your config")
        image_std = OmegaConf.to_container(self.config.transforms.std)
        if not isinstance(image_std, list) or any(not isinstance(x, (int, float)) for x in image_std):
            raise ValueError("Image std is not a list of float or integer values, please check your config")
        reports = create_mask_report(
            stage=stage,
            output=output,
            report_path="analysis_report",
            mean=image_mean,
            std=image_std,
            analysis=True,
            nb_samples=10,
            apply_sigmoid=True,
            show_orj_predictions=True,
        )
        self.metadata["report_files"].extend(reports)
        log.info("%s analysis report completed.", stage)

prepare()

Prepare the evaluation task.

Source code in quadra/tasks/segmentation.py

def prepare(self) -> None:
    """Prepare the evaluation task."""
    super().prepare()
    self.datamodule.setup(stage="fit")
    self.datamodule.setup(stage="test")

test()

Run testing.

Source code in quadra/tasks/segmentation.py

@automatic_datamodule_batch_size(batch_size_attribute_name="batch_size")
def test(self) -> None:
    """Run testing."""
    log.info("Starting inference for analysis.")

    stages: list[str] = []
    dataloaders: list[torch.utils.data.DataLoader] = []

    # if self.datamodule.train_dataset_available:
    #     stages.append("train")
    #     dataloaders.append(self.datamodule.train_dataloader())
    #     if self.datamodule.val_dataset_available:
    #         stages.append("val")
    #         dataloaders.append(self.datamodule.val_dataloader())

    if self.datamodule.test_dataset_available:
        stages.append("test")
        dataloaders.append(self.datamodule.test_dataloader())
    for stage, dataloader in zip(stages, dataloaders):
        log.info("Running inference on %s set with batch size: %d", stage, dataloader.batch_size)
        image_list, mask_list, mask_pred_list, label_list = [], [], [], []
        for batch in dataloader:
            images, masks, labels = batch
            images = images.to(device=self.device, dtype=self.deployment_model.model_dtype)
            if len(masks.shape) == 3:  # BxHxW -> Bx1xHxW
                masks = masks.unsqueeze(1)
            with torch.no_grad():
                image_list.append(images)
                mask_list.append(masks)
                mask_pred_list.append(self.deployment_model(images.to(self.device)))
                label_list.append(labels)

        output = {
            "image": torch.cat(image_list, dim=0),
            "mask": torch.cat(mask_list, dim=0),
            "label": torch.cat(label_list, dim=0),
            "mask_pred": torch.cat(mask_pred_list, dim=0),
        }
        self.test_output[stage] = output

train()

Skip training.

Source code in quadra/tasks/segmentation.py

def train(self) -> None:
    """Skip training."""

SegmentationEvaluation(config, model_path, device='cpu')

Bases: Evaluation[SegmentationDataModuleT]

Segmentation Evaluation Task with deployment models.

Parameters:

• config (DictConfig) –

  The experiment configuration

• model_path (str) –

  The experiment path.

• device (str | None, default: 'cpu' ) –

  Device to use for evaluation. If None, the device is automatically determined.

Raises:

• ValueError –

  If the model path is not provided

Source code in quadra/tasks/segmentation.py

def __init__(
    self,
    config: DictConfig,
    model_path: str,
    device: str | None = "cpu",
):
    super().__init__(config=config, model_path=model_path, device=device)
    self.config = config

inference(dataloader, deployment_model, device)

Run inference on the dataloader and return the output.

Parameters:

• dataloader (DataLoader) –

  The dataloader to run inference on

• deployment_model (BaseEvaluationModel) –

  The deployment model to use

• device (device) –

  The device to run inference on

Source code in quadra/tasks/segmentation.py

@torch.no_grad()
def inference(
    self, dataloader: DataLoader, deployment_model: BaseEvaluationModel, device: torch.device
) -> dict[str, torch.Tensor]:
    """Run inference on the dataloader and return the output.

    Args:
        dataloader: The dataloader to run inference on
        deployment_model: The deployment model to use
        device: The device to run inference on
    """
    image_list, mask_list, mask_pred_list, label_list = [], [], [], []
    for batch in dataloader:
        images, masks, labels = batch
        images = images.to(device)
        masks = masks.to(device)
        labels = labels.to(device)
        image_list.append(images.cpu())
        mask_list.append(masks.cpu())
        mask_pred_list.append(deployment_model(images.to(device)).cpu())
        label_list.append(labels.cpu())
    output = {
        "image": torch.cat(image_list, dim=0),
        "mask": torch.cat(mask_list, dim=0),
        "label": torch.cat(label_list, dim=0),
        "mask_pred": torch.cat(mask_pred_list, dim=0),
    }
    return output

prepare()

Prepare the evaluation.

Source code in quadra/tasks/segmentation.py

def prepare(self) -> None:
    """Prepare the evaluation."""
    super().prepare()
    # TODO: Why we propagate mean and std only in Segmentation?
    self.config.transforms.mean = self.model_data["mean"]
    self.config.transforms.std = self.model_data["std"]
    # Setup datamodule
    if hasattr(self.config.datamodule, "idx_to_class"):
        idx_to_class = self.model_data["classes"]  # dict {index: class}
        self.config.datamodule.idx_to_class = idx_to_class
    self.datamodule = self.config.datamodule
    # prepare_data() must be explicitly called because there is no lightning training
    self.datamodule.prepare_data()

save_config()

Skip saving the config.

Source code in quadra/tasks/segmentation.py

def save_config(self) -> None:
    """Skip saving the config."""

SklearnClassification(config, output, device, automatic_batch_size, save_model_summary=False, half_precision=False, gradcam=False)

Bases: Generic[SklearnClassificationDataModuleT], Task[SklearnClassificationDataModuleT]

Sklearn classification task.

Parameters:

• config (DictConfig) –

  The experiment configuration

• device (str) –

  The device to use. Defaults to None.

• output (DictConfig) –

  Dictionary defining which kind of outputs to generate. Defaults to None.

• automatic_batch_size (DictConfig) –

  Whether to automatically find the largest batch size that fits in memory.

• save_model_summary (bool, default: False ) –

  Whether to save a model_summary.txt file containing the model summary.

• half_precision (bool, default: False ) –

  Whether to use half precision during training.

• gradcam (bool, default: False ) –

  Whether to compute gradcams for test results.

Source code in quadra/tasks/classification.py

def __init__(
    self,
    config: DictConfig,
    output: DictConfig,
    device: str,
    automatic_batch_size: DictConfig,
    save_model_summary: bool = False,
    half_precision: bool = False,
    gradcam: bool = False,
):
    super().__init__(config=config)

    self._device = device
    self.output = output
    self._backbone: ModelSignatureWrapper
    self._trainer: SklearnClassificationTrainer
    self._model: ClassifierMixin
    self.metadata: dict[str, Any] = {
        "test_confusion_matrix": [],
        "test_accuracy": [],
        "test_results": [],
        "test_labels": [],
        "cams": [],
    }
    self.export_folder = "deployment_model"
    self.deploy_info_file = "model.json"
    self.train_dataloader_list: list[torch.utils.data.DataLoader] = []
    self.test_dataloader_list: list[torch.utils.data.DataLoader] = []
    self.automatic_batch_size = automatic_batch_size
    self.save_model_summary = save_model_summary
    self.half_precision = half_precision
    self.gradcam = gradcam

backbone: ModelSignatureWrapper property writable

model: ClassifierMixin property writable

sklearn.base.ClassifierMixin: The model.

trainer: SklearnClassificationTrainer property writable

execute()

Execute the experiment and all the steps.

Source code in quadra/tasks/classification.py

def execute(self) -> None:
    """Execute the experiment and all the steps."""
    self.prepare()
    self.train()
    if self.output.report:
        self.generate_report()
    self.train_full_data()
    if self.config.export is not None and len(self.config.export.types) > 0:
        self.export()
    if self.output.test_full_data:
        self.test_full_data()
    self.finalize()

export()

Generate deployment model for the task.

Source code in quadra/tasks/classification.py

def export(self) -> None:
    """Generate deployment model for the task."""
    if self.config.export is None or len(self.config.export.types) == 0:
        log.info("No export type specified skipping export")
        return

    input_shapes = self.config.export.input_shapes

    idx_to_class = {v: k for k, v in self.datamodule.full_dataset.class_to_idx.items()}

    model_json, export_paths = export_model(
        config=self.config,
        model=self.backbone,
        export_folder=self.export_folder,
        half_precision=self.half_precision,
        input_shapes=input_shapes,
        idx_to_class=idx_to_class,
        pytorch_model_type="backbone",
    )

    dump(self.model, os.path.join(self.export_folder, "classifier.joblib"))

    if len(export_paths) > 0:
        with open(os.path.join(self.export_folder, self.deploy_info_file), "w") as f:
            json.dump(model_json, f)

extract_model_summary(feature_extractor, dl)

Given a dataloader and a PyTorch model, use torchinfo to extract a summary of the model and save it to a file.

Parameters:

• dl (DataLoader) –

  PyTorch dataloader

• feature_extractor (Module | BaseEvaluationModel) –

  PyTorch backbone

Source code in quadra/tasks/classification.py

def extract_model_summary(
    self, feature_extractor: torch.nn.Module | BaseEvaluationModel, dl: torch.utils.data.DataLoader
) -> None:
    """Given a dataloader and a PyTorch model, use torchinfo to extract a summary of the model and save it
    to a file.

    Args:
        dl: PyTorch dataloader
        feature_extractor: PyTorch backbone
    """
    if isinstance(feature_extractor, (TorchEvaluationModel, TorchscriptEvaluationModel)):
        # TODO: I'm not sure torchinfo supports torchscript models
        # If we are working with torch based evaluation models we need to extract the model
        feature_extractor = feature_extractor.model

    for b in tqdm(dl):
        x1, _ = b

        if hasattr(feature_extractor, "parameters"):
            # Move input to the correct device
            parameter = next(feature_extractor.parameters())
            x1 = x1.to(parameter.device).to(parameter.dtype)
            x1 = x1[0].unsqueeze(0)  # Remove batch dimension

            model_info = None

            try:
                try:
                    # TODO: Do we want to print the summary to the console as well?
                    model_info = summary(feature_extractor, input_data=(x1), verbose=0)  # type: ignore[arg-type]
                except Exception:
                    log.warning(
                        "Failed to retrieve model summary using input data information, retrieving only "
                        "parameters information"
                    )
                    model_info = summary(feature_extractor, verbose=0)  # type: ignore[arg-type]
            except Exception as e:
                # If for some reason the summary fails we don't want to stop the training
                log.warning("Failed to retrieve model summary: %s", e)

            if model_info is not None:
                with open("model_summary.txt", "w") as f:
                    f.write(str(model_info))
        else:
            log.warning("Failed to retrieve model summary, current model has no parameters")

        break

generate_report()

Generate report for the task.

Source code in quadra/tasks/classification.py

def generate_report(self) -> None:
    """Generate report for the task."""
    log.info("Generating report!")

    cm_list = []

    for count in range(len(self.metadata["test_accuracy"])):
        current_output_folder = f"{self.output.folder}_{count}"
        os.makedirs(current_output_folder, exist_ok=True)

        c_matrix = self.metadata["test_confusion_matrix"][count]
        cm_list.append(c_matrix)
        save_classification_result(
            results=self.metadata["test_results"][count],
            output_folder=current_output_folder,
            confmat=c_matrix,
            accuracy=self.metadata["test_accuracy"][count],
            test_dataloader=self.test_dataloader_list[count],
            config=self.config,
            output=self.output,
            grayscale_cams=self.metadata["cams"][count],
        )
    final_confusion_matrix = sum(cm_list)

    self.metadata["final_confusion_matrix"] = final_confusion_matrix
    # Save final conf matrix
    final_folder = f"{self.output.folder}"
    os.makedirs(final_folder, exist_ok=True)
    disp = ConfusionMatrixDisplay(
        confusion_matrix=np.array(final_confusion_matrix),
        display_labels=[x.replace("pred:", "") for x in final_confusion_matrix.columns.to_list()],
    )
    disp.plot(include_values=True, cmap=plt.cm.Greens, ax=None, colorbar=False, xticks_rotation=90)
    plt.title(f"Confusion Matrix (Accuracy: {(self.metadata['test_accuracy'][count] * 100):.2f}%)")
    plt.savefig(os.path.join(final_folder, "test_confusion_matrix.png"), bbox_inches="tight", pad_inches=0, dpi=300)
    plt.close()

prepare()

Prepare the experiment.

Source code in quadra/tasks/classification.py

def prepare(self) -> None:
    """Prepare the experiment."""
    self.datamodule = self.config.datamodule

    self.backbone = self.config.backbone

    self.model = self.config.model

    # prepare_data() must be explicitly called if the task does not include a lightining training
    self.datamodule.prepare_data()
    self.datamodule.setup(stage="fit")

    self.trainer = self.config.trainer

test()

Skip test phase.

Source code in quadra/tasks/classification.py

def test(self) -> None:
    """Skip test phase."""

test_full_data()

Test model trained on full dataset.

Source code in quadra/tasks/classification.py

@typing.no_type_check
@automatic_datamodule_batch_size(batch_size_attribute_name="batch_size")
def test_full_data(self) -> None:
    """Test model trained on full dataset."""
    self.config.datamodule.class_to_idx = self.datamodule.full_dataset.class_to_idx
    self.config.datamodule.phase = "test"
    idx_to_class = self.datamodule.full_dataset.idx_to_class
    self.datamodule.setup("test")
    test_dataloader = self.datamodule.test_dataloader()

    if len(self.datamodule.data["samples"]) == 0:
        log.info("No test data, skipping test")
        return

    # Put backbone on the correct device as it may be moved after export
    self.backbone.to(self.device)
    _, pd_cm, accuracy, res, cams = self.trainer.test(
        test_dataloader=test_dataloader, idx_to_class=idx_to_class, predict_proba=True, gradcam=self.gradcam
    )

    output_folder_test = "test"

    os.makedirs(output_folder_test, exist_ok=True)

    save_classification_result(
        results=res,
        output_folder=output_folder_test,
        confmat=pd_cm,
        accuracy=accuracy,
        test_dataloader=test_dataloader,
        config=self.config,
        output=self.output,
        grayscale_cams=cams,
    )

train()

Train the model.

Source code in quadra/tasks/classification.py

@typing.no_type_check
@automatic_datamodule_batch_size(batch_size_attribute_name="batch_size")
def train(self) -> None:
    """Train the model."""
    log.info("Starting training...!")
    all_features = None
    all_labels = None

    class_to_keep = None

    self.train_dataloader_list = list(self.datamodule.train_dataloader())
    self.test_dataloader_list = list(self.datamodule.val_dataloader())

    if hasattr(self.datamodule, "class_to_keep_training") and self.datamodule.class_to_keep_training is not None:
        class_to_keep = self.datamodule.class_to_keep_training

    if self.save_model_summary:
        self.extract_model_summary(feature_extractor=self.backbone, dl=self.datamodule.full_dataloader())

    if hasattr(self.datamodule, "cache") and self.datamodule.cache:
        if self.config.trainer.iteration_over_training != 1:
            raise AttributeError("Cache is only supported when iteration over training is set to 1")

        full_dataloader = self.datamodule.full_dataloader()
        all_features, all_labels, _ = get_feature(
            feature_extractor=self.backbone, dl=full_dataloader, iteration_over_training=1
        )

        sorted_indices = np.argsort(full_dataloader.dataset.x)
        all_features = all_features[sorted_indices]
        all_labels = all_labels[sorted_indices]

    # cycle over all train/test split
    for train_dataloader, test_dataloader in zip(self.train_dataloader_list, self.test_dataloader_list):
        # Reinit classifier
        self.model = self.config.model
        self.trainer.change_classifier(self.model)

        # Train on current training set
        if all_features is not None and all_labels is not None:
            # Find which are the indices used to pass from the sorted list of string to the disordered one
            sorted_indices = np.argsort(np.concatenate([train_dataloader.dataset.x, test_dataloader.dataset.x]))
            revese_sorted_indices = np.argsort(sorted_indices)

            # Use these indices to correctly match the extracted features with the new file order
            all_features_sorted = all_features[revese_sorted_indices]
            all_labels_sorted = all_labels[revese_sorted_indices]

            train_len = len(train_dataloader.dataset.x)

            self.trainer.fit(
                train_features=all_features_sorted[0:train_len], train_labels=all_labels_sorted[0:train_len]
            )

            _, pd_cm, accuracy, res, cams = self.trainer.test(
                test_dataloader=test_dataloader,
                test_features=all_features_sorted[train_len:],
                test_labels=all_labels_sorted[train_len:],
                class_to_keep=class_to_keep,
                idx_to_class=train_dataloader.dataset.idx_to_class,
                predict_proba=True,
                gradcam=self.gradcam,
            )
        else:
            self.trainer.fit(train_dataloader=train_dataloader)
            _, pd_cm, accuracy, res, cams = self.trainer.test(
                test_dataloader=test_dataloader,
                class_to_keep=class_to_keep,
                idx_to_class=train_dataloader.dataset.idx_to_class,
                predict_proba=True,
                gradcam=self.gradcam,
            )

        # save results
        self.metadata["test_confusion_matrix"].append(pd_cm)
        self.metadata["test_accuracy"].append(accuracy)
        self.metadata["test_results"].append(res)
        self.metadata["test_labels"].append(
            [
                train_dataloader.dataset.idx_to_class[i] if i != -1 else "N/A"
                for i in res["real_label"].unique().tolist()
            ]
        )
        self.metadata["cams"].append(cams)

train_full_data()

Train the model on train + validation.

Source code in quadra/tasks/classification.py

@automatic_datamodule_batch_size(batch_size_attribute_name="batch_size")
def train_full_data(self):
    """Train the model on train + validation."""
    # Reinit classifier
    self.model = self.config.model
    self.trainer.change_classifier(self.model)

    self.trainer.fit(train_dataloader=self.datamodule.full_dataloader())

SklearnTestClassification(config, output, model_path, device, gradcam=False, **kwargs)

Bases: Evaluation[SklearnClassificationDataModuleT]

Perform a test using an imported SklearnClassification pytorch model.

Parameters:

• config (DictConfig) –

  The experiment configuration

• output (DictConfig) –

  where to save results

• model_path (str) –

  path to trained model generated from SklearnClassification task.

• device (str) –

  the device where to run the model (cuda or cpu)

• gradcam (bool, default: False ) –

  Whether to compute gradcams

• **kwargs (Any, default: {} ) –

  Additional arguments to pass to the task

Source code in quadra/tasks/classification.py

def __init__(
    self,
    config: DictConfig,
    output: DictConfig,
    model_path: str,
    device: str,
    gradcam: bool = False,
    **kwargs: Any,
):
    super().__init__(config=config, model_path=model_path, device=device, **kwargs)
    self.gradcam = gradcam
    self.output = output
    self._backbone: BaseEvaluationModel
    self._classifier: ClassifierMixin
    self.class_to_idx: dict[str, int]
    self.idx_to_class: dict[int, str]
    self.test_dataloader: torch.utils.data.DataLoader
    self.metadata: dict[str, Any] = {
        "test_confusion_matrix": None,
        "test_accuracy": None,
        "test_results": None,
        "test_labels": None,
        "cams": None,
    }

backbone: BaseEvaluationModel property writable

classifier: ClassifierMixin property writable

deployment_model property writable

Deployment model.

trainer: SklearnClassificationTrainer property writable

execute()

Execute the experiment and all the steps.

Source code in quadra/tasks/classification.py

def execute(self) -> None:
    """Execute the experiment and all the steps."""
    self.prepare()
    self.test()
    if self.output.report:
        self.generate_report()
    self.finalize()

generate_report()

Generate a report for the task.

Source code in quadra/tasks/classification.py

def generate_report(self) -> None:
    """Generate a report for the task."""
    log.info("Generating report!")
    os.makedirs(self.output.folder, exist_ok=True)
    save_classification_result(
        results=self.metadata["test_results"],
        output_folder=self.output.folder,
        confmat=self.metadata["test_confusion_matrix"],
        accuracy=self.metadata["test_accuracy"],
        test_dataloader=self.test_dataloader,
        config=self.config,
        output=self.output,
        grayscale_cams=self.metadata["cams"],
    )

prepare()

Prepare the experiment.

Source code in quadra/tasks/classification.py

def prepare(self) -> None:
    """Prepare the experiment."""
    super().prepare()

    idx_to_class = {}
    class_to_idx = {}
    for k, v in self.model_data["classes"].items():
        idx_to_class[int(k)] = v
        class_to_idx[v] = int(k)

    self.idx_to_class = idx_to_class
    self.class_to_idx = class_to_idx

    self.config.datamodule.class_to_idx = class_to_idx

    self.datamodule = self.config.datamodule
    # prepare_data() must be explicitly called because there is no lightning training
    self.datamodule.prepare_data()
    self.datamodule.setup(stage="test")

    # Configure trainer
    self.trainer = self.config.trainer

test()

Run the test.

Source code in quadra/tasks/classification.py

@automatic_datamodule_batch_size(batch_size_attribute_name="batch_size")
def test(self) -> None:
    """Run the test."""
    self.test_dataloader = self.datamodule.test_dataloader()

    _, pd_cm, accuracy, res, cams = self.trainer.test(
        test_dataloader=self.test_dataloader,
        idx_to_class=self.idx_to_class,
        predict_proba=True,
        gradcam=self.gradcam,
    )

    # save results
    self.metadata["test_confusion_matrix"] = pd_cm
    self.metadata["test_accuracy"] = accuracy
    self.metadata["test_results"] = res
    self.metadata["test_labels"] = [
        self.idx_to_class[i] if i != -1 else "N/A" for i in res["real_label"].unique().tolist()
    ]
    self.metadata["cams"] = cams

Task(config)

Bases: Generic[DataModuleT]

Base Experiment Task.

Parameters:

• config (DictConfig) –

  The experiment configuration.

Source code in quadra/tasks/base.py

def __init__(self, config: DictConfig):
    self.config = config
    self.export_folder: str = "deployment_model"
    self._datamodule: DataModuleT
    self.metadata: dict[str, Any]
    self.save_config()

datamodule: DataModuleT property writable

execute()

Execute the experiment and all the steps.

Source code in quadra/tasks/base.py

def execute(self) -> None:
    """Execute the experiment and all the steps."""
    self.prepare()
    self.train()
    self.test()
    if self.config.export is not None and len(self.config.export.types) > 0:
        self.export()
    self.generate_report()
    self.finalize()

export()

Export model for production.

Source code in quadra/tasks/base.py

def export(self) -> None:
    """Export model for production."""
    log.info("Export model for production not implemented for this task!")

finalize()

Finalize the experiment.

Source code in quadra/tasks/base.py

def finalize(self) -> None:
    """Finalize the experiment."""
    log.info("Results are saved in %s", os.getcwd())

generate_report()

Generate a report.

Source code in quadra/tasks/base.py

def generate_report(self) -> None:
    """Generate a report."""
    log.info("Report generation not implemented for this task!")

prepare()

Prepare the experiment.

Source code in quadra/tasks/base.py

def prepare(self) -> None:
    """Prepare the experiment."""
    self.datamodule = self.config.datamodule

save_config()

Save the experiment configuration when running an Hydra experiment.

Source code in quadra/tasks/base.py

def save_config(self) -> None:
    """Save the experiment configuration when running an Hydra experiment."""
    if HydraConfig.initialized():
        with open("config_resolved.yaml", "w") as fp:
            OmegaConf.save(config=OmegaConf.to_container(self.config, resolve=True), f=fp.name)

test()

Test the model.

Source code in quadra/tasks/base.py

def test(self) -> Any:
    """Test the model."""
    log.info("Testing not implemented for this task!")

train()

Train the model.

Source code in quadra/tasks/base.py

def train(self) -> Any:
    """Train the model."""
    log.info("Training not implemented for this task!")