Linting: Enable PLR6301, # could be a function, class method or static method (open-edge-platform#2288)

* PLR6301 Method could be a function, class method, or static method

* "PLR6301", # could be a function, class method or static method

Signed-off-by: Samet Akcay <[email protected]>

* revert missing pytest fixture

* refactor: convert instance methods to static methods where applicable

Signed-off-by: Samet Akcay <[email protected]>

* Fix data tests

Signed-off-by: Samet Akcay <[email protected]>

* Fix shanghai tech config filename

* Fix base anomaly module tests

* Fix base anomaly module tests

Signed-off-by: Samet Akcay <[email protected]>

* add staticmethod before parametrize

Signed-off-by: Samet Akcay <[email protected]>

---------

Signed-off-by: Samet Akcay <[email protected]>

Author: samet-akcay

configs/data/shanghaitec.yaml renamed to configs/data/shanghaitech.yaml

@@ -165,7 +165,6 @@ lint.ignore = [
     "A004", # import is shadowing a Python built-in
     "A005", # Module is shadowing a Python built-in
     "B909", # Mutation to loop iterable during iteration
-    "PLR6301", # could be a function, class method or static method
     "PLC2701", # Private name import
     "PLC0415", # import should be at the top of the file
     "PLR0917", # Too many positional arguments

pyproject.toml

@@ -33,7 +33,8 @@ class GraphLogger(Callback):
         >>> engine = Engine(logger=logger, callbacks=callbacks)
     """
 
-    def on_train_start(self, trainer: Trainer, pl_module: LightningModule) -> None:
+    @staticmethod
+    def on_train_start(trainer: Trainer, pl_module: LightningModule) -> None:
         """Log model graph to respective logger.
 
         Args:
@@ -47,7 +48,8 @@ def on_train_start(self, trainer: Trainer, pl_module: LightningModule) -> None:
                 logger.watch(pl_module, log_graph=True, log="all")
                 break
 
-    def on_train_end(self, trainer: Trainer, pl_module: LightningModule) -> None:
+    @staticmethod
+    def on_train_end(trainer: Trainer, pl_module: LightningModule) -> None:
         """Unwatch model if configured for wandb and log it model graph in Tensorboard if specified.
 
         Args:

src/anomalib/callbacks/graph.py

@@ -98,11 +98,8 @@ def setup(
                 pl_module.pixel_metrics = create_metric_collection(pixel_metric_names, "pixel_")
             self._set_threshold(pl_module)
 
-    def on_validation_epoch_start(
-        self,
-        trainer: Trainer,
-        pl_module: AnomalyModule,
-    ) -> None:
+    @staticmethod
+    def on_validation_epoch_start(trainer: Trainer, pl_module: AnomalyModule) -> None:
         del trainer  # Unused argument.
 
         pl_module.image_metrics.reset()
@@ -123,21 +120,14 @@ def on_validation_batch_end(
             self._outputs_to_device(outputs)
             self._update_metrics(pl_module.image_metrics, pl_module.pixel_metrics, outputs)
 
-    def on_validation_epoch_end(
-        self,
-        trainer: Trainer,
-        pl_module: AnomalyModule,
-    ) -> None:
+    def on_validation_epoch_end(self, trainer: Trainer, pl_module: AnomalyModule) -> None:
         del trainer  # Unused argument.
 
         self._set_threshold(pl_module)
         self._log_metrics(pl_module)
 
-    def on_test_epoch_start(
-        self,
-        trainer: Trainer,
-        pl_module: AnomalyModule,
-    ) -> None:
+    @staticmethod
+    def on_test_epoch_start(trainer: Trainer, pl_module: AnomalyModule) -> None:
         del trainer  # Unused argument.
 
         pl_module.image_metrics.reset()
@@ -158,16 +148,13 @@ def on_test_batch_end(
             self._outputs_to_device(outputs)
             self._update_metrics(pl_module.image_metrics, pl_module.pixel_metrics, outputs)
 
-    def on_test_epoch_end(
-        self,
-        trainer: Trainer,
-        pl_module: AnomalyModule,
-    ) -> None:
+    def on_test_epoch_end(self, trainer: Trainer, pl_module: AnomalyModule) -> None:
         del trainer  # Unused argument.
 
         self._log_metrics(pl_module)
 
-    def _set_threshold(self, pl_module: AnomalyModule) -> None:
+    @staticmethod
+    def _set_threshold(pl_module: AnomalyModule) -> None:
         pl_module.image_metrics.set_threshold(pl_module.image_threshold.value.item())
         pl_module.pixel_metrics.set_threshold(pl_module.pixel_threshold.value.item())
 

src/anomalib/callbacks/metrics.py

@@ -40,7 +40,8 @@ def __next__(self) -> torch.Tensor:
         loaded_item = next(self._data_loader_iter)
         return loaded_item["image"]
 
-    def get_inputs(self, dataloader_output: dict[str, str | torch.Tensor]) -> tuple[tuple, dict]:
+    @staticmethod
+    def get_inputs(dataloader_output: dict[str, str | torch.Tensor]) -> tuple[tuple, dict]:
         """Get input to model.
 
         Returns:
@@ -49,7 +50,8 @@ def get_inputs(self, dataloader_output: dict[str, str | torch.Tensor]) -> tuple[
         """
         return (dataloader_output,), {}
 
-    def get_target(self, _):  # noqa: ANN001, ANN201
+    @staticmethod
+    def get_target(_) -> None:  # noqa: ANN001
         """Return structure for ground truth in loss criterion based on dataloader output.
 
         This implementation does not do anything and is a placeholder.

src/anomalib/callbacks/nncf/utils.py

@@ -23,7 +23,8 @@ class _MinMaxNormalizationCallback(NormalizationCallback):
     Note: This callback is set within the Engine.
     """
 
-    def setup(self, trainer: Trainer, pl_module: AnomalyModule, stage: str | None = None) -> None:
+    @staticmethod
+    def setup(trainer: Trainer, pl_module: AnomalyModule, stage: str | None = None) -> None:
         """Add min_max metrics to normalization metrics."""
         del trainer, stage  # These variables are not used.
 
@@ -48,23 +49,25 @@ def setup(self, trainer: Trainer, pl_module: AnomalyModule, stage: str | None =
                 msg = f"Expected normalization_metric {name} to be of type MinMax, got {type(metric)}"
                 raise TypeError(msg)
 
-    def on_test_start(self, trainer: Trainer, pl_module: AnomalyModule) -> None:
+    @staticmethod
+    def on_test_start(trainer: Trainer, pl_module: AnomalyModule) -> None:
         """Call when the test begins."""
         del trainer  # `trainer` variable is not used.
 
         for metric in (pl_module.image_metrics, pl_module.pixel_metrics):
             if metric is not None:
                 metric.set_threshold(0.5)
 
-    def on_validation_epoch_start(self, trainer: Trainer, pl_module: AnomalyModule) -> None:
+    @staticmethod
+    def on_validation_epoch_start(trainer: Trainer, pl_module: AnomalyModule) -> None:
         """Call when the validation epoch begins."""
         del trainer  # `trainer` variable is not used.
 
         if hasattr(pl_module, "normalization_metrics"):
             pl_module.normalization_metrics.reset()
 
+    @staticmethod
     def on_validation_batch_end(
-        self,
         trainer: Trainer,
         pl_module: AnomalyModule,
         outputs: STEP_OUTPUT,

src/anomalib/callbacks/normalization/min_max_normalization.py

@@ -133,7 +133,8 @@ def _load_from_config(self, threshold: DictConfig | str | ListConfig | list[dict
             msg = f"Invalid threshold config {threshold}"
             raise TypeError(msg)
 
-    def _get_threshold_from_config(self, threshold: DictConfig | str | dict[str, str | float]) -> Threshold:
+    @staticmethod
+    def _get_threshold_from_config(threshold: DictConfig | str | dict[str, str | float]) -> Threshold:
         """Return the instantiated threshold object.
 
         Example:
@@ -170,7 +171,8 @@ def _get_threshold_from_config(self, threshold: DictConfig | str | dict[str, str
         class_ = getattr(module, class_path)
         return class_(**init_args)
 
-    def _reset(self, pl_module: AnomalyModule) -> None:
+    @staticmethod
+    def _reset(pl_module: AnomalyModule) -> None:
         pl_module.image_threshold.reset()
         pl_module.pixel_threshold.reset()
 
@@ -182,14 +184,16 @@ def _outputs_to_cpu(self, output: STEP_OUTPUT) -> STEP_OUTPUT | dict[str, Any]:
             output = output.cpu()
         return output
 
-    def _update(self, pl_module: AnomalyModule, outputs: STEP_OUTPUT) -> None:
+    @staticmethod
+    def _update(pl_module: AnomalyModule, outputs: STEP_OUTPUT) -> None:
         pl_module.image_threshold.cpu()
         pl_module.image_threshold.update(outputs["pred_scores"], outputs["label"].int())
         if "mask" in outputs and "anomaly_maps" in outputs:
             pl_module.pixel_threshold.cpu()
             pl_module.pixel_threshold.update(outputs["anomaly_maps"], outputs["mask"].int())
 
-    def _compute(self, pl_module: AnomalyModule) -> None:
+    @staticmethod
+    def _compute(pl_module: AnomalyModule) -> None:
         pl_module.image_threshold.compute()
         if pl_module.pixel_threshold._update_called:  # noqa: SLF001
             pl_module.pixel_threshold.compute()

src/anomalib/callbacks/thresholding.py

@@ -146,8 +146,8 @@ def on_predict_batch_end(
     def on_predict_end(self, trainer: Trainer, pl_module: AnomalyModule) -> None:
         return self.on_test_end(trainer, pl_module)
 
+    @staticmethod
     def _add_to_logger(
-        self,
         result: GeneratorResult,
         module: AnomalyModule,
         trainer: Trainer,

src/anomalib/callbacks/visualizer.py

@@ -64,7 +64,8 @@ def __init__(self, args: Sequence[str] | None = None, run: bool = True) -> None:
         if run:
             self._run_subcommand()
 
-    def init_parser(self, **kwargs) -> ArgumentParser:
+    @staticmethod
+    def init_parser(**kwargs) -> ArgumentParser:
         """Method that instantiates the argument parser."""
         kwargs.setdefault("dump_header", [f"anomalib=={__version__}"])
         parser = ArgumentParser(formatter_class=CustomHelpFormatter, **kwargs)
@@ -139,7 +140,8 @@ def add_subcommands(self, **kwargs) -> None:
                 self.subcommand_parsers[subcommand] = sub_parser
                 parser_subcommands.add_subcommand(subcommand, sub_parser, help=value["description"])
 
-    def add_arguments_to_parser(self, parser: ArgumentParser) -> None:
+    @staticmethod
+    def add_arguments_to_parser(parser: ArgumentParser) -> None:
         """Extend trainer's arguments to add engine arguments.
 
         .. note::
@@ -399,8 +401,8 @@ def export(self) -> Callable:
         """Export the model using engine's export method."""
         return self.engine.export
 
+    @staticmethod
     def _add_trainer_arguments_to_parser(
-        self,
         parser: ArgumentParser,
         add_optimizer: bool = False,
         add_scheduler: bool = False,
@@ -427,7 +429,8 @@ def _add_trainer_arguments_to_parser(
                 **scheduler_kwargs,
             )
 
-    def _add_default_arguments_to_parser(self, parser: ArgumentParser) -> None:
+    @staticmethod
+    def _add_default_arguments_to_parser(parser: ArgumentParser) -> None:
         """Adds default arguments to the parser."""
         parser.add_argument(
             "--seed_everything",

src/anomalib/cli/cli.py

@@ -118,7 +118,8 @@ class ShanghaiTechTrainClipsIndexer(ClipsIndexer):
     clips indexer implementations are needed.
     """
 
-    def get_mask(self, idx: int) -> torch.Tensor | None:
+    @staticmethod
+    def get_mask(idx: int) -> torch.Tensor | None:
         """No masks available for training set."""
         del idx  # Unused argument
         return None

src/anomalib/data/video/shanghaitech.py

@@ -118,7 +118,7 @@ def _normalize(
 
         return anomaly_maps, float(pred_scores)
 
-    def _load_metadata(self, path: str | Path | dict | None = None) -> dict | DictConfig:
+    def _load_metadata(self, path: str | Path | dict | None = None) -> dict | DictConfig:  # noqa: PLR6301
         """Load the meta data from the given path.
 
         Args:

src/anomalib/deploy/inferencers/base_inferencer.py

@@ -150,7 +150,8 @@ def load_model(self, path: str | Path | tuple[bytes, bytes]) -> tuple[Any, Any,
 
         return input_blob, output_blob, compile_model
 
-    def pre_process(self, image: np.ndarray) -> np.ndarray:
+    @staticmethod
+    def pre_process(image: np.ndarray) -> np.ndarray:
         """Pre-process the input image by applying transformations.
 
         Args:

src/anomalib/deploy/inferencers/openvino_inferencer.py

@@ -23,7 +23,7 @@ class Threshold(Metric):
     def __init__(self, **kwargs) -> None:
         super().__init__(**kwargs)
 
-    def compute(self) -> torch.Tensor:
+    def compute(self) -> torch.Tensor:  # noqa: PLR6301
         """Compute the threshold.
 
         Returns:
@@ -32,7 +32,7 @@ def compute(self) -> torch.Tensor:
         msg = "Subclass of Threshold must implement the compute method"
         raise NotImplementedError(msg)
 
-    def update(self, *args, **kwargs) -> None:  # noqa: ARG002
+    def update(self, *args, **kwargs) -> None:  # noqa: ARG002, PLR6301
         """Update the metric state.
 
         Args:

src/anomalib/metrics/threshold/base.py

@@ -56,7 +56,8 @@ def compute(self) -> torch.Tensor:
         """
         return self.value
 
-    def update(self, *args, **kwargs) -> None:
+    @staticmethod
+    def update(*args, **kwargs) -> None:
         """Do nothing.
 
         Args:

src/anomalib/metrics/threshold/manual_threshold.py

@@ -215,7 +215,8 @@ def _add_metrics(self, name: str, state_dict: OrderedDict[str, torch.Tensor]) ->
                 logger.info("Loading %s metrics from state dict", class_name)
                 metrics.add_metrics(metrics_cls())
 
-    def _get_instance(self, state_dict: OrderedDict[str, Any], dict_key: str) -> Threshold:
+    @staticmethod
+    def _get_instance(state_dict: OrderedDict[str, Any], dict_key: str) -> Threshold:
         """Get the threshold class from the ``state_dict``."""
         class_path = state_dict.pop(dict_key)
         module = importlib.import_module(".".join(class_path.split(".")[:-1]))
@@ -240,7 +241,7 @@ def set_transform(self, transform: Transform) -> None:
         """Update the transform linked to the model instance."""
         self._transform = transform
 
-    def configure_transforms(self, image_size: tuple[int, int] | None = None) -> Transform:
+    def configure_transforms(self, image_size: tuple[int, int] | None = None) -> Transform:  # noqa: PLR6301
         """Default transforms.
 
         The default transform is resize to 256x256 and normalize to ImageNet stats. Individual models can override

src/anomalib/models/components/base/anomaly_module.py

@@ -310,8 +310,8 @@ def _compress_ov_model(
 
         return model
 
+    @staticmethod
     def _post_training_quantization_ov(
-        self,
         model: "CompiledModel",
         datamodule: AnomalibDataModule | None = None,
     ) -> "CompiledModel":
@@ -346,8 +346,8 @@ def _post_training_quantization_ov(
         calibration_dataset = nncf.Dataset(dataloader, lambda x: x["image"])
         return nncf.quantize(model, calibration_dataset)
 
+    @staticmethod
     def _accuracy_control_quantization_ov(
-        self,
         model: "CompiledModel",
         datamodule: AnomalibDataModule | None = None,
         metric: Metric | str | None = None,

src/anomalib/models/components/base/export_mixin.py

@@ -98,7 +98,8 @@ def _sparse_random_matrix(self, n_features: int) -> torch.Tensor:
 
         return components
 
-    def _johnson_lindenstrauss_min_dim(self, n_samples: int, eps: float = 0.1) -> int | np.integer:
+    @staticmethod
+    def _johnson_lindenstrauss_min_dim(n_samples: int, eps: float = 0.1) -> int | np.integer:
         """Find a 'safe' number of components to randomly project to.
 
         Ref eqn 2.1 https://cseweb.ucsd.edu/~dasgupta/papers/jl.pdf

src/anomalib/models/components/dimensionality_reduction/random_projection.py

@@ -330,7 +330,8 @@ def forward(
 
         return (x_out,), log_jac_det
 
-    def output_dims(self, input_dims: list[tuple[int]]) -> list[tuple[int]]:
+    @staticmethod
+    def output_dims(input_dims: list[tuple[int]]) -> list[tuple[int]]:
         """Output dimensions of the layer.
 
         Args:

src/anomalib/models/components/flow/all_in_one_block.py

@@ -104,7 +104,8 @@ def validation_step(self, batch: dict[str, str | torch.Tensor], *args, **kwargs)
         batch["anomaly_maps"] = self.model(batch["image"])
         return batch
 
-    def backward(self, loss: torch.Tensor, *args, **kwargs) -> None:
+    @staticmethod
+    def backward(loss: torch.Tensor, *args, **kwargs) -> None:
         """Perform backward-pass for the CFA model.
 
         Args:

src/anomalib/models/image/cfa/lightning_model.py

@@ -10,7 +10,8 @@
 class CsFlowLoss(nn.Module):
     """Loss function for the CS-Flow Model Implementation."""
 
-    def forward(self, z_dist: torch.Tensor, jacobians: torch.Tensor) -> torch.Tensor:
+    @staticmethod
+    def forward(z_dist: torch.Tensor, jacobians: torch.Tensor) -> torch.Tensor:
         """Compute the loss CS-Flow.
 
         Args: