fix(general): Typo annotations and ruff

For what ever reason those checks were missed.
THey are now partial fixed.

Check #25 for final solution

Author: MaKaNu

tests/unit/test_builder.py

@@ -1,20 +1,12 @@
 from __future__ import annotations
 
 import ast
-import sys
 from pathlib import Path
+from tomllib import TOMLDecodeError
 from unittest.mock import mock_open, patch
 
 import pytest
 
-try:
-    from tomllib import TOMLDecodeError
-except ImportError:
-    try:
-        from tomli import TOMLDecodeError  # type: ignore  # noqa: PGH003
-    except ImportError:
-        sys.exit("Error: This program requires either tomllib or tomli but neither is available")
-
 from tipi.abstractions import Permanence, PipelineProcess
 from tipi.core.builder import PipelineBuilder, get_objects_for_pipeline
 from tipi.errors import (

tipi/core/processes.py

@@ -1,6 +1,3 @@
-# type: ignore
-# ruff: noqa
-
 # Random selected images for final result
 # Training: [tensor([28]), tensor([46]), tensor([60]), tensor([63]), tensor([90])]
 # Validation: [tensor([10]), tensor([18]), tensor([35]), tensor([57]), tensor([79]]]
@@ -14,14 +11,20 @@
 #   mask against GT if available, Difference of Mask with GT if available.
 
 from collections import OrderedDict
-from typing import Any
+from typing import Any, Protocol
 
 import torch
-
 import wandb
+
 from tipi.abstractions import PipelineProcess
 
 
+class ProgressTaskCallable(Protocol):
+    """Protocol for functions decorated with progress_task."""
+
+    def __call__(self, total: int) -> None: ...
+
+
 class ResultProcess(PipelineProcess):
     """
     ResultProcess is a class that handles the logging of images and their corresponding masks during
@@ -76,26 +79,27 @@ def __init__(self, controller: Any, force: bool, selected_images: dict[str, list
         self.valset = self.datasets.segnet_dataset_val
         self.testset = self.datasets.segnet_dataset_test
 
-    def execute(self):
+    def execute(self) -> None:
         train_image_log = self._get_log_train_images()
         val_image_log = self._get_log_val_images()
         test_image_log = self._get_log_test_images()
 
-        train_image_log(len(self.train_images_indices))
+        if self.train_images_indices:
+            train_image_log(len(self.train_images_indices))
         if self.datasets.val_available() and self.val_images_indices:
             val_image_log(len(self.val_images_indices))
         if self.datasets.test_available() and self.test_images_indices:
             test_image_log(len(self.test_images_indices))
 
-    def _get_log_train_images(self) -> callable:
+    def _get_log_train_images(self) -> ProgressTaskCallable:
         @self.progress_manager.progress_task("result", visible=False)
-        def _inner_log_image(total, task_id, progress):
+        def _inner_log_image(total: int, task_id: int, progress: Any) -> None:
             image_stack = []
             mask_stack = []
             pred_mask_stack = []
             for idx in range(total):
                 progress.advance(task_id)
-                selected_images = self.train_images_indices[idx]
+                selected_images = self.train_images_indices[idx]  # type: ignore[index]
                 image, mask = self.trainset[selected_images]
                 image_stack.append(image)
                 mask_stack.append(mask)
@@ -106,17 +110,17 @@ def _inner_log_image(total, task_id, progress):
             pred_masks = torch.cat(pred_mask_stack, 1)
             self._log_image(images, masks, pred_masks, "train")
 
-        return _inner_log_image
+        return _inner_log_image  # type: ignore[no-any-return]
 
-    def _get_log_val_images(self) -> callable:
+    def _get_log_val_images(self) -> ProgressTaskCallable:
         @self.progress_manager.progress_task("result", visible=False)
-        def _inner_log_image(total, task_id, progress):
+        def _inner_log_image(total: int, task_id: int, progress: Any) -> None:
             image_stack = []
             mask_stack = []
             pred_mask_stack = []
             for idx in range(total):
                 progress.advance(task_id)
-                selected_images = self.val_images_indices[idx]
+                selected_images = self.val_images_indices[idx]  # type: ignore[index]
                 image, mask = self.valset[selected_images]
                 image_stack.append(image)
                 mask_stack.append(mask)
@@ -127,17 +131,17 @@ def _inner_log_image(total, task_id, progress):
             pred_masks = torch.cat(pred_mask_stack, 1)
             self._log_image(images, masks, pred_masks, "val")
 
-        return _inner_log_image
+        return _inner_log_image  # type: ignore[no-any-return]
 
-    def _get_log_test_images(self) -> callable:
+    def _get_log_test_images(self) -> ProgressTaskCallable:
         @self.progress_manager.progress_task("result", visible=False)
-        def _inner_log_image(total, task_id, progress):
+        def _inner_log_image(total: int, task_id: int, progress: Any) -> None:
             image_stack = []
             mask_stack = []
             pred_mask_stack = []
             for idx in range(total):
                 progress.advance(task_id)
-                selected_images = self.test_images_indices[idx]
+                selected_images = self.test_images_indices[idx]  # type: ignore[index]
                 image, mask = self.testset[selected_images]
                 image_stack.append(image)
                 mask_stack.append(mask)
@@ -148,26 +152,26 @@ def _inner_log_image(total, task_id, progress):
             pred_masks = torch.cat(pred_mask_stack, 1)
             self._log_image(images, masks, pred_masks, "test")
 
-        return _inner_log_image
+        return _inner_log_image  # type: ignore[no-any-return]
 
-    def _inference_model(self, image):
+    def _inference_model(self, image: Any) -> Any:
         self.model.eval()
         with torch.no_grad():
             return self.model(image.unsqueeze(0).to(self.device))
 
-    def _get_pred_mask(self, pred):
+    def _get_pred_mask(self, pred: Any) -> Any:
         if isinstance(pred, OrderedDict):
             pred = pred["out"]
         return pred.argmax(dim=1).squeeze(0).cpu()
 
-    def _get_mask_difference(self, mask, pred_mask):
+    def _get_mask_difference(self, mask: Any, pred_mask: Any) -> Any:
         mask[mask == 255] = 0
         mask_difference = mask - pred_mask
         if mask_difference.min() < 0:
             mask_difference = mask_difference + mask_difference.min().abs()
         return mask_difference.to(torch.uint8)
 
-    def _log_image(self, image, mask, pred_mask, dataset):
+    def _log_image(self, image: Any, mask: Any, pred_mask: Any, dataset: str) -> None:
         class_labels = self._swap_labels(self.datasets.data_container.classes)
         just_image = wandb.Image(image, caption=f"{dataset} images")
         image_with_mask = wandb.Image(
@@ -186,8 +190,8 @@ def _log_image(self, image, mask, pred_mask, dataset):
         wandb.log({f"{dataset}_images_with_mask": image_with_mask})
         wandb.log({f"{dataset}_mask_difference": mask_difference})
 
-    def _swap_labels(self, labels):
+    def _swap_labels(self, labels: dict[Any, Any]) -> dict[Any, Any]:
         return {v: k for k, v in labels.items()}
 
-    def skip(self):
+    def skip(self) -> bool:
         return False

tipi/decorators.py

@@ -97,7 +97,7 @@ def __init__(self, controller: Any, force: bool = False, **kwargs: Any) -> None:
             def execute(self) -> None:
                 """Execute the wrapped function."""
                 # Set pipeline context so helpers work
-                from tipi.helpers import (  # type: ignore[attr-defined]
+                from tipi.helpers import (
                     clear_pipeline_context,
                     set_pipeline_context,
                 )

tipi/errors.py

@@ -47,40 +47,47 @@ def __init__(self, code: str, message: str):
 @dataclass
 class BuilderError(RuntimeError):
     error_value: Any
+    error_code: ErrorCode | None = None
 
-    def __post_init__(self, error_code: ErrorCode) -> None:  # type: ignore  # noqa: PGH003
+    def __post_init__(self) -> None:
+        pass
+
+    def _set_error_code(self, error_code: ErrorCode) -> None:
+        """Set the error code (called by subclasses)."""
         self.error_code = error_code
 
     def __str__(self) -> str:
-        return f"[{self.error_code.code}]: {self.error_code.message}: {self.error_value}"
+        if self.error_code:
+            return f"[{self.error_code.code}]: {self.error_code.message}: {self.error_value}"
+        return f"BuilderError: {self.error_value}"
 
 
 class ConfigNotFoundError(BuilderError):
     """Raised when the builder configuration file does not exists"""
 
     def __post_init__(self) -> None:
-        super().__post_init__(ErrorCode.CONFIG_MISSING)
+        self._set_error_code(ErrorCode.CONFIG_MISSING)
 
 
 class ConfigPermissionError(BuilderError):
     """Raised when the builder configuration file does not exists"""
 
     def __post_init__(self) -> None:
-        super().__post_init__(ErrorCode.CONFIG_PERMISSION)
+        self._set_error_code(ErrorCode.CONFIG_PERMISSION)
 
 
 class ConfigInvalidTomlError(BuilderError):
     """Raised when the configuration file is not valid toml"""
 
     def __post_init__(self) -> None:
-        super().__post_init__(ErrorCode.CONFIG_INVALID)
+        self._set_error_code(ErrorCode.CONFIG_INVALID)
 
 
 class ConfigSectionError(BuilderError):
     """Raised for config section missing"""
 
     def __post_init__(self) -> None:
-        super().__post_init__(ErrorCode.CONFIG_SECTION)
+        self._set_error_code(ErrorCode.CONFIG_SECTION)
 
 
 class InvalidConfigError(Exception):
@@ -95,21 +102,21 @@ class RegistryError(BuilderError):
     """Raised for class registration issues"""
 
     def __post_init__(self) -> None:
-        super().__post_init__(ErrorCode.REGISTRY_INVALID)
+        self._set_error_code(ErrorCode.REGISTRY_INVALID)
 
 
 class RegistryParamError(BuilderError):
     """Raised for class instatioation with wrong params"""
 
     def __post_init__(self) -> None:
-        super().__post_init__(ErrorCode.REGISTRY_PARAM)
+        self._set_error_code(ErrorCode.REGISTRY_PARAM)
 
 
 class InstTypeError(BuilderError):
     """Raised when type in config not set"""
 
     def __post_init__(self) -> None:
-        super().__post_init__(ErrorCode.INST_TYPE)
+        self._set_error_code(ErrorCode.INST_TYPE)
 
 
 ## Execution

tipi/helpers.py

@@ -1,5 +1,3 @@
-# type: ignore
-# ruff: noqa
 """Helper utilities for smooth transition from scripts to pipeline.
 
 This module provides simple, script-friendly functions that work standalone
@@ -23,13 +21,14 @@
 
 from __future__ import annotations
 
-from typing import Any, Iterable, Optional
+from collections.abc import Iterable
+from typing import Any
 
 import torch
 from rich.progress import track
 
 # Global context that pipeline can set
-_pipeline_context: Optional[dict[str, Any]] = None
+_pipeline_context: dict[str, Any] | None = None
 
 
 def set_pipeline_context(context: dict[str, Any]) -> None:
@@ -51,7 +50,7 @@ def clear_pipeline_context() -> None:
 def progress_bar(
     iterable: Iterable,
     desc: str = "Processing",
-    total: Optional[int] = None,
+    total: int | None = None,
 ) -> Iterable:
     """Progress bar that auto-integrates with pipeline or uses rich.track.
 
@@ -92,12 +91,12 @@ class Logger:
     - Pipeline: Uses pipeline's WandBManager if available
     """
 
-    def __init__(self):
+    def __init__(self) -> None:
         self._wandb_initialized = False
-        self._project = None
-        self._entity = None
+        self._project: str | None = None
+        self._entity: str | None = None
 
-    def init(self, project: str, entity: Optional[str] = None, name: Optional[str] = None, **kwargs) -> None:
+    def init(self, project: str, entity: str | None = None, name: str | None = None, **kwargs: Any) -> None:
         """Initialize logger (manual WandB setup or use pipeline's).
 
         Args:
@@ -125,7 +124,7 @@ def init(self, project: str, entity: Optional[str] = None, name: Optional[str] =
         except ImportError:
             print("WandB not available, logging to console only")
 
-    def log(self, metrics: dict[str, Any], step: Optional[int] = None) -> None:
+    def log(self, metrics: dict[str, Any], step: int | None = None) -> None:
         """Log metrics to WandB or console.
 
         Args:
@@ -174,7 +173,7 @@ def get_device(self) -> torch.device:
         if _pipeline_context:
             device_perm = _pipeline_context.get("device")
             if device_perm:
-                return device_perm.device
+                return torch.device(device_perm.device)
 
         # Standalone: Select best device
         if torch.cuda.is_available():
@@ -189,9 +188,9 @@ def get_device(self) -> torch.device:
 
 
 __all__ = [
-    "progress_bar",
-    "logger",
+    "clear_pipeline_context",
     "device_manager",
+    "logger",
+    "progress_bar",
     "set_pipeline_context",
-    "clear_pipeline_context",
 ]