patch

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