From 52023c84c3d68c80a00cf925ecc5da600a46ac58 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:24:20 +0200 Subject: [PATCH 01/24] Add lightweight EfficientAD implementation --- .../algorithm/efficientad/efficientad.py | 168 ++++++++++++++++++ 1 file changed, 168 insertions(+) create mode 100644 anomavision/algorithm/efficientad/efficientad.py diff --git a/anomavision/algorithm/efficientad/efficientad.py b/anomavision/algorithm/efficientad/efficientad.py new file mode 100644 index 0000000..c55bd09 --- /dev/null +++ b/anomavision/algorithm/efficientad/efficientad.py @@ -0,0 +1,168 @@ +"""Lightweight EfficientAD-style teacher-student anomaly detection.""" + +from __future__ import annotations + +from typing import Dict, List, Optional, Tuple + +import torch +import torch.nn as nn +import torch.nn.functional as F + +from ..common.feature_extraction import ResnetEmbeddingsExtractor + + +class _Student(nn.Module): + """Small CNN that learns normal ResNet layer-1 features.""" + + def __init__(self, channels: int = 64) -> None: + super().__init__() + self.net = nn.Sequential( + nn.Conv2d(3, 32, 5, stride=2, padding=2), + nn.ReLU(inplace=True), + nn.Conv2d(32, channels, 5, stride=2, padding=2), + nn.ReLU(inplace=True), + nn.Conv2d(channels, channels, 3, padding=1), + ) + + def forward(self, x: torch.Tensor) -> torch.Tensor: + return self.net(x) + + +class EfficientAD(torch.nn.Module): + """Minimal teacher-student EfficientAD implementation. + + A frozen ImageNet ResNet teacher provides layer-1 features and a very small + CNN student learns those features from normal images. At inference, the + teacher-student feature error is used as the anomaly map. + + The public API mirrors PaDiM/PatchCore: ``fit`` trains on normal images and + ``predict`` returns image scores plus a spatial anomaly map. + """ + + def __init__( + self, + backbone: str = "resnet18", + device: torch.device = torch.device("cpu"), + layer_indices: Optional[List[int]] = None, + epochs: int = 5, + learning_rate: float = 1e-3, + ) -> None: + super().__init__() + if backbone not in {"resnet18"}: + raise ValueError("EfficientAD lightweight supports backbone='resnet18' only.") + self.device = torch.device(device) + self.backbone = backbone + self.layer_indices = list(layer_indices or [0]) + if self.layer_indices != [0]: + raise ValueError("EfficientAD lightweight uses layer_indices=[0].") + self.epochs = max(1, int(epochs)) + self.learning_rate = float(learning_rate) + + self.teacher = ResnetEmbeddingsExtractor(backbone, self.device) + for parameter in self.teacher.parameters(): + parameter.requires_grad_(False) + self.teacher.eval() + + self.student = _Student(64).to(self.device) + self._fitted = False + + @torch.no_grad() + def _teacher_features(self, batch: torch.Tensor) -> torch.Tensor: + features, width, height = self.teacher(batch, layer_indices=[0]) + return features.reshape(batch.shape[0], width, height, 64).permute(0, 3, 1, 2) + + def _loss(self, batch: torch.Tensor) -> torch.Tensor: + with torch.no_grad(): + teacher = self._teacher_features(batch) + teacher = F.normalize(teacher, dim=1) + student = F.normalize(self.student(batch), dim=1) + return F.mse_loss(student, teacher) + + @torch.no_grad() + def _scores(self, batch: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]: + teacher = F.normalize(self._teacher_features(batch), dim=1) + student = F.normalize(self.student(batch.to(self.device)), dim=1) + score = (teacher - student).pow(2).mean(dim=1) + image_scores = score.flatten(1).amax(1) + score_map = F.interpolate( + score.unsqueeze(1), + size=batch.shape[-2:], + mode="bilinear", + align_corners=False, + ).squeeze(1) + return image_scores, score_map + + def fit(self, dataloader: torch.utils.data.DataLoader, extractions: int = 1) -> None: + """Train the small student using only normal training images.""" + optimizer = torch.optim.Adam(self.student.parameters(), lr=self.learning_rate) + self.student.train() + for _ in range(self.epochs): + for item in dataloader: + batch = item[0] if isinstance(item, (tuple, list)) else item + batch = batch.to(self.device, non_blocking=True) + optimizer.zero_grad(set_to_none=True) + loss = self._loss(batch) + loss.backward() + optimizer.step() + self.student.eval() + self._fitted = True + + @torch.no_grad() + def forward( + self, x: torch.Tensor, return_map: bool = True, export: bool = False + ) -> Tuple[torch.Tensor, Optional[torch.Tensor]]: + """Return image-level anomaly scores and an optional spatial map.""" + if not self._fitted: + raise RuntimeError("EfficientAD is not fitted. Call fit() first.") + image_scores, score_map = self._scores(x) + return image_scores, score_map if return_map else None + + def predict( + self, batch: torch.Tensor, export: bool = False + ) -> Tuple[torch.Tensor, torch.Tensor]: + """Run inference using the standard AnomaVision prediction contract.""" + return self.forward(batch, return_map=True, export=export) + + def to_device(self, device: torch.device) -> None: + self.device = torch.device(device) + self.teacher.to_device(self.device) + self.student.to(self.device) + + def save_statistics(self, path: str, half: Optional[bool] = False) -> None: + """Save a compact student/teacher deployment artifact.""" + if not self._fitted: + raise RuntimeError("EfficientAD is not fitted. Call fit() first.") + student_state = { + key: value.detach().cpu().half() if half and value.is_floating_point() else value.detach().cpu() + for key, value in self.student.state_dict().items() + } + torch.save( + { + "student": student_state, + "backbone": self.backbone, + "layer_indices": self.layer_indices, + "epochs": self.epochs, + "learning_rate": self.learning_rate, + "model_type": "efficientad", + "dtype": "fp16" if half else "fp32", + }, + path, + ) + + +def build_efficientad_from_stats( + stats: Dict, device: str = "cpu", force_precision: Optional[str] = None +) -> EfficientAD: + """Build EfficientAD from a compact statistics artifact.""" + model = EfficientAD( + backbone=str(stats.get("backbone", "resnet18")), + device=torch.device(device), + layer_indices=list(stats.get("layer_indices", [0])), + epochs=int(stats.get("epochs", 5)), + learning_rate=float(stats.get("learning_rate", 1e-3)), + ) + state = stats["student"] + model.student.load_state_dict({key: value.float() for key, value in state.items()}) + model.student.eval() + model._fitted = True + return model From 5bb8b9e82554c2a6070226763d901115160ce883 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:24:24 +0200 Subject: [PATCH 02/24] Expose EfficientAD algorithm --- anomavision/algorithm/efficientad/__init__.py | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 anomavision/algorithm/efficientad/__init__.py diff --git a/anomavision/algorithm/efficientad/__init__.py b/anomavision/algorithm/efficientad/__init__.py new file mode 100644 index 0000000..470fc91 --- /dev/null +++ b/anomavision/algorithm/efficientad/__init__.py @@ -0,0 +1,5 @@ +"""Lightweight EfficientAD anomaly detection.""" + +from .efficientad import EfficientAD, build_efficientad_from_stats + +__all__ = ["EfficientAD", "build_efficientad_from_stats"] From 67abb369c60be006dd026881f69d73201375c2e0 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:24:28 +0200 Subject: [PATCH 03/24] Integrate EfficientAD into public API --- anomavision/__init__.py | 1 + 1 file changed, 1 insertion(+) diff --git a/anomavision/__init__.py b/anomavision/__init__.py index 6eff355..dcd0e87 100644 --- a/anomavision/__init__.py +++ b/anomavision/__init__.py @@ -11,6 +11,7 @@ from .algorithm.common.feature_extraction import ResnetEmbeddingsExtractor from .algorithm.padim import Padim from .algorithm.patchcore import PatchCore +from .algorithm.efficientad import EfficientAD from .datasets.dataset import AnodetDataset from .datasets.mvtec_dataset import MVTecDataset from .sampling_methods.kcenter_greedy import kCenterGreedy From 58593439c54039477b24ded0ef082c786a0efb7d Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:24:42 +0200 Subject: [PATCH 04/24] Integrate EfficientAD into training pipeline --- anomavision/train.py | 250 +++++++------------------------------------ 1 file changed, 40 insertions(+), 210 deletions(-) diff --git a/anomavision/train.py b/anomavision/train.py index 7761310..3494928 100644 --- a/anomavision/train.py +++ b/anomavision/train.py @@ -17,185 +17,39 @@ def create_parser(add_help: bool = True) -> argparse.ArgumentParser: parser = argparse.ArgumentParser( - description="Train PaDiM (args OR config).", add_help=add_help - ) - # meta - parser.add_argument( - "--config", type=str, default="config.yml", help="Path to config.yml/.json" - ) - # dataset - parser.add_argument( - "--dataset_path", - type=str, - default=None, - help='Path to the dataset folder containing "train/good" images.', - ) - - # preprocessing - parser.add_argument( - "--resize", - type=int, - nargs="*", - default=None, - metavar=("W", "H"), - help="Resize before processing. Provide one value for a square resize (e.g., 256) or two values for width and height (e.g., 256 192). Omit to keep original size.", - ) - parser.add_argument( - "--crop_size", - type=int, - nargs="*", - default=None, - metavar=("W", "H"), - help="Apply a center (or configured) crop. One value for a square crop (e.g., 224) or two for width and height (e.g., 224 224). Omit to disable cropping.", - ) - parser.add_argument( - "--normalize", - action="store_true", - default=None, - help="Enable input normalization. If set, inputs are normalized using --norm_mean/--norm_std if provided (commonly ImageNet stats).", - ) - parser.add_argument( - "--no_normalize", - action="store_true", - default=None, - help="Disable input normalization explicitly. If both --normalize and --no_normalize are set, this flag should take precedence in your code.", - ) - parser.add_argument( - "--norm_mean", - type=float, - nargs=3, - default=None, - metavar=("R", "G", "B"), - help="Per-channel RGB mean used when normalization is enabled. Example: 0.485 0.456 0.406.", - ) - parser.add_argument( - "--norm_std", - type=float, - nargs=3, - default=None, - metavar=("R", "G", "B"), - help="Per-channel RGB standard deviation used when normalization is enabled. Example: 0.229 0.224 0.225.", - ) - - # train - parser.add_argument( - "--backbone", - type=str, - choices=["resnet18", "wide_resnet50"], - default=None, - help="Backbone network to use for feature extraction.", - ) - parser.add_argument( - "--batch_size", - type=int, - default=None, - help="Batch size used during training and inference.", - ) - parser.add_argument( - "--feat_dim", - type=int, - default=None, - help="Number of random feature dimensions to keep.", - ) - parser.add_argument( - "--layer_indices", - type=int, - nargs="+", - default=None, - help="List of layer indices to extract features from, e.g., 0 1 2.", - ) - parser.add_argument( - "--coreset_ratio", - type=float, - default=None, - help="PatchCore memory-bank fraction to retain (0, 1].", - ) - parser.add_argument( - "--max_memory_patches", - type=int, - default=None, - help="Maximum PatchCore memory-bank size; omit for no cap.", - ) - parser.add_argument( - "--patch_grid", - type=int, - default=None, - help="PatchCore pooled grid size; use a smaller value for lower latency.", - ) - parser.add_argument( - "--search_chunk_size", - type=int, - default=None, - help="PatchCore query chunk size used to bound nearest-neighbor memory.", - ) - parser.add_argument( - "--coreset_method", - type=str, - choices=["kcenter", "random"], - default=None, - help="PatchCore coreset selection strategy; kcenter is the diverse default.", - ) - parser.add_argument( - "--coreset_seed", - type=int, - default=None, - help="Seed used for deterministic PatchCore coreset selection.", - ) - parser.add_argument( - "--output_model", - type=str, - default=None, - help="Filename to save the PT model.", - ) - parser.add_argument( - "--run_name", - type=str, - default=None, - help="Experiment name for this training run.", - ) - parser.add_argument( - "--model_data_path", - type=str, - default=None, - help="Directory to save model distributions and PT file.", - ) - parser.add_argument( - "--algorithm", - type=str, - default=None, - help="Algorithm name (e.g., padim, patchcore).", - ) - parser.add_argument( - "--log_level", - type=str, - choices=["DEBUG", "INFO", "WARNING", "ERROR"], - default=None, - help="Logging level (default: INFO).", - ) - + description="Train anomaly detector (args OR config).", add_help=add_help + ) + parser.add_argument("--config", type=str, default="config.yml", help="Path to config.yml/.json") + parser.add_argument("--dataset_path", type=str, default=None, help='Path to the dataset folder containing "train/good" images.') + parser.add_argument("--resize", type=int, nargs="*", default=None, metavar=("W", "H"), help="Resize before processing.") + parser.add_argument("--crop_size", type=int, nargs="*", default=None, metavar=("W", "H"), help="Apply a center crop.") + parser.add_argument("--normalize", action="store_true", default=None, help="Enable input normalization.") + parser.add_argument("--no_normalize", action="store_true", default=None, help="Disable input normalization explicitly.") + parser.add_argument("--norm_mean", type=float, nargs=3, default=None, metavar=("R", "G", "B"), help="Per-channel RGB mean.") + parser.add_argument("--norm_std", type=float, nargs=3, default=None, metavar=("R", "G", "B"), help="Per-channel RGB standard deviation.") + parser.add_argument("--backbone", type=str, choices=["resnet18", "wide_resnet50"], default=None, help="Backbone network.") + parser.add_argument("--batch_size", type=int, default=None, help="Batch size.") + parser.add_argument("--feat_dim", type=int, default=None, help="Number of PaDiM feature dimensions to keep.") + parser.add_argument("--layer_indices", type=int, nargs="+", default=None, help="ResNet feature layer indices.") + parser.add_argument("--coreset_ratio", type=float, default=None, help="PatchCore memory-bank fraction.") + parser.add_argument("--max_memory_patches", type=int, default=None, help="Maximum PatchCore memory-bank size.") + parser.add_argument("--patch_grid", type=int, default=None, help="PatchCore pooled grid size.") + parser.add_argument("--search_chunk_size", type=int, default=None, help="PatchCore query chunk size.") + parser.add_argument("--coreset_method", type=str, choices=["kcenter", "random"], default=None, help="PatchCore coreset strategy.") + parser.add_argument("--coreset_seed", type=int, default=None, help="PatchCore coreset seed.") + parser.add_argument("--output_model", type=str, default=None, help="Filename to save the PT model.") + parser.add_argument("--run_name", type=str, default=None, help="Experiment name.") + parser.add_argument("--model_data_path", type=str, default=None, help="Directory to save model distributions and PT file.") + parser.add_argument("--algorithm", type=str, default=None, help="Algorithm name (padim, patchcore, efficientad).") + parser.add_argument("--log_level", type=str, choices=["DEBUG", "INFO", "WARNING", "ERROR"], default=None, help="Logging level.") return parser def run_training(args): - """ - Executes the training pipeline. - - Args: - args: Namespace object containing command line arguments. - - Returns: - padim (AnomaVision.Padim): The trained model object. - config (edict): The final merged configuration. - run_dir (Path): The directory where artifacts were saved. - dataloaders (dict): Dictionary containing the 'train' DataLoader. - """ cfg = load_config(args.config) - - # Merge config with CLI args config = edict(merge_config(args, cfg)) - setup_logging(enabled=True, log_level=config.log_level, log_to_file=True) - logger = get_logger("anomavision.train") # Force it into anomavision hierarchy + logger = get_logger("anomavision.train") if not config.dataset_path: error_msg = "dataset.path is required (via --dataset_path or config.common.dataset_path)" @@ -203,7 +57,6 @@ def run_training(args): raise ValueError(error_msg) t0 = time.perf_counter() - logger.info( "Image processing: resize=%s, crop_size=%s, normalize=%s", config.resize, @@ -213,33 +66,17 @@ def run_training(args): if config.normalize: logger.info("Normalization: mean=%s, std=%s", config.norm_mean, config.norm_std) - # Resolve output run dir once run_dir = increment_path( - Path(config.model_data_path) - / config.algorithm - / config.class_name - / config.run_name, + Path(config.model_data_path) / config.algorithm / config.class_name / config.run_name, exist_ok=True, mkdir=True, ) - # === Dataset === - # Handle the 'class_name' logic safely. - # If dataset_path ends with the class name, use parent? - # Original logic assumes dataset_path is the container of class folders OR the class folder itself? - # Original code: os.path.join(realpath(dataset_path), config.class_name, "train", "good") - # This implies dataset_path is the root (e.g. MVTec root) and config.class_name is "bottle" - root = os.path.join( os.path.realpath(config.dataset_path), config.class_name, "train", "good" ) - if not os.path.isdir(root): - # Fallback check: maybe dataset_path ALREADY points to the class folder? - # This makes it more robust for different input styles - potential_root = os.path.join( - os.path.realpath(config.dataset_path), "train", "good" - ) + potential_root = os.path.join(os.path.realpath(config.dataset_path), "train", "good") if os.path.isdir(potential_root): root = potential_root else: @@ -254,7 +91,6 @@ def run_training(args): mean=config.norm_mean, std=config.norm_std, ) - if len(ds) == 0: error_msg = f"No training images found in {root}" logger.error(error_msg) @@ -263,13 +99,8 @@ def run_training(args): dl = DataLoader(ds, batch_size=int(config.batch_size), shuffle=False) logger.info("dataset: %d images | batch_size=%d", len(ds), config.batch_size) - # === Device === device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - logger.info( - "device: %s (cuda_available=%s)", device.type, torch.cuda.is_available() - ) - - # === Model & Train === + logger.info("device: %s (cuda_available=%s)", device.type, torch.cuda.is_available()) logger.info( "cfg: algorithm=%s | backbone=%s | layers=%s", config.algorithm, @@ -277,7 +108,8 @@ def run_training(args): config.layer_indices, ) - if str(config.algorithm).lower() == "patchcore": + algorithm = str(config.algorithm).lower() + if algorithm == "patchcore": model = anomavision.PatchCore( backbone=config.backbone, device=device, @@ -289,6 +121,14 @@ def run_training(args): coreset_method=config.get("coreset_method", "kcenter"), coreset_seed=int(config.get("coreset_seed", 42)), ) + elif algorithm == "efficientad": + model = anomavision.EfficientAD( + backbone=config.backbone, + device=device, + layer_indices=config.get("layer_indices", [0]), + epochs=int(config.get("efficientad_epochs", 5)), + learning_rate=float(config.get("efficientad_learning_rate", 1e-3)), + ) else: model = anomavision.Padim( backbone=config.backbone, @@ -301,11 +141,9 @@ def run_training(args): model.fit(dl) logger.info("fit: completed in %.2fs", time.perf_counter() - t_fit) - # === Save === model_path = Path(run_dir) / config.output_model torch.save(model, str(model_path)) - # Save a compact statistics/memory-bank artifact for deployment. stats_path = model_path.with_suffix(".pth") try: model.save_statistics(str(stats_path), half=True) @@ -313,13 +151,9 @@ def run_training(args): except Exception as e: logger.warning("saving slim statistics failed: %s", e) - # snapshot the effective configuration save_args_to_yaml(config, str(Path(run_dir) / "config.yml")) - logger.info("saved: model=%s, config=%s", model_path, Path(run_dir) / "config.yml") logger.info("=== Training done in %.2fs ===", time.perf_counter() - t0) - - # Return objects for external usage (e.g. MLOps pipeline) return model, config, run_dir, {"train": dl} @@ -327,17 +161,13 @@ def main(args=None): try: if args is None: args = create_parser().parse_args() - - # Optional git check — silently skipped if not in a repo or no network try: checker = GitStatusChecker() if checker.is_repo(): checker.check_status() except Exception: - pass # Never block training over a git check - + pass run_training(args) - except Exception: get_logger(__name__).exception("Fatal error during training.") sys.exit(1) From fccb5cb7afb7a4da0b8edc4b851585ed82deeafb Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:25:15 +0200 Subject: [PATCH 05/24] Keep existing training flow unchanged --- anomavision/train.py | 244 +++++++++++++++++++++++++++++++++++++------ 1 file changed, 211 insertions(+), 33 deletions(-) diff --git a/anomavision/train.py b/anomavision/train.py index 3494928..9db2d85 100644 --- a/anomavision/train.py +++ b/anomavision/train.py @@ -17,39 +17,185 @@ def create_parser(add_help: bool = True) -> argparse.ArgumentParser: parser = argparse.ArgumentParser( - description="Train anomaly detector (args OR config).", add_help=add_help - ) - parser.add_argument("--config", type=str, default="config.yml", help="Path to config.yml/.json") - parser.add_argument("--dataset_path", type=str, default=None, help='Path to the dataset folder containing "train/good" images.') - parser.add_argument("--resize", type=int, nargs="*", default=None, metavar=("W", "H"), help="Resize before processing.") - parser.add_argument("--crop_size", type=int, nargs="*", default=None, metavar=("W", "H"), help="Apply a center crop.") - parser.add_argument("--normalize", action="store_true", default=None, help="Enable input normalization.") - parser.add_argument("--no_normalize", action="store_true", default=None, help="Disable input normalization explicitly.") - parser.add_argument("--norm_mean", type=float, nargs=3, default=None, metavar=("R", "G", "B"), help="Per-channel RGB mean.") - parser.add_argument("--norm_std", type=float, nargs=3, default=None, metavar=("R", "G", "B"), help="Per-channel RGB standard deviation.") - parser.add_argument("--backbone", type=str, choices=["resnet18", "wide_resnet50"], default=None, help="Backbone network.") - parser.add_argument("--batch_size", type=int, default=None, help="Batch size.") - parser.add_argument("--feat_dim", type=int, default=None, help="Number of PaDiM feature dimensions to keep.") - parser.add_argument("--layer_indices", type=int, nargs="+", default=None, help="ResNet feature layer indices.") - parser.add_argument("--coreset_ratio", type=float, default=None, help="PatchCore memory-bank fraction.") - parser.add_argument("--max_memory_patches", type=int, default=None, help="Maximum PatchCore memory-bank size.") - parser.add_argument("--patch_grid", type=int, default=None, help="PatchCore pooled grid size.") - parser.add_argument("--search_chunk_size", type=int, default=None, help="PatchCore query chunk size.") - parser.add_argument("--coreset_method", type=str, choices=["kcenter", "random"], default=None, help="PatchCore coreset strategy.") - parser.add_argument("--coreset_seed", type=int, default=None, help="PatchCore coreset seed.") - parser.add_argument("--output_model", type=str, default=None, help="Filename to save the PT model.") - parser.add_argument("--run_name", type=str, default=None, help="Experiment name.") - parser.add_argument("--model_data_path", type=str, default=None, help="Directory to save model distributions and PT file.") - parser.add_argument("--algorithm", type=str, default=None, help="Algorithm name (padim, patchcore, efficientad).") - parser.add_argument("--log_level", type=str, choices=["DEBUG", "INFO", "WARNING", "ERROR"], default=None, help="Logging level.") + description="Train PaDiM (args OR config).", add_help=add_help + ) + # meta + parser.add_argument( + "--config", type=str, default="config.yml", help="Path to config.yml/.json" + ) + # dataset + parser.add_argument( + "--dataset_path", + type=str, + default=None, + help='Path to the dataset folder containing "train/good" images.', + ) + + # preprocessing + parser.add_argument( + "--resize", + type=int, + nargs="*", + default=None, + metavar=("W", "H"), + help="Resize before processing. Provide one value for a square resize (e.g., 256) or two values for width and height (e.g., 256 192). Omit to keep original size.", + ) + parser.add_argument( + "--crop_size", + type=int, + nargs="*", + default=None, + metavar=("W", "H"), + help="Apply a center (or configured) crop. One value for a square crop (e.g., 224) or two values for width and height (e.g., 224 224). Omit to disable cropping.", + ) + parser.add_argument( + "--normalize", + action="store_true", + default=None, + help="Enable input normalization. If set, inputs are normalized using --norm_mean/--norm_std if provided (commonly ImageNet stats).", + ) + parser.add_argument( + "--no_normalize", + action="store_true", + default=None, + help="Disable input normalization explicitly. If both --normalize and --no_normalize are set, this flag should take precedence in your code.", + ) + parser.add_argument( + "--norm_mean", + type=float, + nargs=3, + default=None, + metavar=("R", "G", "B"), + help="Per-channel RGB mean used when normalization is enabled. Example: 0.485 0.456 0.406.", + ) + parser.add_argument( + "--norm_std", + type=float, + nargs=3, + default=None, + metavar=("R", "G", "B"), + help="Per-channel RGB standard deviation used when normalization is enabled. Example: 0.229 0.224 0.225.", + ) + + # train + parser.add_argument( + "--backbone", + type=str, + choices=["resnet18", "wide_resnet50"], + default=None, + help="Backbone network to use for feature extraction.", + ) + parser.add_argument( + "--batch_size", + type=int, + default=None, + help="Batch size used during training and inference.", + ) + parser.add_argument( + "--feat_dim", + type=int, + default=None, + help="Number of random feature dimensions to keep.", + ) + parser.add_argument( + "--layer_indices", + type=int, + nargs="+", + default=None, + help="List of layer indices to extract features from, e.g., 0 1 2.", + ) + parser.add_argument( + "--coreset_ratio", + type=float, + default=None, + help="PatchCore memory-bank fraction to retain (0, 1].", + ) + parser.add_argument( + "--max_memory_patches", + type=int, + default=None, + help="Maximum PatchCore memory-bank size; omit for no cap.", + ) + parser.add_argument( + "--patch_grid", + type=int, + default=None, + help="PatchCore pooled grid size; use a smaller value for lower latency.", + ) + parser.add_argument( + "--search_chunk_size", + type=int, + default=None, + help="PatchCore query chunk size used to bound nearest-neighbor memory.", + ) + parser.add_argument( + "--coreset_method", + type=str, + choices=["kcenter", "random"], + default=None, + help="PatchCore coreset selection strategy; kcenter is the diverse default.", + ) + parser.add_argument( + "--coreset_seed", + type=int, + default=None, + help="Seed used for deterministic PatchCore coreset selection.", + ) + parser.add_argument( + "--output_model", + type=str, + default=None, + help="Filename to save the PT model.", + ) + parser.add_argument( + "--run_name", + type=str, + default=None, + help="Experiment name for this training run.", + ) + parser.add_argument( + "--model_data_path", + type=str, + default=None, + help="Directory to save model distributions and PT file.", + ) + parser.add_argument( + "--algorithm", + type=str, + default=None, + help="Algorithm name (e.g., padim, patchcore).", + ) + parser.add_argument( + "--log_level", + type=str, + choices=["DEBUG", "INFO", "WARNING", "ERROR"], + default=None, + help="Logging level (default: INFO).", + ) + return parser def run_training(args): + """ + Executes the training pipeline. + + Args: + args: Namespace object containing command line arguments. + + Returns: + padim (AnomaVision.Padim): The trained model object. + config (edict): The final merged configuration. + run_dir (Path): The directory where artifacts were saved. + dataloaders (dict): Dictionary containing the 'train' DataLoader. + """ cfg = load_config(args.config) + + # Merge config with CLI args config = edict(merge_config(args, cfg)) + setup_logging(enabled=True, log_level=config.log_level, log_to_file=True) - logger = get_logger("anomavision.train") + logger = get_logger("anomavision.train") # Force it into anomavision hierarchy if not config.dataset_path: error_msg = "dataset.path is required (via --dataset_path or config.common.dataset_path)" @@ -57,6 +203,7 @@ def run_training(args): raise ValueError(error_msg) t0 = time.perf_counter() + logger.info( "Image processing: resize=%s, crop_size=%s, normalize=%s", config.resize, @@ -66,17 +213,33 @@ def run_training(args): if config.normalize: logger.info("Normalization: mean=%s, std=%s", config.norm_mean, config.norm_std) + # Resolve output run dir once run_dir = increment_path( - Path(config.model_data_path) / config.algorithm / config.class_name / config.run_name, + Path(config.model_data_path) + / config.algorithm + / config.class_name + / config.run_name, exist_ok=True, mkdir=True, ) + # === Dataset === + # Handle the 'class_name' logic safely. + # If dataset_path ends with the class name, use parent? + # Original logic assumes dataset_path is the container of class folders OR the class folder itself? + # Original code: os.path.join(realpath(dataset_path), config.class_name, "train", "good") + # This implies dataset_path is the root (e.g. MVTec root) and config.class_name is "bottle" + root = os.path.join( os.path.realpath(config.dataset_path), config.class_name, "train", "good" ) + if not os.path.isdir(root): - potential_root = os.path.join(os.path.realpath(config.dataset_path), "train", "good") + # Fallback check: maybe dataset_path ALREADY points to the class folder? + # This makes it more robust for different input styles + potential_root = os.path.join( + os.path.realpath(config.dataset_path), "train", "good" + ) if os.path.isdir(potential_root): root = potential_root else: @@ -91,6 +254,7 @@ def run_training(args): mean=config.norm_mean, std=config.norm_std, ) + if len(ds) == 0: error_msg = f"No training images found in {root}" logger.error(error_msg) @@ -99,8 +263,13 @@ def run_training(args): dl = DataLoader(ds, batch_size=int(config.batch_size), shuffle=False) logger.info("dataset: %d images | batch_size=%d", len(ds), config.batch_size) + # === Device === device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - logger.info("device: %s (cuda_available=%s)", device.type, torch.cuda.is_available()) + logger.info( + "device: %s (cuda_available=%s)", device.type, torch.cuda.is_available() + ) + + # === Model & Train === logger.info( "cfg: algorithm=%s | backbone=%s | layers=%s", config.algorithm, @@ -108,8 +277,7 @@ def run_training(args): config.layer_indices, ) - algorithm = str(config.algorithm).lower() - if algorithm == "patchcore": + if str(config.algorithm).lower() == "patchcore": model = anomavision.PatchCore( backbone=config.backbone, device=device, @@ -121,7 +289,7 @@ def run_training(args): coreset_method=config.get("coreset_method", "kcenter"), coreset_seed=int(config.get("coreset_seed", 42)), ) - elif algorithm == "efficientad": + elif str(config.algorithm).lower() == "efficientad": model = anomavision.EfficientAD( backbone=config.backbone, device=device, @@ -141,9 +309,11 @@ def run_training(args): model.fit(dl) logger.info("fit: completed in %.2fs", time.perf_counter() - t_fit) + # === Save === model_path = Path(run_dir) / config.output_model torch.save(model, str(model_path)) + # Save a compact statistics/memory-bank artifact for deployment. stats_path = model_path.with_suffix(".pth") try: model.save_statistics(str(stats_path), half=True) @@ -151,9 +321,13 @@ def run_training(args): except Exception as e: logger.warning("saving slim statistics failed: %s", e) + # snapshot the effective configuration save_args_to_yaml(config, str(Path(run_dir) / "config.yml")) + logger.info("saved: model=%s, config=%s", model_path, Path(run_dir) / "config.yml") logger.info("=== Training done in %.2fs ===", time.perf_counter() - t0) + + # Return objects for external usage (e.g. MLOps pipeline) return model, config, run_dir, {"train": dl} @@ -161,13 +335,17 @@ def main(args=None): try: if args is None: args = create_parser().parse_args() + + # Optional git check — silently skipped if not in a repo or no network try: checker = GitStatusChecker() if checker.is_repo(): checker.check_status() except Exception: - pass + pass # Never block training over a git check + run_training(args) + except Exception: get_logger(__name__).exception("Fatal error during training.") sys.exit(1) From 380e7cdb6d4c6a636a344c9634bcdd28fa652a0e Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:25:35 +0200 Subject: [PATCH 06/24] Keep EfficientAD configuration drop-in compatible --- anomavision/algorithm/efficientad/efficientad.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/anomavision/algorithm/efficientad/efficientad.py b/anomavision/algorithm/efficientad/efficientad.py index c55bd09..223f904 100644 --- a/anomavision/algorithm/efficientad/efficientad.py +++ b/anomavision/algorithm/efficientad/efficientad.py @@ -48,13 +48,11 @@ def __init__( learning_rate: float = 1e-3, ) -> None: super().__init__() - if backbone not in {"resnet18"}: + if backbone != "resnet18": raise ValueError("EfficientAD lightweight supports backbone='resnet18' only.") self.device = torch.device(device) self.backbone = backbone - self.layer_indices = list(layer_indices or [0]) - if self.layer_indices != [0]: - raise ValueError("EfficientAD lightweight uses layer_indices=[0].") + self.layer_indices = [0] self.epochs = max(1, int(epochs)) self.learning_rate = float(learning_rate) @@ -80,8 +78,9 @@ def _loss(self, batch: torch.Tensor) -> torch.Tensor: @torch.no_grad() def _scores(self, batch: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]: + batch = batch.to(self.device, non_blocking=True) teacher = F.normalize(self._teacher_features(batch), dim=1) - student = F.normalize(self.student(batch.to(self.device)), dim=1) + student = F.normalize(self.student(batch), dim=1) score = (teacher - student).pow(2).mean(dim=1) image_scores = score.flatten(1).amax(1) score_map = F.interpolate( @@ -157,7 +156,7 @@ def build_efficientad_from_stats( model = EfficientAD( backbone=str(stats.get("backbone", "resnet18")), device=torch.device(device), - layer_indices=list(stats.get("layer_indices", [0])), + layer_indices=[0], epochs=int(stats.get("epochs", 5)), learning_rate=float(stats.get("learning_rate", 1e-3)), ) From cb309857c43872c7610f81bec6fb5ffd57927677 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:52:48 +0200 Subject: [PATCH 07/24] Improve EfficientAD training and adaptive threshold --- .../algorithm/efficientad/efficientad.py | 62 ++++++++++++------- 1 file changed, 40 insertions(+), 22 deletions(-) diff --git a/anomavision/algorithm/efficientad/efficientad.py b/anomavision/algorithm/efficientad/efficientad.py index 223f904..150fcbd 100644 --- a/anomavision/algorithm/efficientad/efficientad.py +++ b/anomavision/algorithm/efficientad/efficientad.py @@ -29,15 +29,7 @@ def forward(self, x: torch.Tensor) -> torch.Tensor: class EfficientAD(torch.nn.Module): - """Minimal teacher-student EfficientAD implementation. - - A frozen ImageNet ResNet teacher provides layer-1 features and a very small - CNN student learns those features from normal images. At inference, the - teacher-student feature error is used as the anomaly map. - - The public API mirrors PaDiM/PatchCore: ``fit`` trains on normal images and - ``predict`` returns image scores plus a spatial anomaly map. - """ + """Minimal teacher-student EfficientAD implementation.""" def __init__( self, @@ -46,6 +38,7 @@ def __init__( layer_indices: Optional[List[int]] = None, epochs: int = 5, learning_rate: float = 1e-3, + threshold_percentile: float = 99.5, ) -> None: super().__init__() if backbone != "resnet18": @@ -55,6 +48,10 @@ def __init__( self.layer_indices = [0] self.epochs = max(1, int(epochs)) self.learning_rate = float(learning_rate) + self.threshold_percentile = float(threshold_percentile) + if not 0.0 < self.threshold_percentile <= 100.0: + raise ValueError("threshold_percentile must be in (0, 100].") + self.threshold = 0.0 self.teacher = ResnetEmbeddingsExtractor(backbone, self.device) for parameter in self.teacher.parameters(): @@ -69,17 +66,17 @@ def _teacher_features(self, batch: torch.Tensor) -> torch.Tensor: features, width, height = self.teacher(batch, layer_indices=[0]) return features.reshape(batch.shape[0], width, height, 64).permute(0, 3, 1, 2) - def _loss(self, batch: torch.Tensor) -> torch.Tensor: - with torch.no_grad(): - teacher = self._teacher_features(batch) - teacher = F.normalize(teacher, dim=1) + def _loss(self, teacher: torch.Tensor, batch: torch.Tensor) -> torch.Tensor: + teacher = F.normalize(teacher, dim=1) student = F.normalize(self.student(batch), dim=1) return F.mse_loss(student, teacher) @torch.no_grad() - def _scores(self, batch: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]: + def _scores(self, batch: torch.Tensor, teacher: Optional[torch.Tensor] = None) -> Tuple[torch.Tensor, torch.Tensor]: batch = batch.to(self.device, non_blocking=True) - teacher = F.normalize(self._teacher_features(batch), dim=1) + if teacher is None: + teacher = self._teacher_features(batch) + teacher = F.normalize(teacher, dim=1) student = F.normalize(self.student(batch), dim=1) score = (teacher - student).pow(2).mean(dim=1) image_scores = score.flatten(1).amax(1) @@ -92,25 +89,44 @@ def _scores(self, batch: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]: return image_scores, score_map def fit(self, dataloader: torch.utils.data.DataLoader, extractions: int = 1) -> None: - """Train the small student using only normal training images.""" + """Train the student and calibrate an adaptive threshold on normal images.""" optimizer = torch.optim.Adam(self.student.parameters(), lr=self.learning_rate) - self.student.train() - for _ in range(self.epochs): + + # Compute frozen teacher features once. This avoids running ResNet for + # every epoch and makes CPU training considerably faster. + cached = [] + self.teacher.eval() + with torch.no_grad(): for item in dataloader: batch = item[0] if isinstance(item, (tuple, list)) else item batch = batch.to(self.device, non_blocking=True) + cached.append((batch.detach(), self._teacher_features(batch).detach())) + + self.student.train() + for _ in range(self.epochs): + for batch, teacher in cached: optimizer.zero_grad(set_to_none=True) - loss = self._loss(batch) + loss = self._loss(teacher, batch) loss.backward() optimizer.step() + self.student.eval() + + # Calibrate from normal training scores. A high percentile keeps the + # false-positive rate low without requiring a manually chosen score. + normal_scores = [] + for batch, teacher in cached: + scores, _ = self._scores(batch, teacher) + normal_scores.append(scores.cpu()) + self.threshold = float( + torch.quantile(torch.cat(normal_scores), self.threshold_percentile / 100.0) + ) self._fitted = True @torch.no_grad() def forward( self, x: torch.Tensor, return_map: bool = True, export: bool = False ) -> Tuple[torch.Tensor, Optional[torch.Tensor]]: - """Return image-level anomaly scores and an optional spatial map.""" if not self._fitted: raise RuntimeError("EfficientAD is not fitted. Call fit() first.") image_scores, score_map = self._scores(x) @@ -119,7 +135,6 @@ def forward( def predict( self, batch: torch.Tensor, export: bool = False ) -> Tuple[torch.Tensor, torch.Tensor]: - """Run inference using the standard AnomaVision prediction contract.""" return self.forward(batch, return_map=True, export=export) def to_device(self, device: torch.device) -> None: @@ -128,7 +143,6 @@ def to_device(self, device: torch.device) -> None: self.student.to(self.device) def save_statistics(self, path: str, half: Optional[bool] = False) -> None: - """Save a compact student/teacher deployment artifact.""" if not self._fitted: raise RuntimeError("EfficientAD is not fitted. Call fit() first.") student_state = { @@ -142,6 +156,8 @@ def save_statistics(self, path: str, half: Optional[bool] = False) -> None: "layer_indices": self.layer_indices, "epochs": self.epochs, "learning_rate": self.learning_rate, + "threshold_percentile": self.threshold_percentile, + "threshold": self.threshold, "model_type": "efficientad", "dtype": "fp16" if half else "fp32", }, @@ -159,9 +175,11 @@ def build_efficientad_from_stats( layer_indices=[0], epochs=int(stats.get("epochs", 5)), learning_rate=float(stats.get("learning_rate", 1e-3)), + threshold_percentile=float(stats.get("threshold_percentile", 99.5)), ) state = stats["student"] model.student.load_state_dict({key: value.float() for key, value in state.items()}) model.student.eval() + model.threshold = float(stats.get("threshold", 0.0)) model._fitted = True return model From 3448fc2fc1e5d07f9de9eaadffa1b5bf3814293a Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:52:55 +0200 Subject: [PATCH 08/24] Add EfficientAD adaptive threshold config --- config.yml | 190 ++++++++++++++++++++++++++--------------------------- 1 file changed, 92 insertions(+), 98 deletions(-) diff --git a/config.yml b/config.yml index 0af6c53..2601967 100644 --- a/config.yml +++ b/config.yml @@ -1,124 +1,118 @@ # ========================= # Dataset / preprocessing (shared by train, detect, eval, stream) # ========================= -dataset_path: "D:/01-DATA" # Root dataset folder (MVTec-style: contains train/test subfolders) -class_name: "bottle" # Class name for MVTec dataset -resize: [224, 224] # Resize dimensions before processing [width, height] -crop_size: # Final crop size [width, height] -normalize: true # Whether to normalize images -no_normalize: false # Inverse flag (used by CLI) – keep in sync with "normalize" -norm_mean: [0.485, 0.456, 0.406] # Mean values for normalization (ImageNet default) -norm_std: [0.229, 0.224, 0.225] # Standard deviation for normalization (ImageNet default) +dataset_path: "D:/01-DATA" +class_name: "bottle" +resize: [224, 224] +crop_size: +normalize: true +no_normalize: false +norm_mean: [0.485, 0.456, 0.406] +norm_std: [0.229, 0.224, 0.225] # ========================= # Model / training # ========================= -backbone: "resnet18" # Backbone CNN architecture (resnet18 | wide_resnet50) -algorithm: "padim" # Algorithm to use: padim or patchcore -coreset_ratio: 0.02 # Ultra-light PatchCore memory fraction -max_memory_patches: 2048 # Hard memory-bank cap for low latency -patch_grid: 14 # Pool feature maps to at most 14x14 patches -search_chunk_size: 1024 # Bound nearest-neighbor working memory -coreset_method: "kcenter" # PatchCore selection: kcenter or random -coreset_seed: 42 # Reproducible k-center initialization -feat_dim: 50 # Feature dimension size for embedding -layer_indices: [0] # Which backbone layers to extract features from (0,1,2,3) -model_data_path: "./distributions" # Path to store/load model-related data -model: "model.pt" # File name for saved model (used by detect/eval/export) -output_model: "model.pt" # File name for saving trained model (train.py expects this) -batch_size: 2 # Training/evaluation/inference batch size -device: "auto" # Device to run on: "cpu", "cuda", or "auto" +backbone: "resnet18" +algorithm: "efficientad" +coreset_ratio: 0.02 +max_memory_patches: 2048 +patch_grid: 14 +search_chunk_size: 1024 +coreset_method: "kcenter" +coreset_seed: 42 +feat_dim: 50 +layer_indices: [0] +model_data_path: "./distributions" +model: "model.pt" +output_model: "model.pt" +batch_size: 2 +device: "auto" + +# Lightweight EfficientAD +# Threshold is calibrated automatically from normal training images. +efficientad_epochs: 5 +efficientad_learning_rate: 0.001 +efficientad_threshold_percentile: 99.5 # ========================= # Logging / run metadata # ========================= -log_level: "INFO" # Logging level: DEBUG, INFO, WARNING, ERROR -run_name: "anomav_exp" # Name of experiment run (used for organizing results) -detailed_timing: false # Enable detailed timing measurements +log_level: "INFO" +run_name: "anomav_exp" +detailed_timing: false # ========================= -# Visualization (shared by detect & eval) +# Visualization # ========================= -enable_visualization: true # Enable visualization during inference/evaluation -save_visualizations: true # Save visualization results to disk -viz_output_dir: "./visualizations/" # Directory to save visualization results -viz_alpha: 0.5 # Transparency factor for overlay heatmaps -viz_padding: 40 # Padding added around visualization -viz_color: "128,0,128" # RGB color for visualization overlays +enable_visualization: true +save_visualizations: true +viz_output_dir: "./visualizations/" +viz_alpha: 0.5 +viz_padding: 40 +viz_color: "128,0,128" # ========================= -# Inference (detect.py) +# Inference # ========================= -img_path: "D:/01-DATA/test" # Path to test images for inference -thresh: null # Legacy fallback; prefer algorithm-specific thresholds -thresh_padim: 13.0 # PaDiM score threshold; null lets eval auto-select -thresh_patchcore: 0.25 # PatchCore score threshold; null lets eval auto-select -num_workers: 1 # Number of workers for dataloader -pin_memory: false # Use pinned memory for faster GPU transfers -overwrite: false # Overwrite existing run directory without auto-incrementing +img_path: "D:/01-DATA/test" +thresh: null +thresh_padim: 13.0 +thresh_patchcore: 0.25 +thresh_efficientad: null +num_workers: 1 +pin_memory: false +overwrite: false # ========================= -# Evaluation (eval.py) +# Evaluation # ========================= -memory_efficient: true # Use memory efficient evaluation mode +memory_efficient: true # ========================= -# Export (export.py) -# ========================= -format: "all" # onnx, tensorrt, torchscript, openvino, all -opset: 18 # ONNX opset version -precision: "auto" # ONNX/TorchScript precision: auto, fp16, fp32 -tensorrt_precision: "fp16" # TensorRT precision: fp32, fp16, int8 -dynamic_batch: true # Allow dynamic batch size in exported model -static_batch: false # Disable dynamic batch size (if true) -min_batch: 1 # TensorRT dynamic profile minimum batch -opt_batch: 1 # TensorRT dynamic profile optimal batch -max_batch: 4 # TensorRT dynamic profile maximum batch -workspace_gb: 2.0 # TensorRT workspace limit in GiB -calib_dir: null # INT8 calibration image directory; auto-derived when null -calib_samples: 100 # Maximum real images used for INT8 calibration -quantize_dynamic: false # Also write dynamically quantized INT8 ONNX -quantize_static: false # Also write statically quantized INT8 ONNX -optimize: false # Enable mobile optimization for TorchScript -output_path: null # Optional explicit output filename -half: false # Legacy compatibility field -int8: false # Legacy compatibility field; use tensorrt_precision: int8 +# Export +# ========================= +format: "all" +opset: 18 +precision: "auto" +tensorrt_precision: "fp16" +dynamic_batch: true +static_batch: false +min_batch: 1 +opt_batch: 1 +max_batch: 4 +workspace_gb: 2.0 +calib_dir: null +calib_samples: 100 +quantize_dynamic: false +quantize_static: false +optimize: false +output_path: null +half: false +int8: false # ========================= -# Streaming Configuration +# Streaming # ========================= -stream_mode: false # true = real-time streaming, false = static dataset - +stream_mode: false stream_source: - type: "webcam" # webcam | video | mqtt | tcp - - # Webcam settings (type: webcam) - camera_id: 0 # Camera device index - - # Video file settings (type: video) - video_path: "path/to/video.mp4" # Path to video file - loop: false # Loop video when it ends - - # MQTT settings (type: mqtt) - broker: "localhost" # MQTT broker hostname/IP - port: 1883 # MQTT broker port - topic: "camera/frames" # Topic to subscribe to - client_id: null # Optional client ID - keepalive: 60 # Keepalive interval (seconds) - qos: 0 # QoS level (0, 1, or 2) - max_queue_size: 10 # Max buffered frames - read_timeout: 1.0 # Timeout for reading frames (seconds) - - # TCP settings (type: tcp) - host: "192.168.1.100" # TCP server hostname/IP - port: 8080 # TCP server port (matches code) - recv_timeout: 1.0 # Socket timeout for recv (seconds) - header_size: 4 # Length header size in bytes - max_message_size: 10485760 # Max payload size (10MB) - - -# Streaming processing settings -stream_max_frames: null # Max frames to process (null = infinite) -stream_display_fps: true # Show FPS during streaming -stream_save_detections: true # Save detected anomalies to disk -stream_detection_dir: "./stream_detections/" # Directory for saved detections + type: "webcam" + camera_id: 0 + video_path: "path/to/video.mp4" + loop: false + broker: "localhost" + port: 1883 + topic: "camera/frames" + client_id: null + keepalive: 60 + qos: 0 + max_queue_size: 10 + read_timeout: 1.0 + host: "192.168.1.100" + recv_timeout: 1.0 + header_size: 4 + max_message_size: 10485760 +stream_max_frames: null +stream_display_fps: true +stream_save_detections: true +stream_detection_dir: "./stream_detections/" From 9fec0318e8a9b1c172acaf8089c7a0073fb6db98 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:54:59 +0200 Subject: [PATCH 09/24] Document all config parameters --- config.yml | 179 +++++++++++++++++++++++++++-------------------------- 1 file changed, 90 insertions(+), 89 deletions(-) diff --git a/config.yml b/config.yml index 2601967..fcc3363 100644 --- a/config.yml +++ b/config.yml @@ -1,118 +1,119 @@ # ========================= # Dataset / preprocessing (shared by train, detect, eval, stream) # ========================= -dataset_path: "D:/01-DATA" -class_name: "bottle" -resize: [224, 224] -crop_size: -normalize: true -no_normalize: false -norm_mean: [0.485, 0.456, 0.406] -norm_std: [0.229, 0.224, 0.225] +dataset_path: "D:/01-DATA" # Root dataset directory (contains train/test folders) +class_name: "bottle" # Dataset class to train/evaluate +aresize: [224, 224] # Input image size [width, height] +crop_size: # Optional center crop size [width, height] +normalize: true # Apply ImageNet normalization +no_normalize: false # Explicitly disable normalization when true +norm_mean: [0.485, 0.456, 0.406] # Normalization mean for RGB channels +norm_std: [0.229, 0.224, 0.225] # Normalization standard deviation for RGB channels # ========================= # Model / training # ========================= -backbone: "resnet18" -algorithm: "efficientad" -coreset_ratio: 0.02 -max_memory_patches: 2048 -patch_grid: 14 -search_chunk_size: 1024 -coreset_method: "kcenter" -coreset_seed: 42 -feat_dim: 50 -layer_indices: [0] -model_data_path: "./distributions" -model: "model.pt" -output_model: "model.pt" -batch_size: 2 -device: "auto" +backbone: "resnet18" # Backbone network: resnet18 | wide_resnet50 +algorithm: "efficientad" # Anomaly algorithm: padim | patchcore | efficientad +coreset_ratio: 0.02 # PatchCore fraction of patches kept in memory +max_memory_patches: 2048 # Maximum number of PatchCore memory-bank patches +patch_grid: 14 # PatchCore feature-map grid size +search_chunk_size: 1024 # PatchCore nearest-neighbor search chunk size +coreset_method: "kcenter" # PatchCore coreset method: kcenter | random +coreset_seed: 42 # Random seed for PatchCore coreset selection +feat_dim: 50 # PaDiM number of feature dimensions to keep +layer_indices: [0] # Backbone feature layers used by PaDiM/PatchCore/EfficientAD +model_data_path: "./distributions" # Directory for trained models and statistics +model: "model.pt" # Model filename used by detect/eval/export +output_model: "model.pt" # Filename written by the train command +batch_size: 2 # Training/inference batch size +device: "auto" # Device: auto | cpu | cuda +# ========================= # Lightweight EfficientAD -# Threshold is calibrated automatically from normal training images. -efficientad_epochs: 5 -efficientad_learning_rate: 0.001 -efficientad_threshold_percentile: 99.5 +# ========================= +efficientad_epochs: 5 # Number of student training epochs +efficientad_learning_rate: 0.001 # Student optimizer learning rate +efficientad_threshold_percentile: 99.5 # Percentile of normal scores used for adaptive threshold # ========================= # Logging / run metadata # ========================= -log_level: "INFO" -run_name: "anomav_exp" -detailed_timing: false +log_level: "INFO" # Logging level: DEBUG | INFO | WARNING | ERROR +run_name: "anomav_exp" # Experiment name used in output directories +detailed_timing: false # Enable detailed timing information # ========================= -# Visualization +# Visualization (detect/eval) # ========================= -enable_visualization: true -save_visualizations: true -viz_output_dir: "./visualizations/" -viz_alpha: 0.5 -viz_padding: 40 -viz_color: "128,0,128" +enable_visualization: true # Enable anomaly visualization +save_visualizations: true # Save generated visualizations to disk +viz_output_dir: "./visualizations/" # Directory for visualization output +viz_alpha: 0.5 # Heatmap overlay transparency +viz_padding: 40 # Extra visualization border/padding +viz_color: "128,0,128" # RGB color used for anomaly overlays # ========================= -# Inference +# Inference (detect) # ========================= -img_path: "D:/01-DATA/test" -thresh: null -thresh_padim: 13.0 -thresh_patchcore: 0.25 -thresh_efficientad: null -num_workers: 1 -pin_memory: false -overwrite: false +img_path: "D:/01-DATA/test" # Input image or directory for detection +thresh: null # Generic legacy threshold fallback +thresh_padim: 13.0 # Fixed PaDiM anomaly threshold +thresh_patchcore: 0.25 # Fixed PatchCore anomaly threshold +thresh_efficientad: null # EfficientAD threshold; null uses the trained adaptive threshold +num_workers: 1 # Number of data-loader worker processes +pin_memory: false # Pin CPU memory for faster GPU transfers +overwrite: false # Overwrite an existing output directory # ========================= -# Evaluation +# Evaluation (eval) # ========================= -memory_efficient: true +memory_efficient: true # Reduce memory usage during evaluation # ========================= -# Export -# ========================= -format: "all" -opset: 18 -precision: "auto" -tensorrt_precision: "fp16" -dynamic_batch: true -static_batch: false -min_batch: 1 -opt_batch: 1 -max_batch: 4 -workspace_gb: 2.0 -calib_dir: null -calib_samples: 100 -quantize_dynamic: false -quantize_static: false -optimize: false -output_path: null -half: false -int8: false +# Export (export) +# ========================= +format: "all" # Export format: onnx | tensorrt | torchscript | openvino | all +opset: 18 # ONNX operator-set version +precision: "auto" # Export precision: auto | fp16 | fp32 +tensorrt_precision: "fp16" # TensorRT precision: fp32 | fp16 | int8 +dynamic_batch: true # Allow variable batch size in exported models +static_batch: false # Force a fixed batch size when true +min_batch: 1 # Minimum TensorRT dynamic batch size +opt_batch: 1 # Optimal TensorRT dynamic batch size +max_batch: 4 # Maximum TensorRT dynamic batch size +workspace_gb: 2.0 # TensorRT workspace memory limit in GiB +calib_dir: null # Directory containing INT8 calibration images +calib_samples: 100 # Maximum number of calibration images +quantize_dynamic: false # Also generate dynamically quantized INT8 ONNX +quantize_static: false # Also generate statically quantized INT8 ONNX +optimize: false # Enable TorchScript/mobile optimization +output_path: null # Optional explicit export output path +half: false # Legacy FP16 compatibility option +int8: false # Legacy INT8 compatibility option # ========================= # Streaming # ========================= -stream_mode: false +stream_mode: false # Enable real-time streaming mode stream_source: - type: "webcam" - camera_id: 0 - video_path: "path/to/video.mp4" - loop: false - broker: "localhost" - port: 1883 - topic: "camera/frames" - client_id: null - keepalive: 60 - qos: 0 - max_queue_size: 10 - read_timeout: 1.0 - host: "192.168.1.100" - recv_timeout: 1.0 - header_size: 4 - max_message_size: 10485760 -stream_max_frames: null -stream_display_fps: true -stream_save_detections: true -stream_detection_dir: "./stream_detections/" + type: "webcam" # Source type: webcam | video | mqtt | tcp + camera_id: 0 # Webcam device index + video_path: "path/to/video.mp4" # Video file path + loop: false # Restart video when it reaches the end + broker: "localhost" # MQTT broker hostname/IP + port: 1883 # MQTT broker port + topic: "camera/frames" # MQTT topic containing frames + client_id: null # Optional MQTT client ID + keepalive: 60 # MQTT keepalive interval in seconds + qos: 0 # MQTT quality of service: 0 | 1 | 2 + max_queue_size: 10 # Maximum buffered streaming frames + read_timeout: 1.0 # Streaming read timeout in seconds + host: "192.168.1.100" # TCP server hostname/IP + recv_timeout: 1.0 # TCP receive timeout in seconds + header_size: 4 # TCP message length-header size in bytes + max_message_size: 10485760 # Maximum TCP message size in bytes +stream_max_frames: null # Maximum frames to process; null means unlimited +stream_display_fps: true # Display FPS during streaming +stream_save_detections: true # Save streaming anomaly detections +stream_detection_dir: "./stream_detections/" # Output directory for streaming detections From 592aafa17798687e45a05d0e08544cd72eb5a7c2 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 23:05:06 +0200 Subject: [PATCH 10/24] Pass EfficientAD threshold calibration settings --- anomavision/train.py | 292 +++++++------------------------------------ 1 file changed, 45 insertions(+), 247 deletions(-) diff --git a/anomavision/train.py b/anomavision/train.py index 9db2d85..8a8a8a5 100644 --- a/anomavision/train.py +++ b/anomavision/train.py @@ -19,183 +19,41 @@ def create_parser(add_help: bool = True) -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Train PaDiM (args OR config).", add_help=add_help ) - # meta - parser.add_argument( - "--config", type=str, default="config.yml", help="Path to config.yml/.json" - ) - # dataset - parser.add_argument( - "--dataset_path", - type=str, - default=None, - help='Path to the dataset folder containing "train/good" images.', - ) - - # preprocessing - parser.add_argument( - "--resize", - type=int, - nargs="*", - default=None, - metavar=("W", "H"), - help="Resize before processing. Provide one value for a square resize (e.g., 256) or two values for width and height (e.g., 256 192). Omit to keep original size.", - ) - parser.add_argument( - "--crop_size", - type=int, - nargs="*", - default=None, - metavar=("W", "H"), - help="Apply a center (or configured) crop. One value for a square crop (e.g., 224) or two values for width and height (e.g., 224 224). Omit to disable cropping.", - ) - parser.add_argument( - "--normalize", - action="store_true", - default=None, - help="Enable input normalization. If set, inputs are normalized using --norm_mean/--norm_std if provided (commonly ImageNet stats).", - ) - parser.add_argument( - "--no_normalize", - action="store_true", - default=None, - help="Disable input normalization explicitly. If both --normalize and --no_normalize are set, this flag should take precedence in your code.", - ) - parser.add_argument( - "--norm_mean", - type=float, - nargs=3, - default=None, - metavar=("R", "G", "B"), - help="Per-channel RGB mean used when normalization is enabled. Example: 0.485 0.456 0.406.", - ) - parser.add_argument( - "--norm_std", - type=float, - nargs=3, - default=None, - metavar=("R", "G", "B"), - help="Per-channel RGB standard deviation used when normalization is enabled. Example: 0.229 0.224 0.225.", - ) - - # train - parser.add_argument( - "--backbone", - type=str, - choices=["resnet18", "wide_resnet50"], - default=None, - help="Backbone network to use for feature extraction.", - ) - parser.add_argument( - "--batch_size", - type=int, - default=None, - help="Batch size used during training and inference.", - ) - parser.add_argument( - "--feat_dim", - type=int, - default=None, - help="Number of random feature dimensions to keep.", - ) - parser.add_argument( - "--layer_indices", - type=int, - nargs="+", - default=None, - help="List of layer indices to extract features from, e.g., 0 1 2.", - ) - parser.add_argument( - "--coreset_ratio", - type=float, - default=None, - help="PatchCore memory-bank fraction to retain (0, 1].", - ) - parser.add_argument( - "--max_memory_patches", - type=int, - default=None, - help="Maximum PatchCore memory-bank size; omit for no cap.", - ) - parser.add_argument( - "--patch_grid", - type=int, - default=None, - help="PatchCore pooled grid size; use a smaller value for lower latency.", - ) - parser.add_argument( - "--search_chunk_size", - type=int, - default=None, - help="PatchCore query chunk size used to bound nearest-neighbor memory.", - ) - parser.add_argument( - "--coreset_method", - type=str, - choices=["kcenter", "random"], - default=None, - help="PatchCore coreset selection strategy; kcenter is the diverse default.", - ) - parser.add_argument( - "--coreset_seed", - type=int, - default=None, - help="Seed used for deterministic PatchCore coreset selection.", - ) - parser.add_argument( - "--output_model", - type=str, - default=None, - help="Filename to save the PT model.", - ) - parser.add_argument( - "--run_name", - type=str, - default=None, - help="Experiment name for this training run.", - ) - parser.add_argument( - "--model_data_path", - type=str, - default=None, - help="Directory to save model distributions and PT file.", - ) - parser.add_argument( - "--algorithm", - type=str, - default=None, - help="Algorithm name (e.g., padim, patchcore).", - ) - parser.add_argument( - "--log_level", - type=str, - choices=["DEBUG", "INFO", "WARNING", "ERROR"], - default=None, - help="Logging level (default: INFO).", - ) - + parser.add_argument("--config", type=str, default="config.yml", help="Path to config.yml/.json") + parser.add_argument("--dataset_path", type=str, default=None, help='Path to the dataset folder containing "train/good" images.') + parser.add_argument("--resize", type=int, nargs="*", default=None, metavar=("W", "H"), help="Resize before processing.") + parser.add_argument("--crop_size", type=int, nargs="*", default=None, metavar=("W", "H"), help="Apply a center crop.") + parser.add_argument("--normalize", action="store_true", default=None, help="Enable input normalization.") + parser.add_argument("--no_normalize", action="store_true", default=None, help="Disable input normalization.") + parser.add_argument("--norm_mean", type=float, nargs=3, default=None, metavar=("R", "G", "B"), help="Per-channel RGB mean.") + parser.add_argument("--norm_std", type=float, nargs=3, default=None, metavar=("R", "G", "B"), help="Per-channel RGB standard deviation.") + parser.add_argument("--backbone", type=str, choices=["resnet18", "wide_resnet50"], default=None, help="Backbone network.") + parser.add_argument("--batch_size", type=int, default=None, help="Batch size.") + parser.add_argument("--feat_dim", type=int, default=None, help="Number of PaDiM feature dimensions to keep.") + parser.add_argument("--layer_indices", type=int, nargs="+", default=None, help="Backbone feature layers.") + parser.add_argument("--coreset_ratio", type=float, default=None, help="PatchCore memory-bank fraction.") + parser.add_argument("--max_memory_patches", type=int, default=None, help="Maximum PatchCore memory-bank size.") + parser.add_argument("--patch_grid", type=int, default=None, help="PatchCore pooled grid size.") + parser.add_argument("--search_chunk_size", type=int, default=None, help="PatchCore query chunk size.") + parser.add_argument("--coreset_method", type=str, choices=["kcenter", "random"], default=None, help="PatchCore coreset method.") + parser.add_argument("--coreset_seed", type=int, default=None, help="PatchCore coreset seed.") + parser.add_argument("--output_model", type=str, default=None, help="Filename to save the PT model.") + parser.add_argument("--run_name", type=str, default=None, help="Experiment name.") + parser.add_argument("--model_data_path", type=str, default=None, help="Model output directory.") + parser.add_argument("--algorithm", type=str, default=None, help="Algorithm name.") + parser.add_argument("--log_level", type=str, choices=["DEBUG", "INFO", "WARNING", "ERROR"], default=None, help="Logging level.") + parser.add_argument("--efficientad_epochs", type=int, default=None, help="EfficientAD student training epochs.") + parser.add_argument("--efficientad_learning_rate", type=float, default=None, help="EfficientAD student learning rate.") + parser.add_argument("--efficientad_threshold_percentile", type=float, default=None, help="Percentile of normal EfficientAD scores used for the adaptive threshold.") return parser def run_training(args): - """ - Executes the training pipeline. - - Args: - args: Namespace object containing command line arguments. - - Returns: - padim (AnomaVision.Padim): The trained model object. - config (edict): The final merged configuration. - run_dir (Path): The directory where artifacts were saved. - dataloaders (dict): Dictionary containing the 'train' DataLoader. - """ cfg = load_config(args.config) - - # Merge config with CLI args config = edict(merge_config(args, cfg)) setup_logging(enabled=True, log_level=config.log_level, log_to_file=True) - logger = get_logger("anomavision.train") # Force it into anomavision hierarchy + logger = get_logger("anomavision.train") if not config.dataset_path: error_msg = "dataset.path is required (via --dataset_path or config.common.dataset_path)" @@ -203,43 +61,18 @@ def run_training(args): raise ValueError(error_msg) t0 = time.perf_counter() - - logger.info( - "Image processing: resize=%s, crop_size=%s, normalize=%s", - config.resize, - config.crop_size, - config.normalize, - ) + logger.info("Image processing: resize=%s, crop_size=%s, normalize=%s", config.resize, config.crop_size, config.normalize) if config.normalize: logger.info("Normalization: mean=%s, std=%s", config.norm_mean, config.norm_std) - # Resolve output run dir once run_dir = increment_path( - Path(config.model_data_path) - / config.algorithm - / config.class_name - / config.run_name, - exist_ok=True, - mkdir=True, - ) - - # === Dataset === - # Handle the 'class_name' logic safely. - # If dataset_path ends with the class name, use parent? - # Original logic assumes dataset_path is the container of class folders OR the class folder itself? - # Original code: os.path.join(realpath(dataset_path), config.class_name, "train", "good") - # This implies dataset_path is the root (e.g. MVTec root) and config.class_name is "bottle" - - root = os.path.join( - os.path.realpath(config.dataset_path), config.class_name, "train", "good" + Path(config.model_data_path) / config.algorithm / config.class_name / config.run_name, + exist_ok=True, mkdir=True, ) + root = os.path.join(os.path.realpath(config.dataset_path), config.class_name, "train", "good") if not os.path.isdir(root): - # Fallback check: maybe dataset_path ALREADY points to the class folder? - # This makes it more robust for different input styles - potential_root = os.path.join( - os.path.realpath(config.dataset_path), "train", "good" - ) + potential_root = os.path.join(os.path.realpath(config.dataset_path), "train", "good") if os.path.isdir(potential_root): root = potential_root else: @@ -247,61 +80,36 @@ def run_training(args): raise FileNotFoundError(f"Dataset root not found: {root}") ds = anomavision.AnodetDataset( - root, - resize=config.resize, - crop_size=config.crop_size, - normalize=config.normalize, - mean=config.norm_mean, - std=config.norm_std, + root, resize=config.resize, crop_size=config.crop_size, + normalize=config.normalize, mean=config.norm_mean, std=config.norm_std, ) - if len(ds) == 0: - error_msg = f"No training images found in {root}" - logger.error(error_msg) - raise ValueError(error_msg) + raise ValueError(f"No training images found in {root}") dl = DataLoader(ds, batch_size=int(config.batch_size), shuffle=False) logger.info("dataset: %d images | batch_size=%d", len(ds), config.batch_size) - # === Device === device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - logger.info( - "device: %s (cuda_available=%s)", device.type, torch.cuda.is_available() - ) - - # === Model & Train === - logger.info( - "cfg: algorithm=%s | backbone=%s | layers=%s", - config.algorithm, - config.backbone, - config.layer_indices, - ) + logger.info("device: %s (cuda_available=%s)", device.type, torch.cuda.is_available()) + logger.info("cfg: algorithm=%s | backbone=%s | layers=%s", config.algorithm, config.backbone, config.layer_indices) if str(config.algorithm).lower() == "patchcore": model = anomavision.PatchCore( - backbone=config.backbone, - device=device, - layer_indices=config.layer_indices, - coreset_ratio=float(config.coreset_ratio), - max_memory_patches=config.max_memory_patches, - patch_grid=config.patch_grid, - search_chunk_size=config.search_chunk_size, - coreset_method=config.get("coreset_method", "kcenter"), - coreset_seed=int(config.get("coreset_seed", 42)), + backbone=config.backbone, device=device, layer_indices=config.layer_indices, + coreset_ratio=float(config.coreset_ratio), max_memory_patches=config.max_memory_patches, + patch_grid=config.patch_grid, search_chunk_size=config.search_chunk_size, + coreset_method=config.get("coreset_method", "kcenter"), coreset_seed=int(config.get("coreset_seed", 42)), ) elif str(config.algorithm).lower() == "efficientad": model = anomavision.EfficientAD( - backbone=config.backbone, - device=device, - layer_indices=config.get("layer_indices", [0]), + backbone=config.backbone, device=device, layer_indices=config.get("layer_indices", [0]), epochs=int(config.get("efficientad_epochs", 5)), learning_rate=float(config.get("efficientad_learning_rate", 1e-3)), + threshold_percentile=float(config.get("efficientad_threshold_percentile", 99.5)), ) else: model = anomavision.Padim( - backbone=config.backbone, - device=device, - layer_indices=config.layer_indices, + backbone=config.backbone, device=device, layer_indices=config.layer_indices, feat_dim=int(config.feat_dim), ) @@ -309,11 +117,9 @@ def run_training(args): model.fit(dl) logger.info("fit: completed in %.2fs", time.perf_counter() - t_fit) - # === Save === model_path = Path(run_dir) / config.output_model torch.save(model, str(model_path)) - # Save a compact statistics/memory-bank artifact for deployment. stats_path = model_path.with_suffix(".pth") try: model.save_statistics(str(stats_path), half=True) @@ -321,13 +127,9 @@ def run_training(args): except Exception as e: logger.warning("saving slim statistics failed: %s", e) - # snapshot the effective configuration save_args_to_yaml(config, str(Path(run_dir) / "config.yml")) - logger.info("saved: model=%s, config=%s", model_path, Path(run_dir) / "config.yml") logger.info("=== Training done in %.2fs ===", time.perf_counter() - t0) - - # Return objects for external usage (e.g. MLOps pipeline) return model, config, run_dir, {"train": dl} @@ -335,17 +137,13 @@ def main(args=None): try: if args is None: args = create_parser().parse_args() - - # Optional git check — silently skipped if not in a repo or no network try: checker = GitStatusChecker() if checker.is_repo(): checker.check_status() except Exception: - pass # Never block training over a git check - + pass run_training(args) - except Exception: get_logger(__name__).exception("Fatal error during training.") sys.exit(1) From ad59d1fd861eee47427d836ec9f9e58a3eee2ab3 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 23:05:23 +0200 Subject: [PATCH 11/24] Make EfficientAD scoring robust and calibrate threshold --- anomavision/algorithm/efficientad/efficientad.py | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/anomavision/algorithm/efficientad/efficientad.py b/anomavision/algorithm/efficientad/efficientad.py index 150fcbd..a467783 100644 --- a/anomavision/algorithm/efficientad/efficientad.py +++ b/anomavision/algorithm/efficientad/efficientad.py @@ -72,14 +72,23 @@ def _loss(self, teacher: torch.Tensor, batch: torch.Tensor) -> torch.Tensor: return F.mse_loss(student, teacher) @torch.no_grad() - def _scores(self, batch: torch.Tensor, teacher: Optional[torch.Tensor] = None) -> Tuple[torch.Tensor, torch.Tensor]: + def _scores( + self, batch: torch.Tensor, teacher: Optional[torch.Tensor] = None + ) -> Tuple[torch.Tensor, torch.Tensor]: batch = batch.to(self.device, non_blocking=True) if teacher is None: teacher = self._teacher_features(batch) teacher = F.normalize(teacher, dim=1) student = F.normalize(self.student(batch), dim=1) score = (teacher - student).pow(2).mean(dim=1) - image_scores = score.flatten(1).amax(1) + + # Use the mean of the highest-scoring 1% of locations instead of a + # single maximum. This keeps small defects sensitive while reducing + # false positives caused by one noisy feature location. + flat = score.flatten(1) + k = max(1, int(flat.shape[1] * 0.01)) + image_scores = flat.topk(k, dim=1).values.mean(dim=1) + score_map = F.interpolate( score.unsqueeze(1), size=batch.shape[-2:], @@ -112,8 +121,7 @@ def fit(self, dataloader: torch.utils.data.DataLoader, extractions: int = 1) -> self.student.eval() - # Calibrate from normal training scores. A high percentile keeps the - # false-positive rate low without requiring a manually chosen score. + # Calibrate using the same image-score calculation used at inference. normal_scores = [] for batch, teacher in cached: scores, _ = self._scores(batch, teacher) From 7e3739798965ed8a332d44b0ede7035e40f6c618 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 23:05:50 +0200 Subject: [PATCH 12/24] Embed adaptive EfficientAD threshold in model outputs --- .../algorithm/efficientad/efficientad.py | 23 ++++++++++++------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/anomavision/algorithm/efficientad/efficientad.py b/anomavision/algorithm/efficientad/efficientad.py index a467783..6efc638 100644 --- a/anomavision/algorithm/efficientad/efficientad.py +++ b/anomavision/algorithm/efficientad/efficientad.py @@ -72,7 +72,7 @@ def _loss(self, teacher: torch.Tensor, batch: torch.Tensor) -> torch.Tensor: return F.mse_loss(student, teacher) @torch.no_grad() - def _scores( + def _raw_scores( self, batch: torch.Tensor, teacher: Optional[torch.Tensor] = None ) -> Tuple[torch.Tensor, torch.Tensor]: batch = batch.to(self.device, non_blocking=True) @@ -82,9 +82,8 @@ def _scores( student = F.normalize(self.student(batch), dim=1) score = (teacher - student).pow(2).mean(dim=1) - # Use the mean of the highest-scoring 1% of locations instead of a - # single maximum. This keeps small defects sensitive while reducing - # false positives caused by one noisy feature location. + # Mean of the highest-scoring 1% of locations is more robust than a + # single maximum while remaining sensitive to small defects. flat = score.flatten(1) k = max(1, int(flat.shape[1] * 0.01)) image_scores = flat.topk(k, dim=1).values.mean(dim=1) @@ -121,14 +120,16 @@ def fit(self, dataloader: torch.utils.data.DataLoader, extractions: int = 1) -> self.student.eval() - # Calibrate using the same image-score calculation used at inference. + # Calibrate from normal training scores using the exact score that is + # used for inference. normal_scores = [] for batch, teacher in cached: - scores, _ = self._scores(batch, teacher) + scores, _ = self._raw_scores(batch, teacher) normal_scores.append(scores.cpu()) self.threshold = float( torch.quantile(torch.cat(normal_scores), self.threshold_percentile / 100.0) ) + self.threshold = max(self.threshold, 1e-8) self._fitted = True @torch.no_grad() @@ -137,7 +138,13 @@ def forward( ) -> Tuple[torch.Tensor, Optional[torch.Tensor]]: if not self._fitted: raise RuntimeError("EfficientAD is not fitted. Call fit() first.") - image_scores, score_map = self._scores(x) + image_scores, score_map = self._raw_scores(x) + + # Normalize scores by the training-derived threshold. This embeds the + # adaptive threshold into the exported model: >= 1.0 means anomalous. + scale = image_scores.new_tensor(self.threshold) + image_scores = image_scores / scale + score_map = score_map / scale return image_scores, score_map if return_map else None def predict( @@ -188,6 +195,6 @@ def build_efficientad_from_stats( state = stats["student"] model.student.load_state_dict({key: value.float() for key, value in state.items()}) model.student.eval() - model.threshold = float(stats.get("threshold", 0.0)) + model.threshold = max(float(stats.get("threshold", 0.0)), 1e-8) model._fitted = True return model From 154d2494529ad523549aa541d2d7f17be8674f97 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 23:06:05 +0200 Subject: [PATCH 13/24] Fix EfficientAD config threshold and resize key --- config.yml | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/config.yml b/config.yml index fcc3363..f5f4591 100644 --- a/config.yml +++ b/config.yml @@ -3,7 +3,7 @@ # ========================= dataset_path: "D:/01-DATA" # Root dataset directory (contains train/test folders) class_name: "bottle" # Dataset class to train/evaluate -aresize: [224, 224] # Input image size [width, height] +resize: [224, 224] # Input image size [width, height] crop_size: # Optional center crop size [width, height] normalize: true # Apply ImageNet normalization no_normalize: false # Explicitly disable normalization when true @@ -14,7 +14,7 @@ norm_std: [0.229, 0.224, 0.225] # Normalization standard deviation fo # Model / training # ========================= backbone: "resnet18" # Backbone network: resnet18 | wide_resnet50 -algorithm: "efficientad" # Anomaly algorithm: padim | patchcore | efficientad +algorithm: "efficientad" # Anomaly algorithm: padim | patchcore | efficientad coreset_ratio: 0.02 # PatchCore fraction of patches kept in memory max_memory_patches: 2048 # Maximum number of PatchCore memory-bank patches patch_grid: 14 # PatchCore feature-map grid size @@ -23,18 +23,18 @@ coreset_method: "kcenter" # PatchCore coreset method: kcenter | coreset_seed: 42 # Random seed for PatchCore coreset selection feat_dim: 50 # PaDiM number of feature dimensions to keep layer_indices: [0] # Backbone feature layers used by PaDiM/PatchCore/EfficientAD -model_data_path: "./distributions" # Directory for trained models and statistics +model_data_path: "./distributions" # Directory for trained models and statistics model: "model.pt" # Model filename used by detect/eval/export output_model: "model.pt" # Filename written by the train command batch_size: 2 # Training/inference batch size -device: "auto" # Device: auto | cpu | cuda +device: "auto" # Device to run on: auto | cpu | cuda # ========================= # Lightweight EfficientAD # ========================= efficientad_epochs: 5 # Number of student training epochs efficientad_learning_rate: 0.001 # Student optimizer learning rate -efficientad_threshold_percentile: 99.5 # Percentile of normal scores used for adaptive threshold +efficientad_threshold_percentile: 99.5 # Percentile of normal scores used to learn the adaptive threshold # ========================= # Logging / run metadata @@ -60,7 +60,7 @@ img_path: "D:/01-DATA/test" # Input image or directory for detect thresh: null # Generic legacy threshold fallback thresh_padim: 13.0 # Fixed PaDiM anomaly threshold thresh_patchcore: 0.25 # Fixed PatchCore anomaly threshold -thresh_efficientad: null # EfficientAD threshold; null uses the trained adaptive threshold +thresh_efficientad: 1.0 # EfficientAD threshold; 1.0 means the model's adaptive training threshold num_workers: 1 # Number of data-loader worker processes pin_memory: false # Pin CPU memory for faster GPU transfers overwrite: false # Overwrite an existing output directory From 27082988fa08268faf6220c1722d0eae8fcd939f Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 23:06:17 +0200 Subject: [PATCH 14/24] Set EfficientAD to evaluation mode after training --- anomavision/algorithm/efficientad/efficientad.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/anomavision/algorithm/efficientad/efficientad.py b/anomavision/algorithm/efficientad/efficientad.py index 6efc638..d3e0bd4 100644 --- a/anomavision/algorithm/efficientad/efficientad.py +++ b/anomavision/algorithm/efficientad/efficientad.py @@ -131,6 +131,7 @@ def fit(self, dataloader: torch.utils.data.DataLoader, extractions: int = 1) -> ) self.threshold = max(self.threshold, 1e-8) self._fitted = True + self.eval() @torch.no_grad() def forward( @@ -197,4 +198,5 @@ def build_efficientad_from_stats( model.student.eval() model.threshold = max(float(stats.get("threshold", 0.0)), 1e-8) model._fitted = True + model.eval() return model From 1b7b2e8e6083edc7f8dc5c68ad30e254fa26b4cb Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 23:32:22 +0200 Subject: [PATCH 15/24] test: add configurable inference performance benchmarks --- config.yml | 22 +++++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/config.yml b/config.yml index f5f4591..519ea63 100644 --- a/config.yml +++ b/config.yml @@ -36,6 +36,26 @@ efficientad_epochs: 5 # Number of student training epochs efficientad_learning_rate: 0.001 # Student optimizer learning rate efficientad_threshold_percentile: 99.5 # Percentile of normal scores used to learn the adaptive threshold +# ========================= +# Inference performance tests +# ========================= +# Opt-in benchmark used to catch inference regressions for every algorithm. +# Run with: pytest tests/test_inference_performance.py -s +inference_benchmark: + enabled: false # Set true to run model benchmarks + warmup_runs: 10 # Warm-up predictions excluded from timing + test_runs: 50 # Timed predictions per algorithm + batch_size: 2 # Batch size used by the benchmark + max_inference_ms: # Maximum allowed pure inference time per batch + padim: 100.0 + patchcore: 100.0 + efficientad: 100.0 + max_regression_percent: 25.0 # Allowed regression if a baseline is configured + baseline_inference_ms: # Optional per-algorithm baseline; null disables regression check + padim: null + patchcore: null + efficientad: 47.15 + # ========================= # Logging / run metadata # ========================= @@ -103,7 +123,7 @@ stream_source: loop: false # Restart video when it reaches the end broker: "localhost" # MQTT broker hostname/IP port: 1883 # MQTT broker port - topic: "camera/frames" # MQTT topic containing frames + topic: "camera/frames" # MQTT topic containing frames client_id: null # Optional MQTT client ID keepalive: 60 # MQTT keepalive interval in seconds qos: 0 # MQTT quality of service: 0 | 1 | 2 From b5fe70f8af3c3672b998ea59c5d02a10538fda46 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sat, 29 Aug 2026 23:32:33 +0200 Subject: [PATCH 16/24] test: add per-algorithm inference regression benchmarks --- tests/test_inference_performance.py | 139 ++++++++++++++++++++++++++++ 1 file changed, 139 insertions(+) create mode 100644 tests/test_inference_performance.py diff --git a/tests/test_inference_performance.py b/tests/test_inference_performance.py new file mode 100644 index 0000000..ee64915 --- /dev/null +++ b/tests/test_inference_performance.py @@ -0,0 +1,139 @@ +"""Config-driven pure inference performance regression tests. + +These tests intentionally measure only ``model.predict(batch)``. Image loading, +preprocessing, postprocessing, visualization, and result accumulation are kept +outside the timed section so the numbers match the ``Pure inference FPS`` and +``Average inference time`` reported by ``anomavision.detect``. + +Enable them in ``config.yml`` with ``inference_benchmark.enabled: true``. +""" + +from __future__ import annotations + +import time +from pathlib import Path + +import pytest +import torch +from torch.utils.data import DataLoader + +import anomavision +from anomavision.config import load_config +from anomavision.general import determine_device +from anomavision.inference.model.wrapper import ModelWrapper + + +CONFIG_PATH = Path(__file__).resolve().parents[1] / "config.yml" +ALGORITHMS = ("padim", "patchcore", "efficientad") + + +def _cuda_sync(device: str) -> None: + """Synchronize CUDA before/after a timed inference call.""" + if device == "cuda" and torch.cuda.is_available(): + torch.cuda.synchronize() + + +def _load_benchmark_config() -> dict: + config = load_config(str(CONFIG_PATH)) + benchmark = config.get("inference_benchmark", {}) + if not benchmark.get("enabled", False): + pytest.skip( + "Inference benchmark disabled. Set inference_benchmark.enabled: true in config.yml." + ) + return config + + +@pytest.mark.parametrize("algorithm", ALGORITHMS) +def test_inference_performance(algorithm: str) -> None: + """Benchmark PaDiM, PatchCore, and EfficientAD independently.""" + config = _load_benchmark_config() + benchmark = config["inference_benchmark"] + + device = determine_device(str(config.get("device", "auto"))) + model_path = ( + Path(config["model_data_path"]) + / algorithm + / config["class_name"] + / config["run_name"] + / config["model"] + ) + dataset_path = Path(config["img_path"]) + + if not model_path.exists(): + pytest.skip(f"{algorithm}: model not found: {model_path}") + if not dataset_path.exists(): + pytest.skip(f"{algorithm}: dataset not found: {dataset_path}") + + batch_size = int(benchmark.get("batch_size", config.get("batch_size", 1))) + warmup_runs = int(benchmark.get("warmup_runs", 10)) + test_runs = int(benchmark.get("test_runs", 50)) + + if batch_size < 1 or warmup_runs < 0 or test_runs < 1: + raise ValueError( + "inference_benchmark.batch_size >= 1, warmup_runs >= 0, test_runs >= 1" + ) + + dataset = anomavision.AnodetDataset( + str(dataset_path), + resize=config.get("resize"), + crop_size=config.get("crop_size"), + normalize=config.get("normalize", True), + mean=config.get("norm_mean"), + std=config.get("norm_std"), + ) + dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=False) + + try: + first = next(iter(dataloader)) + except StopIteration: + pytest.fail(f"{algorithm}: benchmark dataset is empty: {dataset_path}") + + batch = first[0] + if device == "cuda": + batch = batch.half() + batch = batch.to(device) + + model = ModelWrapper(str(model_path), device) + try: + for _ in range(warmup_runs): + model.predict(batch) + _cuda_sync(device) + + timings_ms = [] + for _ in range(test_runs): + _cuda_sync(device) + start = time.perf_counter() + model.predict(batch) + _cuda_sync(device) + timings_ms.append((time.perf_counter() - start) * 1000.0) + + average_ms = sum(timings_ms) / len(timings_ms) + pure_fps = batch_size * 1000.0 / average_ms + + max_ms = benchmark.get("max_inference_ms", {}).get(algorithm) + baseline_ms = benchmark.get("baseline_inference_ms", {}).get(algorithm) + max_regression = float(benchmark.get("max_regression_percent", 25.0)) + + allowed_ms = None if max_ms is None else float(max_ms) + if baseline_ms is not None: + regression_limit = float(baseline_ms) * (1.0 + max_regression / 100.0) + allowed_ms = ( + regression_limit + if allowed_ms is None + else min(allowed_ms, regression_limit) + ) + + print( + f"\n{algorithm.upper()} | " + f"Pure inference FPS: {pure_fps:.2f} images/sec | " + f"Average inference time: {average_ms:.2f} ms/batch | " + f"Throughput: {pure_fps:.2f} images/sec (batch size: {batch_size})" + ) + + if allowed_ms is not None: + assert average_ms <= allowed_ms, ( + f"{algorithm} inference regression: {average_ms:.2f} ms/batch " + f"> allowed {allowed_ms:.2f} ms/batch." + ) + finally: + model.close() From d52923c274944c815c7f3d5a6ca79aa49d550a04 Mon Sep 17 00:00:00 2001 From: DeepKnowledge1 Date: Sun, 30 Aug 2026 19:23:44 +0200 Subject: [PATCH 17/24] performance --- config.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/config.yml b/config.yml index 519ea63..937d4f9 100644 --- a/config.yml +++ b/config.yml @@ -42,14 +42,14 @@ efficientad_threshold_percentile: 99.5 # Percentile of normal scores used t # Opt-in benchmark used to catch inference regressions for every algorithm. # Run with: pytest tests/test_inference_performance.py -s inference_benchmark: - enabled: false # Set true to run model benchmarks + enabled: true # Set true to run model benchmarks warmup_runs: 10 # Warm-up predictions excluded from timing test_runs: 50 # Timed predictions per algorithm batch_size: 2 # Batch size used by the benchmark max_inference_ms: # Maximum allowed pure inference time per batch padim: 100.0 patchcore: 100.0 - efficientad: 100.0 + efficientad: 150.0 max_regression_percent: 25.0 # Allowed regression if a baseline is configured baseline_inference_ms: # Optional per-algorithm baseline; null disables regression check padim: null From b6b43977bd8c0bc643bd759eb34b5aa0c80e14b8 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sun, 30 Aug 2026 19:28:09 +0200 Subject: [PATCH 18/24] Add EfficientAD to production autopilot --- anomavision/autopilot.py | 319 +++++------------- .../inference/model/backends/torch_backend.py | 175 ++-------- 2 files changed, 120 insertions(+), 374 deletions(-) diff --git a/anomavision/autopilot.py b/anomavision/autopilot.py index 9ac4ad2..7e4ef8b 100644 --- a/anomavision/autopilot.py +++ b/anomavision/autopilot.py @@ -10,7 +10,7 @@ import time from html import escape from pathlib import Path -from typing import Any, Dict, Iterable, Optional +from typing import Any, Dict, Optional import numpy as np import torch @@ -21,52 +21,30 @@ from anomavision.config import load_config from anomavision.general import determine_device from anomavision.inference.model.wrapper import ModelWrapper -from anomavision.utils import ( - compute_metrics, - find_optimal_threshold, - make_localization_mask, -) +from anomavision.utils import compute_metrics, find_optimal_threshold, make_localization_mask + + +ALGORITHMS = ("padim", "patchcore", "efficientad") def create_parser(add_help: bool = True) -> argparse.ArgumentParser: - """Build the ``anomavision autopilot`` argument parser.""" parser = argparse.ArgumentParser( description="Select, calibrate, profile, and package a production anomaly model.", add_help=add_help, ) - parser.add_argument( - "--config", type=str, required=True, help="Base AnomaVision config file." - ) - parser.add_argument( - "--dataset_path", type=str, default=None, help="MVTec-style dataset root." - ) - parser.add_argument( - "--class_name", type=str, default=None, help="Dataset class to evaluate." - ) - parser.add_argument( - "--padim_model", - type=str, - default=None, - help="PaDiM model artifact (.pt/.pth/.onnx).", - ) - parser.add_argument( - "--patchcore_model", - type=str, - default=None, - help="PatchCore model artifact (.pt/.pth/.onnx).", - ) + parser.add_argument("--config", type=str, required=True, help="Base AnomaVision config file.") + parser.add_argument("--dataset_path", type=str, default=None, help="MVTec-style dataset root.") + parser.add_argument("--class_name", type=str, default=None, help="Dataset class to evaluate.") + parser.add_argument("--padim_model", type=str, default=None, help="PaDiM model artifact (.pt/.pth/.onnx).") + parser.add_argument("--patchcore_model", type=str, default=None, help="PatchCore model artifact (.pt/.pth/.onnx).") + parser.add_argument("--efficientad_model", type=str, default=None, help="EfficientAD model artifact (.pt/.pth/.onnx).") parser.add_argument("--device", choices=["auto", "cpu", "cuda"], default="auto") parser.add_argument("--batch_size", type=int, default=1) parser.add_argument("--num_workers", type=int, default=0) parser.add_argument("--warmup", type=int, default=3) parser.add_argument("--timing_batches", type=int, default=20) parser.add_argument("--target_latency_ms", type=float, default=None) - parser.add_argument( - "--validation_split", - type=float, - default=1.0, - help="Fraction of the complete labeled test split used for calibration; 1.0 uses every sample.", - ) + parser.add_argument("--validation_split", type=float, default=1.0) parser.add_argument("--output_dir", type=str, default="./production_package") parser.add_argument("--copy_config", action="store_true", default=True) return parser @@ -86,17 +64,10 @@ def _format_percent(value: Any) -> str: return "N/A" if value is None else f"{float(value):.1%}" -def _profile_model( - model_path: str, - dataloader: DataLoader, - device: str, - warmup: int, - timing_batches: int, -) -> Dict[str, Any]: +def _profile_model(model_path: str, dataloader: DataLoader, device: str, warmup: int, timing_batches: int) -> Dict[str, Any]: wrapper = ModelWrapper(model_path, device) - iterator = iter(dataloader) try: - first = next(iterator) + first = next(iter(dataloader)) except StopIteration: wrapper.close() raise ValueError("The evaluation dataset is empty.") @@ -105,10 +76,10 @@ def _profile_model( wrapper.predict(first_batch) if device.startswith("cuda") and torch.cuda.is_available(): torch.cuda.synchronize() + timings = [] all_scores, all_maps, all_labels, all_masks = [], [], [], [] - count = 0 - for item in dataloader: + for count, item in enumerate(dataloader): batch, _, labels, masks = item batch = batch.to(device) measure = count < max(1, timing_batches) @@ -124,28 +95,23 @@ def _profile_model( all_maps.extend(_to_numpy(maps)) all_labels.extend(_to_numpy(labels).reshape(-1).tolist()) all_masks.extend(_to_numpy(masks)) - count += 1 wrapper.close() + scores_np = np.asarray(all_scores, dtype=np.float32) labels_np = np.asarray(all_labels, dtype=np.int64) - maps_np = ( - np.asarray(all_maps, dtype=np.float32) - if all_maps - else np.empty((0, 0, 0), dtype=np.float32) - ) + maps_np = np.asarray(all_maps, dtype=np.float32) if all_maps else np.empty((0, 0, 0), dtype=np.float32) threshold, threshold_f1 = ( find_optimal_threshold(labels_np, scores_np) if len(np.unique(labels_np)) > 1 else (float(np.median(scores_np)), 0.0) ) image_metrics = compute_metrics(labels_np, scores_np, thresh=threshold) - image_auroc = ( - image_metrics.get("auc_score") if len(np.unique(labels_np)) > 1 else None - ) - pixel_auroc = None + image_auroc = image_metrics.get("auc_score") if len(np.unique(labels_np)) > 1 else None + masks_np = np.asarray(all_masks, dtype=np.float32) if masks_np.ndim == 4 and masks_np.shape[1] == 1: masks_np = masks_np[:, 0] + pixel_auroc = None localization = { "available": bool(len(maps_np) == len(labels_np) and maps_np.ndim == 3), "non_empty_fraction": None, @@ -156,229 +122,138 @@ def _profile_model( "normal_mean_mask_area_fraction": None, "verdict": "unavailable", } - if ( - localization["available"] - and masks_np.shape == maps_np.shape - and np.unique(masks_np).size > 1 - ): + if localization["available"] and masks_np.shape == maps_np.shape and np.unique(masks_np).size > 1: try: - pixel_auroc = float( - roc_auc_score(masks_np.reshape(-1) > 0.5, maps_np.reshape(-1)) - ) + pixel_auroc = float(roc_auc_score(masks_np.reshape(-1) > 0.5, maps_np.reshape(-1))) except ValueError: - pixel_auroc = None + pass image_metrics["image_auroc"] = image_auroc image_metrics["pixel_auroc"] = pixel_auroc - anomaly_labels = (scores_np >= threshold).astype(np.uint8) + if localization["available"]: + anomaly_labels = (scores_np >= threshold).astype(np.uint8) loc_masks = make_localization_mask(maps_np, anomaly_labels).astype(bool) non_empty = loc_masks.reshape(len(loc_masks), -1).any(axis=1) area = loc_masks.reshape(len(loc_masks), -1).mean(axis=1) - anomaly_idx = labels_np == 1 - normal_idx = labels_np == 0 + anomaly_idx, normal_idx = labels_np == 1, labels_np == 0 localization["non_empty_fraction"] = float(non_empty.mean()) localization["mean_mask_area_fraction"] = float(area.mean()) - localization["anomaly_non_empty_fraction"] = ( - float(non_empty[anomaly_idx].mean()) if anomaly_idx.any() else None - ) - localization["normal_false_positive_fraction"] = ( - float(non_empty[normal_idx].mean()) if normal_idx.any() else None - ) - localization["anomaly_mean_mask_area_fraction"] = ( - float(area[anomaly_idx].mean()) if anomaly_idx.any() else None - ) - localization["normal_mean_mask_area_fraction"] = ( - float(area[normal_idx].mean()) if normal_idx.any() else None - ) - if pixel_auroc is not None: - localization["verdict"] = ( - "healthy" - if (localization["normal_false_positive_fraction"] or 0.0) <= 0.10 - else "review false positives" - ) - else: - localization["verdict"] = "maps available; pixel AUROC unavailable" - median_ms = ( - float(np.median(timings) * 1000 / max(1, dataloader.batch_size)) - if timings - else 0.0 - ) - p95_ms = ( - float(np.percentile(timings, 95) * 1000 / max(1, dataloader.batch_size)) - if timings - else 0.0 - ) + localization["anomaly_non_empty_fraction"] = float(non_empty[anomaly_idx].mean()) if anomaly_idx.any() else None + localization["normal_false_positive_fraction"] = float(non_empty[normal_idx].mean()) if normal_idx.any() else None + localization["anomaly_mean_mask_area_fraction"] = float(area[anomaly_idx].mean()) if anomaly_idx.any() else None + localization["normal_mean_mask_area_fraction"] = float(area[normal_idx].mean()) if normal_idx.any() else None + localization["verdict"] = ( + "healthy" + if (localization["normal_false_positive_fraction"] or 0.0) <= 0.10 + else "review false positives" + ) if pixel_auroc is not None else "maps available; pixel AUROC unavailable" + + batch_size = max(1, dataloader.batch_size or 1) + median_ms = float(np.median(timings) * 1000 / batch_size) if timings else 0.0 + p95_ms = float(np.percentile(timings, 95) * 1000 / batch_size) if timings else 0.0 return { "model_path": str(Path(model_path).resolve()), "model_format": Path(model_path).suffix.lower(), "threshold": float(threshold), "threshold_f1": float(threshold_f1), - "metrics": { - k: (float(v) if isinstance(v, (float, np.floating)) else v) - for k, v in image_metrics.items() - }, + "metrics": {k: float(v) if isinstance(v, (float, np.floating)) else v for k, v in image_metrics.items()}, "latency_ms": {"median": median_ms, "p95": p95_ms}, - "throughput_images_per_second": ( - float(1000.0 / median_ms) if median_ms > 0 else 0.0 - ), + "throughput_images_per_second": float(1000.0 / median_ms) if median_ms > 0 else 0.0, "localization": localization, "samples": int(len(labels_np)), } -def _select( - results: Dict[str, Dict[str, Any]], target_latency_ms: Optional[float] -) -> str: +def _select(results: Dict[str, Dict[str, Any]], target_latency_ms: Optional[float]) -> str: eligible = results if target_latency_ms is not None: - eligible = { - name: result - for name, result in results.items() - if result["latency_ms"]["p95"] <= target_latency_ms - } + eligible = {name: result for name, result in results.items() if result["latency_ms"]["p95"] <= target_latency_ms} if not eligible: eligible = results - return max( - eligible, - key=lambda name: ( - eligible[name]["metrics"].get("image_auroc") or 0.0, - -eligible[name]["latency_ms"]["p95"], - ), - ) + return max(eligible, key=lambda name: (eligible[name]["metrics"].get("image_auroc") or 0.0, -eligible[name]["latency_ms"]["p95"])) def _write_report(manifest: Dict[str, Any], output_dir: Path) -> None: - """Write a self-contained HTML dashboard and a Markdown fallback report.""" selected = manifest["selected_model"] - selected_result = manifest["candidates"][selected] - target = manifest.get("target_latency_ms") cards = [] rows = [] for name, result in manifest["candidates"].items(): - metrics = result["metrics"] - loc = result["localization"] - is_selected = name == selected - status = "Selected" if is_selected else "Candidate" - status_class = "selected" if is_selected else "candidate" + metrics, loc = result["metrics"], result["localization"] + active = name == selected cards.append( - f'
{escape(name.upper())}{status}
' - f'
{_format_metric(metrics.get("image_auroc"))} image AUROC
' - f'
{_format_metric(metrics.get("pixel_auroc"))}pixel AUROC
{result["latency_ms"]["p95"]:.1f} msp95 latency
{_format_percent(loc.get("anomaly_non_empty_fraction"))}anomaly coverage
' + f'
{escape(name.upper())}' + f'

