tasks

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

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 (Optional[str]) –

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

• run_test (bool) –

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

• report (bool) –

  Whether to report the results. Defaults to False.

• export_config (Optional[DictConfig]) –

  Dictionary containing the export configuration, it should contain the following keys:

  • types: List of types to export.
  • input_shapes: Optional list of input shapes to use, they must be in the same order of the forward arguments.

Source code in quadra/tasks/anomaly.py

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

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.export_config is None or len(self.export_config.types) == 0:
        log.info("No export type specified skipping export")
        return

    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.export_config.input_shapes

    half_precision = int(self.trainer.precision) == 16

    for export_type in self.export_config.types:
        if export_type == "torchscript":
            out = export_torchscript_model(
                model=model,
                input_shapes=input_shapes,
                output_path=self.export_folder,
                half_precision=half_precision,
            )

            if out is None:
                log.warning("Skipping torchscript export since the model is not supported")
                continue

            _, input_shapes = out

    model_json = {
        "input_size": input_shapes,
        "classes": {0: "good", 1: "defect"},
        "mean": list(self.config.transforms.mean),
        "std": list(self.config.transforms.std),
        "image_threshold": np.round(self.module.image_threshold.value.item(), 3),
        "pixel_threshold": np.round(self.module.pixel_threshold.value.item(), 3),
    }

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

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)

Classification(config, output, checkpoint_path=None, lr_multiplier=None, export_config=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.

• export_config (Optional[DictConfig]) –

  Dictionary containing the export configuration, it should contain the following keys:

  • types: List of types to export.
  • input_shapes: Optional list of input shapes to use, they must be in the same order of the forward arguments.
• gradcam (bool) –

  Whether to compute gradcams

• checkpoint_path (Optional[str]) –

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

• lr_multiplier (Optional[float]) –

  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) –

  Whether to generate a report containing the results after test phase

• run_test (bool) –

  Whether to run the test phase.

Source code in quadra/tasks/classification.py

def __init__(
    self,
    config: DictConfig,
    output: DictConfig,
    checkpoint_path: Optional[str] = None,
    lr_multiplier: Optional[float] = None,
    export_config: Optional[DictConfig] = 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,
        export_config=export_config,
    )
    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: Optional[Dict[str, Any]] = None
    self.export_folder: str = "deployment_model"
    self.deploy_info_file: str = "model.json"
    self.report_confmat: pd.DataFrame

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.export_config is None or len(self.export_config.types) == 0:
        log.info("No export type specified skipping export")
        return

    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()}

    if self.trainer.checkpoint_callback is None:
        raise ValueError("No checkpoint callback found in the trainer")

    best_model_path = self.trainer.checkpoint_callback.best_model_path  # type: ignore[attr-defined]
    log.info("Saving deployment model for %s checkpoint", best_model_path)

    module = self.module.load_from_checkpoint(
        best_model_path,
        model=self.module.model,
        optimizer=self.optimizer,
        lr_scheduler=self.scheduler,
        criterion=self.module.criterion,
        gradcam=False,
    )

    input_shapes = self.export_config.input_shapes

    # TODO: This breaks with bf16 precision!!!
    half_precision = int(self.trainer.precision) == 16

    for export_type in self.export_config.types:
        if export_type == "torchscript":
            out = export_torchscript_model(
                model=module.model,
                input_shapes=input_shapes,
                output_path=self.export_folder,
                half_precision=half_precision,
            )

            if out is None:
                log.warning("Skipping torchscript export since the model is not supported")
                continue

            _, input_shapes = out
        elif export_type == "pytorch":
            export_pytorch_model(
                model=module.model,
                output_path=self.export_folder,
            )
            with open(os.path.join(self.export_folder, "model_config.yaml"), "w") as f:
                OmegaConf.save(self.config.model, f, resolve=True)
        else:
            log.warning("Export type: %s not implemented", export_type)

    self.model_json = {
        "input_size": input_shapes,
        "classes": idx_to_class,
        "mean": list(self.config.transforms.mean),
        "std": list(self.config.transforms.std),
    }

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