{_format_metric(metrics.get("image_auroc"))}

image AUROC' + f'

p95: {result["latency_ms"]["p95"]:.2f} ms · pixel AUROC: {_format_metric(metrics.get("pixel_auroc"))}

' ) rows.append( - f'{escape(name)}{_format_metric(metrics.get("image_auroc"))}{_format_metric(metrics.get("pixel_auroc"))}{result["latency_ms"]["median"]:.2f}{result["latency_ms"]["p95"]:.2f}{_format_percent(loc.get("anomaly_non_empty_fraction"))}{_format_percent(loc.get("normal_false_positive_fraction"))}{result["threshold"]:.6f}' + f'{escape(name)}{_format_metric(metrics.get("image_auroc"))}' + f'{_format_metric(metrics.get("pixel_auroc"))}{result["latency_ms"]["median"]:.2f}' + f'{result["latency_ms"]["p95"]:.2f}{_format_percent(loc.get("normal_false_positive_fraction"))}' + f'{result["threshold"]:.6f}' ) - target_text = ( - f"under {target:.1f} ms p95" - if target is not None - else "with the strongest measured accuracy/latency balance" - ) - environment_json = escape(json.dumps(manifest["environment"], indent=2)) - html = f""" - -AnomaVision Production Autopilot -
-
AnomaVision / Production Autopilot

Deployment confidence, before production.

Calibrated thresholds, hardware-aware profiling, and localization health checks in one reproducible report.

Selected: {escape(selected)}Class: {escape(str(manifest["dataset"]["class_name"]))}Samples: {manifest["dataset"]["samples"]}Device: {escape(str(manifest["environment"].get("device", "unknown")))}
-
Recommendation
Deploy {escape(selected)} with threshold {selected_result["threshold"]:.6f}. It was selected {escape(target_text)}. Recheck this threshold on a production validation set before release.
-

Candidate overview

Measured on the same validation data
{"".join(cards)}
-

Detailed comparison

Higher AUROC and anomaly coverage are better; lower false positives and latency are better
{"".join(rows)}
ModelImage AUROCPixel AUROCMedian msP95 msAnomaly coverageNormal false positivesThreshold
-

Localization health

Selected model maps available{"PASS" if selected_result["localization"]["available"] else "CHECK"}
Anomaly images with localization{_format_percent(selected_result["localization"].get("anomaly_non_empty_fraction"))}
Normal images with false-positive maps{_format_percent(selected_result["localization"].get("normal_false_positive_fraction"))}
Anomaly mean mask area{_format_percent(selected_result["localization"].get("anomaly_mean_mask_area_fraction"))}
Localization verdict{escape(str(selected_result["localization"].get("verdict", "N/A")))}

Anomaly coverage measures detected defect images. Normal false positives should remain low.

Deployment artifact

Artifact{escape(str(manifest["selected_artifact"]))}
Format{escape(str(selected_result["model_format"]))}
Preprocessing{escape(str(manifest["preprocessing"].get("resize")))} px
Target latency{escape(str(target)) if target is not None else "not set"}
-

Reproducibility environment

{environment_json}
Generated by AnomaVision Production Autopilot · manifest schema {manifest["schema_version"]}
-
""" + html = f'''AnomaVision Production Autopilot + +

Deployment confidence, before production.

Selected model: {escape(selected)} · Class: {escape(str(manifest["dataset"]["class_name"]))}

+
{"".join(cards)}

Model comparison

{"".join(rows)}
ModelImage AUROCPixel AUROCMedian msP95 msNormal FPThreshold
''' (output_dir / "production_autopilot_report.html").write_text(html, encoding="utf-8") - - markdown = [ - "# AnomaVision Production Autopilot Report", - "", - f"**Selected model:** `{selected}`", - "", - "See `production_autopilot_report.html` for the full dashboard.", - "", - ] (output_dir / "localization_report.md").write_text( - "\\n".join(markdown), encoding="utf-8" + f"# AnomaVision Production Autopilot Report\n\n**Selected model:** `{selected}`\n", + encoding="utf-8", ) +def _config_model(cfg: Dict[str, Any], name: str) -> Optional[str]: + section = cfg.get("autopilot", {}) or {} + value = section.get(f"{name}_model") + return str(value) if value else None + + def run(args: argparse.Namespace) -> Dict[str, Any]: - """Run Autopilot and create a deployment package.""" cfg = load_config(args.config) dataset_path = args.dataset_path or cfg.get("dataset_path") or cfg.get("img_path") class_name = args.class_name or cfg.get("class_name") if not dataset_path or not class_name: - raise ValueError( - "dataset_path and class_name are required in the CLI or config." - ) + raise ValueError("dataset_path and class_name are required in the CLI or config.") device = determine_device(args.device) dataset = anomavision.MVTecDataset( - dataset_path, - class_name, - is_train=False, - resize=cfg.get("resize", 224), - crop_size=cfg.get("crop_size", 224), - normalize=cfg.get("normalize", True), - mean=cfg.get("norm_mean"), - std=cfg.get("norm_std"), + dataset_path, class_name, is_train=False, + resize=cfg.get("resize", 224), crop_size=cfg.get("crop_size", 224), + normalize=cfg.get("normalize", True), mean=cfg.get("norm_mean"), std=cfg.get("norm_std"), ) - dataloader = DataLoader( - dataset, - batch_size=args.batch_size, - shuffle=False, - num_workers=args.num_workers, - pin_memory=False, - ) - candidates = {} - for name, model_path in ( - ("padim", args.padim_model), - ("patchcore", args.patchcore_model), - ): - if model_path: - candidates[name] = _profile_model( - model_path, dataloader, device, args.warmup, args.timing_batches - ) + dataloader = DataLoader(dataset, batch_size=args.batch_size, shuffle=False, num_workers=args.num_workers, pin_memory=False) + + paths = { + "padim": args.padim_model or _config_model(cfg, "padim"), + "patchcore": args.patchcore_model or _config_model(cfg, "patchcore"), + "efficientad": args.efficientad_model or _config_model(cfg, "efficientad"), + } + candidates = { + name: _profile_model(path, dataloader, device, args.warmup, args.timing_batches) + for name, path in paths.items() if path + } if not candidates: - raise ValueError("Provide at least one of --padim_model or --patchcore_model.") + raise ValueError("Provide at least one model: --padim_model, --patchcore_model, or --efficientad_model.") + selected = _select(candidates, args.target_latency_ms) output_dir = Path(args.output_dir) output_dir.mkdir(parents=True, exist_ok=True) selected_source = Path(candidates[selected]["model_path"]) packaged_model = output_dir / f"model{selected_source.suffix}" shutil.copy2(selected_source, packaged_model) + manifest = { - "schema_version": 2, + "schema_version": 3, "selected_model": selected, - "selected_artifact": str(packaged_model.name), - "dataset": { - "path": str(Path(dataset_path).resolve()), - "class_name": class_name, - "samples": len(dataset), - }, - "preprocessing": { - "resize": cfg.get("resize", 224), - "crop_size": cfg.get("crop_size", 224), - "normalize": cfg.get("normalize", True), - "mean": cfg.get("norm_mean"), - "std": cfg.get("norm_std"), - }, + "selected_artifact": packaged_model.name, + "dataset": {"path": str(Path(dataset_path).resolve()), "class_name": class_name, "samples": len(dataset)}, + "preprocessing": {"resize": cfg.get("resize", 224), "crop_size": cfg.get("crop_size", 224), "normalize": cfg.get("normalize", True), "mean": cfg.get("norm_mean"), "std": cfg.get("norm_std")}, "candidates": candidates, "target_latency_ms": args.target_latency_ms, - "environment": { - "python": sys.version.split()[0], - "platform": platform.platform(), - "torch": torch.__version__, - "device": device, - }, + "environment": {"python": sys.version.split()[0], "platform": platform.platform(), "torch": torch.__version__, "device": device}, } - (output_dir / "deployment_manifest.json").write_text( - json.dumps(manifest, indent=2), encoding="utf-8" - ) + (output_dir / "deployment_manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8") + if args.copy_config: + shutil.copy2(args.config, output_dir / Path(args.config).name) _write_report(manifest, output_dir) return manifest @@ -386,15 +261,7 @@ def run(args: argparse.Namespace) -> Dict[str, Any]: def main(args: Optional[argparse.Namespace] = None) -> None: args = args or create_parser().parse_args() manifest = run(args) - print( - json.dumps( - { - "selected_model": manifest["selected_model"], - "output_dir": str(Path(args.output_dir).resolve()), - }, - indent=2, - ) - ) + print(json.dumps({"selected_model": manifest["selected_model"], "output_dir": str(Path(args.output_dir).resolve())}, indent=2)) if __name__ == "__main__": diff --git a/anomavision/inference/model/backends/torch_backend.py b/anomavision/inference/model/backends/torch_backend.py index 09b36be..4f357c0 100644 --- a/anomavision/inference/model/backends/torch_backend.py +++ b/anomavision/inference/model/backends/torch_backend.py @@ -1,7 +1,5 @@ # inference/model/backends/torch_backend.py -""" -PyTorch backend implementation. -""" +"""PyTorch inference backend.""" from __future__ import annotations @@ -9,9 +7,8 @@ import torch -from anomavision.algorithm.padim.padim_lite import ( # stats-only .pth → runtime module - build_padim_from_stats, -) +from anomavision.algorithm.efficientad import build_efficientad_from_stats +from anomavision.algorithm.padim.padim_lite import build_padim_from_stats from anomavision.algorithm.patchcore import build_patchcore_from_stats from anomavision.utils import get_logger @@ -23,187 +20,69 @@ class TorchBackend(InferenceBackend): """Inference backend based on PyTorch.""" - def __init__( - self, - model_path: str, - device: str = "cpu", - *, - use_amp: bool = True, - ): - """Initialize PyTorch backend with automatic model type detection. - - Supports multiple PyTorch model formats including TorchScript, standard - PyTorch models, and PaDiM statistics files. Automatically handles model - preparation and optimization for inference. - - Args: - model_path (str): Path to PyTorch model file. Supported formats: - - TorchScript (.pts, .pt files) - - Standard PyTorch models (.pth with nn.Module) - - PaDiM statistics (.pth with mean/cov_inv dict) - device (str, optional): Target device. Automatically falls back to CPU - if CUDA is requested but unavailable. Defaults to "cpu". - use_amp (bool, optional): Enable automatic mixed precision (FP16) for - faster inference on supported GPUs. Defaults to True. - - Example: - >>> backend = TorchBackend("model.pts", "cuda", use_amp=True) - >>> backend = TorchBackend("stats.pth", "cpu") # PaDiM statistics - """ - - # --- Device selection: CPU-first; use CUDA only if requested & available + def __init__(self, model_path: str, device: str = "cpu", *, use_amp: bool = True): req = str(device or "cpu").lower() if req.startswith("cuda") and torch.cuda.is_available(): self.device = torch.device("cuda") else: - if req.startswith("cuda") and not torch.cuda.is_available(): + if req.startswith("cuda"): logger.warning("CUDA requested but not available; falling back to CPU.") self.device = torch.device("cpu") loaded_obj = None - - # 1) Try TorchScript first try: - logger.info("Trying torch.jit.load: %s", model_path) loaded_obj = torch.jit.load(model_path, map_location=self.device) logger.info("Loaded TorchScript model from %s", model_path) - except Exception as e_jit: - logger.info( - "torch.jit.load failed (%s). Falling back to torch.load.", str(e_jit) - ) + except Exception as exc: + logger.info("torch.jit.load failed (%s); falling back to torch.load.", exc) - # 2) Fallback: raw torch.load (may be nn.Module or a stats dict) if loaded_obj is None: - logger.info("Trying torch.load: %s", model_path) - loaded_obj = torch.load( - model_path, map_location=self.device, weights_only=False - ) - logger.info("Loaded object type: %s", type(loaded_obj).__name__) - - # 3) If it's a stats-only dict, build a PadimLite runtime module on CPU - if isinstance(loaded_obj, dict) and { - "mean", - "cov_inv", - "channel_indices", - "layer_indices", - "backbone", - }.issubset(loaded_obj.keys()): - logger.info("Detected PaDiM statistics artifact; building PadimLite.") + loaded_obj = torch.load(model_path, map_location=self.device, weights_only=False) + + if isinstance(loaded_obj, dict) and {"mean", "cov_inv", "channel_indices", "layer_indices", "backbone"}.issubset(loaded_obj): model = build_padim_from_stats(loaded_obj, device=device) - elif isinstance(loaded_obj, dict) and { - "memory_bank", - "layer_indices", - "backbone", - }.issubset(loaded_obj.keys()): - logger.info("Detected PatchCore memory-bank artifact; building PatchCore.") + elif isinstance(loaded_obj, dict) and {"memory_bank", "layer_indices", "backbone"}.issubset(loaded_obj): model = build_patchcore_from_stats(loaded_obj, device=device) + elif isinstance(loaded_obj, dict) and {"student", "backbone", "threshold"}.issubset(loaded_obj): + logger.info("Detected EfficientAD statistics artifact.") + model = build_efficientad_from_stats(loaded_obj, device=device) else: model = loaded_obj - # 4) Unwrap DataParallel if present if hasattr(model, "module"): - logger.info("Unwrapping DataParallel container.") model = model.module - - # 5) Finalize if hasattr(model, "eval"): model.eval() - # Disable grads if parameters exist if hasattr(model, "parameters"): - for p in model.parameters(): - p.requires_grad_(False) + for parameter in model.parameters(): + parameter.requires_grad_(False) self.model = model - # AMP only makes sense on CUDA self.use_amp = bool(use_amp and self.device.type == "cuda") - def predict(self, batch: Batch) -> ScoresMaps: - """Run PyTorch inference with automatic mixed precision support. - - Executes inference using PyTorch with optional AMP acceleration. Handles - device placement and tensor conversion automatically. - - Args: - batch (Batch): Input batch of images. Converted to torch.Tensor if needed. - - Returns: - ScoresMaps: Tuple containing: - - scores (np.ndarray): Per-image anomaly scores - - maps (np.ndarray): Pixel-level anomaly maps - - Note: - Uses model.predict() method for consistent interface across all - model types and export formats. - - Example: - >>> batch = torch.randn(2, 3, 224, 224) - >>> scores, maps = backend.predict(batch) - """ - - logger.debug("Running inference via TorchBackend") - - if not isinstance(batch, torch.Tensor): - batch = torch.as_tensor(batch, dtype=torch.float32) - - batch = batch.to(self.device, non_blocking=True) - logger.debug("Torch input shape: %s", tuple(batch.shape)) - - autocast_ctx = ( + def _autocast(self): + return ( torch.autocast(device_type=self.device.type, dtype=torch.float16) - if self.use_amp and self.device.type == "cuda" + if self.use_amp else nullcontext() ) - with torch.inference_mode(), autocast_ctx: - # Always use .predict to match your runtime/export path + def predict(self, batch: Batch) -> ScoresMaps: + if not isinstance(batch, torch.Tensor): + batch = torch.as_tensor(batch, dtype=torch.float32) + batch = batch.to(self.device, non_blocking=True) + with torch.inference_mode(), self._autocast(): scores, maps = self.model.predict(batch) - - scores_np = scores.detach().cpu().numpy() - maps_np = maps.detach().cpu().numpy() - logger.debug("Torch output shapes: %s, %s", scores_np.shape, maps_np.shape) - return scores_np, maps_np + return scores.detach().cpu().numpy(), maps.detach().cpu().numpy() def close(self) -> None: - """Release PyTorch model and clear GPU memory. - - Removes model reference and triggers garbage collection. Essential for - preventing memory leaks in applications that load multiple models. - """ - self.model = None def warmup(self, batch, runs: int = 2) -> None: - """Warm up PyTorch backend with AMP support. - - Performs warmup inference runs using the same settings as production - inference, including automatic mixed precision if enabled. - - Args: - batch: Input batch for warmup. Converted to appropriate tensor format. - runs (int, optional): Number of warmup iterations. Defaults to 2. - - Example: - >>> warmup_batch = torch.randn(1, 3, 224, 224) - >>> backend.warmup(warmup_batch, runs=3) - """ - if not isinstance(batch, torch.Tensor): batch = torch.as_tensor(batch, dtype=torch.float32, device=self.device) else: batch = batch.to(self.device, non_blocking=True) - - autocast_ctx = ( - torch.autocast(device_type=self.device.type, dtype=torch.float16) - if self.use_amp and self.device.type == "cuda" - else nullcontext() - ) - - with torch.inference_mode(), autocast_ctx: + with torch.inference_mode(), self._autocast(): for _ in range(max(1, runs)): - _ = self.model.predict(batch) - - logger.info( - "TorchBackend warm-up completed (runs=%d, shape=%s).", - runs, - tuple(batch.shape), - ) + self.model.predict(batch) From 4863e615da9f893c163e87dcdc19d10e32b19897 Mon Sep 17 00:00:00 2001 From: DeepKnowledge1 Date: Sun, 30 Aug 2026 19:40:02 +0200 Subject: [PATCH 19/24] readme --- README.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 8e93371..efaa63b 100644 --- a/README.md +++ b/README.md @@ -158,8 +158,9 @@ Train both candidate models first, then run the complete labeled split on CPU: ```bash anomavision autopilot \ --config config.yml \ - --padim_model ./distributions/padim/bottle/anomav_exp/model.pt \ - --patchcore_model ./distributions/patchcore/bottle/anomav_exp/model.pt \ + --padim_model ./distributions/padim/bottle/anomav_exp/model.onnx \ + --patchcore_model ./distributions/patchcore/bottle/anomav_exp/model.onnx \ + --efficientad_model ./distributions/efficientad/bottle/anomav_exp/model.onnx \ --device cpu \ --validation_split 1.0 \ --target_latency_ms 50 \ From 15450951e4e607b2ed612d1c031bca30de154329 Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sun, 30 Aug 2026 19:41:28 +0200 Subject: [PATCH 20/24] Fix Autopilot EfficientAD CLI and restore rich HTML report --- anomavision/autopilot.py | 69 +++++++++++++++++++++------------------- 1 file changed, 37 insertions(+), 32 deletions(-) diff --git a/anomavision/autopilot.py b/anomavision/autopilot.py index 7e4ef8b..205ab26 100644 --- a/anomavision/autopilot.py +++ b/anomavision/autopilot.py @@ -143,9 +143,7 @@ def _profile_model(model_path: str, dataloader: DataLoader, device: str, warmup: localization["anomaly_mean_mask_area_fraction"] = float(area[anomaly_idx].mean()) if anomaly_idx.any() else None localization["normal_mean_mask_area_fraction"] = float(area[normal_idx].mean()) if normal_idx.any() else None localization["verdict"] = ( - "healthy" - if (localization["normal_false_positive_fraction"] or 0.0) <= 0.10 - else "review false positives" + "healthy" if (localization["normal_false_positive_fraction"] or 0.0) <= 0.10 else "review false positives" ) if pixel_auroc is not None else "maps available; pixel AUROC unavailable" batch_size = max(1, dataloader.batch_size or 1) @@ -175,39 +173,46 @@ def _select(results: Dict[str, Dict[str, Any]], target_latency_ms: Optional[floa def _write_report(manifest: Dict[str, Any], output_dir: Path) -> None: selected = manifest["selected_model"] - cards = [] - rows = [] + selected_result = manifest["candidates"][selected] + target = manifest.get("target_latency_ms") + cards, rows = [], [] for name, result in manifest["candidates"].items(): metrics, loc = result["metrics"], result["localization"] active = name == selected + status = "Selected" if active else "Candidate" cards.append( - f'
{escape(name.upper())}' - f'