freeze_layers(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(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)

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")
    best_model_path = self.trainer.checkpoint_callback.best_model_path  # type: ignore[union-attr]
    predictions_outputs = self.trainer.predict(
        model=self.module, datamodule=self.datamodule, ckpt_path=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]
    if not all_outputs:
        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_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)
    res = pd.DataFrame(
        {
            "sample": list(test_dataset.x),
            "real_label": all_targets,
            "pred_label": all_outputs,
        }
    )
    os.makedirs(output_folder_test, exist_ok=True)
    save_classification_result(
        results=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):
    """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.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"):
        if self.trainer.checkpoint_callback is None:
            raise ValueError("Checkpoint callback is not defined!")
        log.info("Starting testing!")
        self.datamodule.setup(stage="test")
        log.info("Using best epoch's weights for testing.")
        self.trainer.test(datamodule=self.datamodule, model=self.module, ckpt_path="best")

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) –

  Whether to generate the report of the predictions

• gradcam (bool) –

  Whether to compute gradcams

• device (Optional[str]) –

  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: Optional[str] = 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 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)

    test_dataset = cast(ImageClassificationListDataset, self.datamodule.test_dataloader().dataset)
    res = pd.DataFrame(
        {
            "sample": list(test_dataset.x),
            "real_label": self.metadata["test_labels"],
            "pred_label": self.metadata["test_results"],
        }
    )
    os.makedirs(self.report_path, exist_ok=True)
    save_classification_result(
        results=res,
        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"],
    )

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()}

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 isinstance(self.deployment_model.features_extractor, timm.models.resnet.ResNet):
        target_layers = [
            cast(BaseNetworkBuilder, self.deployment_model).features_extractor.layer4[-1]  # type: ignore[index]
        ]
        self.cam = GradCAM(
            model=self.deployment_model,
            target_layers=target_layers,
            use_cuda=(self.device != "cpu"),
        )
        for p in self.deployment_model.features_extractor.layer4[-1].parameters():
            p.requires_grad = True
    elif is_vision_transformer(cast(BaseNetworkBuilder, self.deployment_model).features_extractor):
        self.grad_rollout = VitAttentionGradRollout(self.deployment_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

def test(self) -> None:
    """Perform test."""
    log.info("Running test")
    # prepare_data() must be explicitly called because there is no training
    self.datamodule.prepare_data()
    self.datamodule.setup(stage="test")
    test_dataloader = self.datamodule.test_dataloader()

    image_labels = []
    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(self.device).detach()
            outputs = self.deployment_model(im).detach()
            probs = torch.softmax(outputs, dim=1)
            preds = torch.max(probs, dim=1).indices

            predicted_classes.append(preds.tolist())
            image_labels.extend(target.tolist())
            if self.gradcam:
                with torch.inference_mode(False):
                    im = im.clone()
                    if isinstance(self.deployment_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).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: Optional[torch.Tensor] = 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]
    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,
    )
    log.info("Avg classification accuracy: %s", test_accuracy)

    # save results
    self.metadata["test_confusion_matrix"] = pd_cm
    self.metadata["test_accuracy"] = test_accuracy
    self.metadata["test_results"] = predicted_classes
    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 (Optional[str]) –

  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: Optional[str] = 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: Union[RecursiveScriptModule, Module]
    self.deployment_model_type: str
    self.model_info_filename = "model.json"
    self.report_path = ""
    self.metadata = {"report_files": []}

deployment_model: Union[RecursiveScriptModule, nn.Module] 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, export_config=None)

Bases: Generic[DataModuleT], Task[DataModuleT]

Base Experiment Task.

Parameters:

• config (DictConfig) –

  The experiment configuration

• checkpoint_path (Optional[str]) –

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

• run_test (bool) –

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

• report (bool) –

  Whether to generate a report. Defaults to False.

• export_config (Optional[DictConfig]) –

  Dictionary containing the export configuration, it should contain the following keys:

  • types: List of types to export.
  • input_shapes: Optional list of input shapes to use, they must be in the same order of the forward arguments.

Source code in quadra/tasks/base.py

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

callbacks: List[Callback] property writable

List[Callback]: The callbacks.

devices: Union[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.export_config is not None and len(self.export_config.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"):
        if 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
    ):
        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, export_config=None)

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.

• export_config (Optional[DictConfig]) –

  Dictionary containing the export configuration, it should contain the following keys:

  • types: List of types to export.
  • input_shapes: Optional list of input shapes to use, they must be in the same order of the forward arguments.

Source code in quadra/tasks/patch.py

def __init__(
    self,
    config: DictConfig,
    output: DictConfig,
    device: str,
    export_config: Optional[DictConfig] = None,
):
    super().__init__(config=config, export_config=export_config)
    self.device: str = device
    self.output: DictConfig = output
    self.return_polygon: bool = True
    self.reconstruction_results: Dict[str, Any]
    self._backbone: torch.nn.Module
    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"

backbone: torch.nn.Module 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.export_config is not None and len(self.export_config.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."""
    if self.export_config is None or len(self.export_config.types) == 0:
        log.info("No export type specified skipping export")
        return

    os.makedirs(self.export_folder, exist_ok=True)

    input_shapes = self.export_config.input_shapes

    for export_type in self.export_config.types:
        if export_type == "torchscript":
            out = export_torchscript_model(
                model=self.backbone,
                input_shapes=input_shapes,
                output_path=self.export_folder,
                half_precision=False,
            )

            if out is None:
                log.warning("Skipping torchscript export since the model is not supported")
                continue

            _, input_shapes = out

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

    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

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

    model_json = {
        "input_size": input_shapes,
        "classes": idx_to_class,
        "mean": self.config.transforms.mean,
        "std": self.config.transforms.std,
        "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)

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

    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.keys() 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) –

  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: torch.nn.Module
    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: torch.nn.Module 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