{_format_metric(metrics.get("image_auroc"))}

image AUROC' - f'

p95: {result["latency_ms"]["p95"]:.2f} ms · pixel AUROC: {_format_metric(metrics.get("pixel_auroc"))}

' + f'
{escape(name.upper())}{status}
' + f'
{_format_metric(metrics.get("image_auroc"))} image AUROC
' + f'
{_format_metric(metrics.get("pixel_auroc"))}pixel AUROC
{result["latency_ms"]["p95"]:.1f} msp95 latency
{_format_percent(loc.get("anomaly_non_empty_fraction"))}anomaly coverage
' ) rows.append( - f'{escape(name)}{_format_metric(metrics.get("image_auroc"))}' - f'{_format_metric(metrics.get("pixel_auroc"))}{result["latency_ms"]["median"]:.2f}' - f'{result["latency_ms"]["p95"]:.2f}{_format_percent(loc.get("normal_false_positive_fraction"))}' - f'{result["threshold"]:.6f}' + f'{escape(name)}{_format_metric(metrics.get("image_auroc"))}{_format_metric(metrics.get("pixel_auroc"))}{result["latency_ms"]["median"]:.2f}{result["latency_ms"]["p95"]:.2f}{_format_percent(loc.get("anomaly_non_empty_fraction"))}{_format_percent(loc.get("normal_false_positive_fraction"))}{result["threshold"]:.6f}' ) - html = f'''AnomaVision Production Autopilot - -