test()

Run the test.

Source code in quadra/tasks/patch.py

def test(self) -> None:
    """Run the test."""
    # prepare_data() must be explicitly called because there is no lightning training
    self.datamodule.prepare_data()
    self.datamodule.setup(stage="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.keys() 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, export_config=None)

Bases: LightningTask

SSL Task.

Parameters:

• config (DictConfig) –

  The experiment configuration

• checkpoint_path (Optional[str]) –

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

• report (bool) –

  Whether to create the report

• run_test (bool) –

  Whether to run final test

• export_config (Optional[DictConfig]) –

  Dictionary containing the export configuration, it should contain the following keys:

  • types: List of types to export.
  • input_shapes: Optional list of input shapes to use, they must be in the same order of the forward arguments.

Source code in quadra/tasks/ssl.py

def __init__(
    self,
    config: DictConfig,
    run_test: bool = False,
    report: bool = False,
    checkpoint_path: Optional[str] = None,
    export_config: Optional[DictConfig] = None,
):
    super().__init__(
        config=config,
        checkpoint_path=checkpoint_path,
        run_test=run_test,
        report=report,
        export_config=export_config,
    )
    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."""
    if self.export_config is None or len(self.export_config.types) == 0:
        log.info("No export type specified skipping export")
        return

    mean = self.config.transforms.mean
    std = self.config.transforms.std
    half_precision = int(self.trainer.precision) == 16

    input_shapes = self.export_config.input_shapes

    for export_type in self.export_config.types:
        if export_type == "torchscript":
            out = export_torchscript_model(
                model=cast(nn.Module, self.module.model),
                input_shapes=input_shapes,
                output_path=self.export_folder,
                half_precision=half_precision,
            )

            if out is None:
                log.warning("Skipping torchscript export since the model is not supported")
                continue

            _, input_shapes = out

    model_json = {
        "input_size": input_shapes,
        "classes": None,
        "mean": mean,
        "std": std,
    }

    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, export_config=None)

Bases: Generic[SegmentationDataModuleT], LightningTask[SegmentationDataModuleT]

Task for segmentation.

Parameters:

• config (DictConfig) –

  Config object

• num_viz_samples (int) –

  Number of samples to visualize. Defaults to 5.

• checkpoint_path (Optional[str]) –

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

• run_test (bool) –

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

• evaluate (Optional[DictConfig]) –

  Dict with evaluation parameters. Defaults to None.

• report (bool) –

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

• export_config (Optional[DictConfig]) –

  Dictionary containing the export configuration, it should contain the following keys:

  • types: List of types to export.
  • input_shapes: Optional list of input shapes to use, they must be in the same order of the forward arguments.

Source code in quadra/tasks/segmentation.py

def __init__(
    self,
    config: DictConfig,
    num_viz_samples: int = 5,
    checkpoint_path: Optional[str] = None,
    run_test: bool = False,
    evaluate: Optional[DictConfig] = None,
    report: bool = False,
    export_config: Optional[DictConfig] = None,
):
    super().__init__(
        config=config,
        checkpoint_path=checkpoint_path,
        run_test=run_test,
        report=report,
        export_config=export_config,
    )
    self.evaluate = evaluate
    self.num_viz_samples = num_viz_samples
    self.export_folder: str = "deployment_model"
    self.exported_model_path: Optional[str] = None
    if self.evaluate and any(self.evaluate.values()):
        if (
            self.export_config is None
            or len(self.export_config.types) == 0
            or "torchscript" not in self.export_config.types
        ):
            log.info(
                "Evaluation is enabled, but training does not export a deployment model. Automatically export the "
                "model as torchscript."
            )
            if self.export_config is None:
                self.export_config = DictConfig({"types": ["torchscript"]})
            else:
                self.export_config.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 None:
        raise ValueError("No checkpoint callback found in the trainer")
    best_model_path = self.trainer.checkpoint_callback.best_model_path  # type: ignore[attr-defined]
    log.info("Loaded best model from %s", best_model_path)

    module = self.module.load_from_checkpoint(
        best_model_path,
        model=self.module.model,
        loss_fun=None,
        optimizer=self.module.optimizer,
        lr_scheduler=self.module.schedulers,
    )

    if "idx_to_class" not in self.config.datamodule:
        log.info("No idx_to_class key")
        classes = {0: "good", 1: "bad"}
    else:
        log.info("idx_to_class is present")
        classes = self.config.datamodule.idx_to_class

    if self.export_config 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."
        )

    input_shapes = self.export_config.input_shapes

    half_precision = self.trainer.precision == 16

    for export_type in self.export_config.types:
        if export_type == "torchscript":
            out = export_torchscript_model(
                model=module.model,
                input_shapes=input_shapes,
                output_path=self.export_folder,
                half_precision=half_precision,
            )

            if out is None:
                log.warning("Skipping torchscript export since the model is not supported")
                continue

            self.exported_model_path, input_shapes = out

    if input_shapes is None:
        log.warning("Not able to export the model in any format")

    model_json = {
        "input_size": input_shapes,
        "classes": classes,
        "mean": self.config.transforms.mean,
        "std": self.config.transforms.std,
    }

    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='cpu')

Bases: SegmentationEvaluation

Segmentation Analysis Evaluation Task

Parameters:

• config (DictConfig) –

  The experiment configuration

• model_path (str) –

  The model path.

• device (Optional[str]) –

  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: Optional[str] = "cpu",
):
    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)

test()

Run testing.

Source code in quadra/tasks/segmentation.py

def test(self) -> None:
    """Run testing."""
    log.info("Starting testing")

    stages: List[str] = []
    dataloaders: List[torch.utils.data.DataLoader] = []
    self.datamodule.setup(stage="fit")
    self.datamodule.setup(stage="test")
    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):
        image_list, mask_list, mask_pred_list, label_list = [], [], [], []
        for batch in dataloader:
            images, masks, labels = batch
            images = images.to(self.device)
            if len(masks.shape) == 3:  # BxHxW -> Bx1xHxW
                masks = masks.unsqueeze(1)
            with torch.no_grad():
                image_list.append(images.cpu())
                mask_list.append(masks.cpu())
                mask_pred_list.append(self.deployment_model(images.to(self.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),
        }
        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 (Optional[str]) –

  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: Optional[str] = "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 (torch.nn.Module) –

  The deployment model to use

• device (torch.device) –

  The device to run inference on

Source code in quadra/tasks/segmentation.py

@torch.no_grad()
def inference(
    self, dataloader: DataLoader, deployment_model: torch.nn.Module, 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, export_config=None)

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.

• export_config (Optional[DictConfig]) –

  Dictionary containing the export configuration, it should contain the following keys:

  • types: List of types to export.
  • input_shapes: Optional list of input shapes to use, they must be in the same order of the forward arguments.

Source code in quadra/tasks/classification.py

def __init__(
    self,
    config: DictConfig,
    output: DictConfig,
    device: str,
    export_config: Optional[DictConfig] = None,
):
    super().__init__(config=config, export_config=export_config)

    self._device = device
    self.output = output
    self._backbone: nn.Module
    self._trainer: SklearnClassificationTrainer
    self._model: ClassifierMixin
    self.metadata: Dict[str, Any] = {
        "test_confusion_matrix": [],
        "test_accuracy": [],
        "test_results": [],
        "test_labels": [],
    }
    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] = []

backbone: nn.Module 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.export_config is not None and len(self.export_config.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.export_config is None or len(self.export_config.types) == 0:
        log.info("No export type specified skipping export")
        return

    input_shapes = self.export_config.input_shapes

    for export_type in self.export_config.types:
        if export_type == "torchscript":
            out = export_torchscript_model(
                model=self.backbone,
                input_shapes=input_shapes,
                output_path=self.export_folder,
                half_precision=False,
            )

            if out is None:
                log.warning("Skipping torchscript export since the model is not supported")
                continue

            _, input_shapes = out
        elif export_type == "pytorch":
            os.makedirs(self.export_folder, exist_ok=True)
            # We only need to save classifier.joblib + backbone config file
            with open(os.path.join(self.export_folder, "backbone_config.yaml"), "w") as f:
                OmegaConf.save(self.config.backbone, f, resolve=True)
            log.info("backbone_config.yaml saved (export type 'pytorch')")
        else:
            log.warning("Export type: %s not implemented", export_type)

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

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

    model_json = {
        "input_size": input_shapes,
        "classes": idx_to_class,
        "mean": list(self.config.transforms.mean),
        "std": list(self.config.transforms.std),
    }

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

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,
        )
    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.train_dataloader_list = list(self.datamodule.train_dataloader())
    self.test_dataloader_list = list(self.datamodule.val_dataloader())
    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
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

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

    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,
    )

train()

Train the model.

Source code in quadra/tasks/classification.py

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

    class_to_keep = None

    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 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, _ = 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,
            )
        else:
            self.trainer.fit(train_dataloader=train_dataloader)
            _, pd_cm, accuracy, res, _ = 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,
            )

        # 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()
            ]
        )

train_full_data()

Train the model on train + validation.

Source code in quadra/tasks/classification.py

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) –

  Whether to compute gradcams

Source code in quadra/tasks/classification.py

def __init__(
    self,  # pylint: disable=W0613
    config: DictConfig,
    output: DictConfig,
    model_path: str,
    device: str,
    gradcam: bool = False,
    **kwargs: Any,
):
    super().__init__(config=config, model_path=model_path, device=device)
    self.gradcam = gradcam
    self.output = output
    self._backbone: nn.Module
    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: nn.Module 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")

    self.test_dataloader = self.datamodule.test_dataloader()

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

test()

Run the test.

Source code in quadra/tasks/classification.py

def test(self) -> None:
    """Run the test."""
    _, 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, export_config=None)

Bases: Generic[DataModuleT]

Base Experiment Task.

Parameters:

• config (DictConfig) –

  The experiment configuration.

• export_config (Optional[DictConfig]) –

  Dictionary containing the export configuration, it should contain the following keys:

  • types: List of types to export.
  • input_shapes: Optional list of input shapes to use, they must be in the same order of the forward arguments.

Source code in quadra/tasks/base.py

def __init__(self, config: DictConfig, export_config: Optional[DictConfig] = None):
    self.config = config
    self.export_config = export_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.export_config is not None and len(self.export_config.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!")