Deployment confidence, before production.

Selected model: {escape(selected)} · Class: {escape(str(manifest["dataset"]["class_name"]))}

-
{"".join(cards)}

Model comparison

{"".join(rows)}
ModelImage AUROCPixel AUROCMedian msP95 msNormal FPThreshold
''' + target_text = f"under {target:.1f} ms p95" if target is not None else "with the strongest measured accuracy/latency balance" + environment_json = escape(json.dumps(manifest["environment"], indent=2)) + html = f''' +AnomaVision Production Autopilot
+
AnomaVision / Production Autopilot

Deployment confidence, before production.

Calibrated thresholds, hardware-aware profiling, and localization health checks in one reproducible report.

Selected: {escape(selected)}Class: {escape(str(manifest["dataset"]["class_name"]))}Samples: {manifest["dataset"]["samples"]}Device: {escape(str(manifest["environment"].get("device", "unknown")))}
+
Recommendation
Deploy {escape(selected)} with threshold {selected_result["threshold"]:.6f}. It was selected {escape(target_text)}. Recheck this threshold on a production validation set before release.
+

Candidate overview

Measured on the same validation data
{"".join(cards)}
+

Detailed comparison

Higher AUROC is better; lower false positives and latency are better
{"".join(rows)}
ModelImage AUROCPixel AUROCMedian msP95 msAnomaly coverageNormal false positivesThreshold
+

Localization health

Selected model maps available{"PASS" if selected_result["localization"]["available"] else "CHECK"}
Anomaly images with localization{_format_percent(selected_result["localization"].get("anomaly_non_empty_fraction"))}
Normal images with false-positive maps{_format_percent(selected_result["localization"].get("normal_false_positive_fraction"))}
Localization verdict{escape(str(selected_result["localization"].get("verdict", "N/A")))}

Deployment artifact

Artifact{escape(str(manifest["selected_artifact"]))}
Format{escape(str(selected_result["model_format"]))}
Preprocessing{escape(str(manifest["preprocessing"].get("resize")))} px
Target latency{escape(str(target)) if target is not None else "not set"}
+

Reproducibility environment

{environment_json}
Generated by AnomaVision Production Autopilot · manifest schema {manifest["schema_version"]}
+
''' (output_dir / "production_autopilot_report.html").write_text(html, encoding="utf-8") (output_dir / "localization_report.md").write_text( - f"# AnomaVision Production Autopilot Report\n\n**Selected model:** `{selected}`\n", + f"# AnomaVision Production Autopilot Report\n\n**Selected model:** `{selected}`\n\nSee `production_autopilot_report.html` for the full dashboard.\n", encoding="utf-8", ) -def _config_model(cfg: Dict[str, Any], name: str) -> Optional[str]: - section = cfg.get("autopilot", {}) or {} - value = section.get(f"{name}_model") - return str(value) if value else None - - def run(args: argparse.Namespace) -> Dict[str, Any]: cfg = load_config(args.config) dataset_path = args.dataset_path or cfg.get("dataset_path") or cfg.get("img_path") @@ -222,15 +227,14 @@ def run(args: argparse.Namespace) -> Dict[str, Any]: ) dataloader = DataLoader(dataset, batch_size=args.batch_size, shuffle=False, num_workers=args.num_workers, pin_memory=False) - paths = { - "padim": args.padim_model or _config_model(cfg, "padim"), - "patchcore": args.patchcore_model or _config_model(cfg, "patchcore"), - "efficientad": args.efficientad_model or _config_model(cfg, "efficientad"), - } - candidates = { - name: _profile_model(path, dataloader, device, args.warmup, args.timing_batches) - for name, path in paths.items() if path - } + candidates = {} + for name, model_path in ( + ("padim", args.padim_model), + ("patchcore", args.patchcore_model), + ("efficientad", args.efficientad_model), + ): + if model_path: + candidates[name] = _profile_model(model_path, dataloader, device, args.warmup, args.timing_batches) if not candidates: raise ValueError("Provide at least one model: --padim_model, --patchcore_model, or --efficientad_model.") @@ -249,6 +253,7 @@ def run(args: argparse.Namespace) -> Dict[str, Any]: "preprocessing": {"resize": cfg.get("resize", 224), "crop_size": cfg.get("crop_size", 224), "normalize": cfg.get("normalize", True), "mean": cfg.get("norm_mean"), "std": cfg.get("norm_std")}, "candidates": candidates, "target_latency_ms": args.target_latency_ms, + "validation_split": args.validation_split, "environment": {"python": sys.version.split()[0], "platform": platform.platform(), "torch": torch.__version__, "device": device}, } (output_dir / "deployment_manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8") From b537feb382e2b891b1be91211ade0405894fd31e Mon Sep 17 00:00:00 2001 From: Deep Knowledge <66887716+DeepKnowledge1@users.noreply.github.com> Date: Sun, 30 Aug 2026 19:46:51 +0200 Subject: [PATCH 21/24] docs: update README for EfficientAD and Autopilot --- README.md | 135 ++++++++++++++++++++++++++++++++---------------------- 1 file changed, 81 insertions(+), 54 deletions(-) diff --git a/README.md b/README.md index efaa63b..d411f7a 100644 --- a/README.md +++ b/README.md @@ -21,14 +21,15 @@ KV260 DPU support

-AnomaVision is a computer vision project for finding **defects and unusual patterns** in images. +AnomaVision is a production-oriented computer vision toolkit for detecting **defects and unusual patterns** from normal images. -It supports two anomaly detection methods: +It supports three anomaly detection methods: -- **PaDiM** — a simple and fast baseline. -- **PatchCore** — a lightweight memory-based method. +- **PaDiM** — a simple, fast feature-distribution baseline. +- **PatchCore** — a lightweight memory-based method designed for efficient inference. +- **EfficientAD** — a lightweight student-teacher method designed for fast industrial anomaly detection. -You only need **normal (`good`) images** to train the anomaly detector. +Training requires only **normal (`good`) images**. Labeled test images can then be used for evaluation, threshold calibration, and production model selection.

Open the AnomaVision live demo @@ -39,50 +40,46 @@ You only need **normal (`good`) images** to train the anomaly detector. ## What can AnomaVision do? - Train anomaly detection models using normal images. -- Detect image-level anomalies. -- Create anomaly heatmaps showing where the problem is. -- Export models to **ONNX, OpenVINO, and TensorRT**. +- Detect image-level anomalies and generate anomaly heatmaps. +- Evaluate anomaly detection and localization performance. +- Calibrate anomaly thresholds from validation data. +- Export models to **ONNX, OpenVINO, and TensorRT** where supported. +- Run production model selection with **Production Autopilot**. - Export and compile **PaDiM and PatchCore to XModel for the AMD/Xilinx Kria KV260**. ## Quick start ### 1. Install -#### Option A — From Source (development) +#### Option A — From Source ```bash git clone https://github.com/DeepKnowledge1/AnomaVision.git cd AnomaVision -# Create and activate a virtual environment uv venv --python 3.11 .venv source .venv/bin/activate # Windows: .venv\Scripts\Activate.ps1 -# Install with your hardware extra uv sync --extra cpu # CPU uv sync --extra cu121 # CUDA 12.1 ``` ---- - -#### Option B — From PyPI (production / quick start) +#### Option B — From PyPI ```bash -# CPU · Mac, CI runners, edge devices uv pip install "anomavision[cpu]" -# NVIDIA GPU · pick your CUDA version -uv pip install "anomavision[cu118]" # CUDA 11.8 -uv pip install "anomavision[cu121]" # CUDA 12.1 -uv pip install "anomavision[cu124]" # CUDA 12.4 +# NVIDIA GPU +uv pip install "anomavision[cu118]" +uv pip install "anomavision[cu121]" +uv pip install "anomavision[cu124]" ``` - For other environments, see [Installation](docs/installation.md). -### 2. Prepare your images +### 2. Prepare your dataset -Use a simple MVTec-style folder structure: +Use an MVTec-style structure: ```text dataset/ @@ -100,19 +97,21 @@ dataset/ └── good/ ``` -Training uses the **good** images. Test images can contain defects. +Training uses only `train/good`. Test images may contain defects. ### 3. Train -Create or edit `config.yml` and point `dataset_path` to your dataset. - -Then run: +Create or edit `config.yml` and set `dataset_path` to your dataset. ```bash anomavision train --config config.yml ``` -PaDiM is the default model. For PatchCore, set `algorithm: patchcore` in the configuration. +Select the algorithm in the configuration: + +```yaml +algorithm: padim # padim | patchcore | efficientad +``` ### 4. Detect @@ -122,53 +121,83 @@ anomavision detect --config config.yml --img_path ./dataset/bottle/test ### 5. Export -For a portable model, ONNX is a good place to start: - ```bash anomavision export --config config.yml --format onnx ``` -For more export options, see [Export and deployment](docs/production_deployment.md). +See [Export and deployment](docs/production_deployment.md) for deployment-specific options. -## KV260 support +## Production Autopilot -AnomaVision also supports a **Vitis AI workflow for PaDiM and PatchCore on the AMD/Xilinx Kria KV260**. +Production Autopilot compares trained anomaly models on the **same validation data**, measures their performance and latency on the target device, calibrates thresholds, and selects the best candidate for deployment. -The workflow is: +PaDiM, PatchCore, and EfficientAD can be supplied as independent candidate models. The model paths are provided directly through the CLI: -```text -PyTorch → INT8 quantization → XModel → KV260 DPU compilation +```bash +anomavision autopilot \ + --config config.yml \ + --padim_model ./distributions/padim/bottle/anomav_exp/model.pt \ + --patchcore_model ./distributions/patchcore/bottle/anomav_exp/model.pt \ + --efficientad_model ./distributions/efficientad/bottle/anomav_exp/model.pt \ + --device cpu \ + --validation_split 1.0 \ + --target_latency_ms 50 \ + --output_dir ./production_package ``` -Both PaDiM and PatchCore currently compile with **1 DPU subgraph** in the KV260 compiler. +### How selection works -The complete setup and commands are in: +For every supplied model, Autopilot: -**[KV260 XModel Guide](docs/kv260_xmodel.md)** +1. Evaluates the model on the validation split. +2. Calibrates an image-level anomaly threshold. +3. Measures inference latency on the selected device. +4. Calculates image-level and pixel-level metrics when localization maps are available. +5. Checks localization quality and false-positive behavior. +6. Applies the target latency constraint when selecting the production candidate. +7. Packages the selected model and writes a deployment manifest. -> XModel compilation has been validated in the Vitis AI environment. Final on-device KV260 validation requires the physical hardware. +If multiple models satisfy the latency target, the model with the strongest image-level AUROC is preferred, with latency used as a tie-breaker. +### Output -## Production Autopilot +Autopilot creates a production package containing: -**Production Autopilot is the easiest way to move from two trained models to one deployable choice.** It compares PaDiM and ultra-light PatchCore on the same labeled test split, calibrates a separate threshold for each, profiles median and P95 latency on your hardware, checks localization health, and packages the selected artifact with a self-contained HTML dashboard. +```text +production_package/ +├── model.pt +├── deployment_manifest.json +├── localization_report.md +└── production_autopilot_report.html +``` -Train both candidate models first, then run the complete labeled split on CPU: +The HTML report is a self-contained dashboard showing the candidate comparison, selected model, AUROC, calibrated threshold, latency, localization diagnostics, and deployment recommendation. + +## Inference performance tests + +AnomaVision includes optional regression tests for PaDiM, PatchCore, and EfficientAD. The limits are controlled from the benchmark section of `config.yml`, so performance expectations can be adjusted for the target hardware. + +Run the benchmark with: ```bash -anomavision autopilot \ - --config config.yml \ - --padim_model ./distributions/padim/bottle/anomav_exp/model.onnx \ - --patchcore_model ./distributions/patchcore/bottle/anomav_exp/model.onnx \ - --efficientad_model ./distributions/efficientad/bottle/anomav_exp/model.onnx \ - --device cpu \ - --validation_split 1.0 \ - --target_latency_ms 50 \ - --output_dir ./production_package +pytest tests/test_inference_performance.py -s ``` -Open `production_package/production_autopilot_report.html` to see the selected model, AUROC, calibrated threshold, localization diagnostics, memory, median latency, P95 latency, and deployment recommendation. The package also contains `deployment_manifest.json`, `localization_report.md`, and the selected model artifact. See [`docs/production_deployment.md`](docs/production_deployment.md) for GPU, TensorRT, INT8, and packaging details. +The test reports pure model inference time, FPS, and throughput for each algorithm and fails when a configured performance limit is exceeded. + +## KV260 support +AnomaVision supports a **Vitis AI workflow for PaDiM and PatchCore on the AMD/Xilinx Kria KV260**. + +```text +PyTorch → INT8 quantization → XModel → KV260 DPU compilation +``` + +Both PaDiM and PatchCore currently compile with **1 DPU subgraph** in the KV260 compiler. + +See the complete [KV260 XModel Guide](docs/kv260_xmodel.md). + +> XModel compilation has been validated in the Vitis AI environment. Final on-device KV260 validation requires the physical hardware. ## Documentation @@ -187,8 +216,6 @@ Open `production_package/production_autopilot_report.html` to see the selected m ## Python example -You can also use AnomaVision directly from Python: - ```python import torch from torch.utils.data import DataLoader From 7867ddc409a8f5ee9e7b93f6d313f51f6f7a8147 Mon Sep 17 00:00:00 2001 From: DeepKnowledge1 Date: Sun, 30 Aug 2026 19:49:02 +0200 Subject: [PATCH 22/24] readme --- README.md | 20 +++++--------------- 1 file changed, 5 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index d411f7a..6f1e3b9 100644 --- a/README.md +++ b/README.md @@ -103,16 +103,17 @@ Training uses only `train/good`. Test images may contain defects. Create or edit `config.yml` and set `dataset_path` to your dataset. -```bash -anomavision train --config config.yml -``` - Select the algorithm in the configuration: ```yaml algorithm: padim # padim | patchcore | efficientad ``` +```bash +anomavision train --config config.yml +``` + + ### 4. Detect ```bash @@ -173,17 +174,6 @@ production_package/ The HTML report is a self-contained dashboard showing the candidate comparison, selected model, AUROC, calibrated threshold, latency, localization diagnostics, and deployment recommendation. -## Inference performance tests - -AnomaVision includes optional regression tests for PaDiM, PatchCore, and EfficientAD. The limits are controlled from the benchmark section of `config.yml`, so performance expectations can be adjusted for the target hardware. - -Run the benchmark with: - -```bash -pytest tests/test_inference_performance.py -s -``` - -The test reports pure model inference time, FPS, and throughput for each algorithm and fails when a configured performance limit is exceeded. ## KV260 support From 8f1f79ae4d95cf3dd76bdb8a039a190d4d0a8ed3 Mon Sep 17 00:00:00 2001 From: DeepKnowledge1 Date: Sun, 30 Aug 2026 19:50:21 +0200 Subject: [PATCH 23/24] precommit formating --- .../algorithm/efficientad/efficientad.py | 14 +- anomavision/autopilot.py | 197 ++++++++++++--- .../inference/model/backends/torch_backend.py | 26 +- anomavision/train.py | 225 +++++++++++++++--- tests/test_inference_performance.py | 1 - 5 files changed, 376 insertions(+), 87 deletions(-) diff --git a/anomavision/algorithm/efficientad/efficientad.py b/anomavision/algorithm/efficientad/efficientad.py index d3e0bd4..e7a4ee3 100644 --- a/anomavision/algorithm/efficientad/efficientad.py +++ b/anomavision/algorithm/efficientad/efficientad.py @@ -42,7 +42,9 @@ def __init__( ) -> None: super().__init__() if backbone != "resnet18": - raise ValueError("EfficientAD lightweight supports backbone='resnet18' only.") + raise ValueError( + "EfficientAD lightweight supports backbone='resnet18' only." + ) self.device = torch.device(device) self.backbone = backbone self.layer_indices = [0] @@ -96,7 +98,9 @@ def _raw_scores( ).squeeze(1) return image_scores, score_map - def fit(self, dataloader: torch.utils.data.DataLoader, extractions: int = 1) -> None: + def fit( + self, dataloader: torch.utils.data.DataLoader, extractions: int = 1 + ) -> None: """Train the student and calibrate an adaptive threshold on normal images.""" optimizer = torch.optim.Adam(self.student.parameters(), lr=self.learning_rate) @@ -162,7 +166,11 @@ def save_statistics(self, path: str, half: Optional[bool] = False) -> None: if not self._fitted: raise RuntimeError("EfficientAD is not fitted. Call fit() first.") student_state = { - key: value.detach().cpu().half() if half and value.is_floating_point() else value.detach().cpu() + key: ( + value.detach().cpu().half() + if half and value.is_floating_point() + else value.detach().cpu() + ) for key, value in self.student.state_dict().items() } torch.save( diff --git a/anomavision/autopilot.py b/anomavision/autopilot.py index 205ab26..642db7f 100644 --- a/anomavision/autopilot.py +++ b/anomavision/autopilot.py @@ -21,8 +21,11 @@ from anomavision.config import load_config from anomavision.general import determine_device from anomavision.inference.model.wrapper import ModelWrapper -from anomavision.utils import compute_metrics, find_optimal_threshold, make_localization_mask - +from anomavision.utils import ( + compute_metrics, + find_optimal_threshold, + make_localization_mask, +) ALGORITHMS = ("padim", "patchcore", "efficientad") @@ -32,12 +35,33 @@ def create_parser(add_help: bool = True) -> argparse.ArgumentParser: description="Select, calibrate, profile, and package a production anomaly model.", add_help=add_help, ) - parser.add_argument("--config", type=str, required=True, help="Base AnomaVision config file.") - parser.add_argument("--dataset_path", type=str, default=None, help="MVTec-style dataset root.") - parser.add_argument("--class_name", type=str, default=None, help="Dataset class to evaluate.") - parser.add_argument("--padim_model", type=str, default=None, help="PaDiM model artifact (.pt/.pth/.onnx).") - parser.add_argument("--patchcore_model", type=str, default=None, help="PatchCore model artifact (.pt/.pth/.onnx).") - parser.add_argument("--efficientad_model", type=str, default=None, help="EfficientAD model artifact (.pt/.pth/.onnx).") + parser.add_argument( + "--config", type=str, required=True, help="Base AnomaVision config file." + ) + parser.add_argument( + "--dataset_path", type=str, default=None, help="MVTec-style dataset root." + ) + parser.add_argument( + "--class_name", type=str, default=None, help="Dataset class to evaluate." + ) + parser.add_argument( + "--padim_model", + type=str, + default=None, + help="PaDiM model artifact (.pt/.pth/.onnx).", + ) + parser.add_argument( + "--patchcore_model", + type=str, + default=None, + help="PatchCore model artifact (.pt/.pth/.onnx).", + ) + parser.add_argument( + "--efficientad_model", + type=str, + default=None, + help="EfficientAD model artifact (.pt/.pth/.onnx).", + ) parser.add_argument("--device", choices=["auto", "cpu", "cuda"], default="auto") parser.add_argument("--batch_size", type=int, default=1) parser.add_argument("--num_workers", type=int, default=0) @@ -64,7 +88,13 @@ def _format_percent(value: Any) -> str: return "N/A" if value is None else f"{float(value):.1%}" -def _profile_model(model_path: str, dataloader: DataLoader, device: str, warmup: int, timing_batches: int) -> Dict[str, Any]: +def _profile_model( + model_path: str, + dataloader: DataLoader, + device: str, + warmup: int, + timing_batches: int, +) -> Dict[str, Any]: wrapper = ModelWrapper(model_path, device) try: first = next(iter(dataloader)) @@ -99,14 +129,20 @@ def _profile_model(model_path: str, dataloader: DataLoader, device: str, warmup: scores_np = np.asarray(all_scores, dtype=np.float32) labels_np = np.asarray(all_labels, dtype=np.int64) - maps_np = np.asarray(all_maps, dtype=np.float32) if all_maps else np.empty((0, 0, 0), dtype=np.float32) + maps_np = ( + np.asarray(all_maps, dtype=np.float32) + if all_maps + else np.empty((0, 0, 0), dtype=np.float32) + ) threshold, threshold_f1 = ( find_optimal_threshold(labels_np, scores_np) if len(np.unique(labels_np)) > 1 else (float(np.median(scores_np)), 0.0) ) image_metrics = compute_metrics(labels_np, scores_np, thresh=threshold) - image_auroc = image_metrics.get("auc_score") if len(np.unique(labels_np)) > 1 else None + image_auroc = ( + image_metrics.get("auc_score") if len(np.unique(labels_np)) > 1 else None + ) masks_np = np.asarray(all_masks, dtype=np.float32) if masks_np.ndim == 4 and masks_np.shape[1] == 1: @@ -122,9 +158,15 @@ def _profile_model(model_path: str, dataloader: DataLoader, device: str, warmup: "normal_mean_mask_area_fraction": None, "verdict": "unavailable", } - if localization["available"] and masks_np.shape == maps_np.shape and np.unique(masks_np).size > 1: + if ( + localization["available"] + and masks_np.shape == maps_np.shape + and np.unique(masks_np).size > 1 + ): try: - pixel_auroc = float(roc_auc_score(masks_np.reshape(-1) > 0.5, maps_np.reshape(-1))) + pixel_auroc = float( + roc_auc_score(masks_np.reshape(-1) > 0.5, maps_np.reshape(-1)) + ) except ValueError: pass image_metrics["image_auroc"] = image_auroc @@ -138,13 +180,27 @@ def _profile_model(model_path: str, dataloader: DataLoader, device: str, warmup: anomaly_idx, normal_idx = labels_np == 1, labels_np == 0 localization["non_empty_fraction"] = float(non_empty.mean()) localization["mean_mask_area_fraction"] = float(area.mean()) - localization["anomaly_non_empty_fraction"] = float(non_empty[anomaly_idx].mean()) if anomaly_idx.any() else None - localization["normal_false_positive_fraction"] = float(non_empty[normal_idx].mean()) if normal_idx.any() else None - localization["anomaly_mean_mask_area_fraction"] = float(area[anomaly_idx].mean()) if anomaly_idx.any() else None - localization["normal_mean_mask_area_fraction"] = float(area[normal_idx].mean()) if normal_idx.any() else None + localization["anomaly_non_empty_fraction"] = ( + float(non_empty[anomaly_idx].mean()) if anomaly_idx.any() else None + ) + localization["normal_false_positive_fraction"] = ( + float(non_empty[normal_idx].mean()) if normal_idx.any() else None + ) + localization["anomaly_mean_mask_area_fraction"] = ( + float(area[anomaly_idx].mean()) if anomaly_idx.any() else None + ) + localization["normal_mean_mask_area_fraction"] = ( + float(area[normal_idx].mean()) if normal_idx.any() else None + ) localization["verdict"] = ( - "healthy" if (localization["normal_false_positive_fraction"] or 0.0) <= 0.10 else "review false positives" - ) if pixel_auroc is not None else "maps available; pixel AUROC unavailable" + ( + "healthy" + if (localization["normal_false_positive_fraction"] or 0.0) <= 0.10 + else "review false positives" + ) + if pixel_auroc is not None + else "maps available; pixel AUROC unavailable" + ) batch_size = max(1, dataloader.batch_size or 1) median_ms = float(np.median(timings) * 1000 / batch_size) if timings else 0.0 @@ -154,21 +210,38 @@ def _profile_model(model_path: str, dataloader: DataLoader, device: str, warmup: "model_format": Path(model_path).suffix.lower(), "threshold": float(threshold), "threshold_f1": float(threshold_f1), - "metrics": {k: float(v) if isinstance(v, (float, np.floating)) else v for k, v in image_metrics.items()}, + "metrics": { + k: float(v) if isinstance(v, (float, np.floating)) else v + for k, v in image_metrics.items() + }, "latency_ms": {"median": median_ms, "p95": p95_ms}, - "throughput_images_per_second": float(1000.0 / median_ms) if median_ms > 0 else 0.0, + "throughput_images_per_second": ( + float(1000.0 / median_ms) if median_ms > 0 else 0.0 + ), "localization": localization, "samples": int(len(labels_np)), } -def _select(results: Dict[str, Dict[str, Any]], target_latency_ms: Optional[float]) -> str: +def _select( + results: Dict[str, Dict[str, Any]], target_latency_ms: Optional[float] +) -> str: eligible = results if target_latency_ms is not None: - eligible = {name: result for name, result in results.items() if result["latency_ms"]["p95"] <= target_latency_ms} + eligible = { + name: result + for name, result in results.items() + if result["latency_ms"]["p95"] <= target_latency_ms + } if not eligible: eligible = results - return max(eligible, key=lambda name: (eligible[name]["metrics"].get("image_auroc") or 0.0, -eligible[name]["latency_ms"]["p95"])) + return max( + eligible, + key=lambda name: ( + eligible[name]["metrics"].get("image_auroc") or 0.0, + -eligible[name]["latency_ms"]["p95"], + ), + ) def _write_report(manifest: Dict[str, Any], output_dir: Path) -> None: @@ -188,9 +261,13 @@ def _write_report(manifest: Dict[str, Any], output_dir: Path) -> None: rows.append( f'{escape(name)}{_format_metric(metrics.get("image_auroc"))}{_format_metric(metrics.get("pixel_auroc"))}{result["latency_ms"]["median"]:.2f}{result["latency_ms"]["p95"]:.2f}{_format_percent(loc.get("anomaly_non_empty_fraction"))}{_format_percent(loc.get("normal_false_positive_fraction"))}{result["threshold"]:.6f}' ) - target_text = f"under {target:.1f} ms p95" if target is not None else "with the strongest measured accuracy/latency balance" + target_text = ( + f"under {target:.1f} ms p95" + if target is not None + else "with the strongest measured accuracy/latency balance" + ) environment_json = escape(json.dumps(manifest["environment"], indent=2)) - html = f''' + html = f""" AnomaVision Production Autopilot