unlearning LF

unlearning/LinearFiltration.py

@@ -1,29 +1,48 @@
 
 import torch
-from Strategy import Strategy
+import torch.nn as nn
+from .Strategy import Strategy
 
-class NormalizingLinearFiltration(Strategy):
-    def __init__(self, target_class_idx):
+class LinearFiltration(Strategy):
+    def __init__(self, target_class_idx: int):
+        super().__init__()  # Automatically configures 'NormalizingLinearFiltration_metrics.txt'
         self.target_class_idx = target_class_idx
 
-    def apply(self, model, forget_loader, retain_loader):
+    def _run(self, model: nn.Module) -> nn.Module:
         model.eval()
-        # Freeze parameters structurally
         for param in model.parameters():
             param.requires_grad = False
             
         with torch.no_grad():
-            # we modify only classification head
-            # Shape: [num_classes, feature_dim]
-            W = model.fc.weight.data
-            
-            # Compute the normalization transformation projection matrix (A)
-            # (In your full code, calculate A here matching Baumhauer et al.'s equations)
+            W = model.fc.weight.data.clone()
             num_classes = W.shape[0]
-            A = torch.eye(num_classes, device=W.device)
-            # Mask/blend target class index distribution configurations here...
-            A[self.target_class_idx, :] = 0.0 
             
-            # 3. Direct weight matrix override: W_filtered = A * W
+            A = self._calculate_filtration_matrix(num_classes, self.target_class_idx, W.device)
             sanitized_W = torch.mm(A, W)
-            model.fc.weight.copy_(sanitized_W)
+            model.fc.weight.copy_(sanitized_W)
+            
+        return model
+
+    '''@staticmethod
+    def _calculate_filtration_matrix(num_classes: int, forget_class: int, device: torch.device) -> torch.Tensor:
+        A = torch.eye(num_classes, device=device)
+        num_remaining_classes = num_classes - 1
+        for j in range(num_classes):
+            if j == forget_class:
+                A[forget_class, j] = 0.0
+            else:
+                A[forget_class, j] = 1.0 / num_remaining_classes
+        return A'''
+    @staticmethod
+    def _calculate_filtration_matrix(num_classes: int, forget_class: int, device: torch.device) -> torch.Tensor:
+        A = torch.eye(num_classes, device=device)
+        num_remaining = num_classes - 1
+        
+        # The row of the forgotten class should average the inputs of all other classes
+        for j in range(num_classes):
+            if j == forget_class:
+                A[forget_class, j] = 0.0
+            else:
+                A[forget_class, j] = 1.0 / num_remaining
+                
+        return A

unlearning/Strategy.py

@@ -0,0 +1,47 @@
+
+import torch.nn as nn
+import time
+import os
+
+class Strategy:
+    """Abstract base class for unlearning algorithms with automated, strategy-specific logging."""
+    
+    def __init__(self):
+        # Dynamically set file name based on the class name (e.g., 'NormalizingLinearFiltration.txt')
+        self.strategy_name = self.__class__.__name__
+        self.log_file = f"reports/{self.strategy_name}_metrics.txt"
+        self._initialize_log_file()
+
+    def _initialize_log_file(self):
+        """Creates a unique log file for this strategy with a header if it doesn't exist."""
+        if not os.path.exists(self.log_file):
+            with open(self.log_file, "w") as f:
+                f.write("execution_time_sec\n")
+
+    def log_metric(self, execution_time: float):
+        """Appends the execution time to this strategy's specific file."""
+        with open(self.log_file, "a") as f:
+            f.write(f"{execution_time:.6f}\n")
+
+    def apply(self, model: nn.Module) -> nn.Module:
+        """
+        Wraps the unlearning execution with automated timing and strategy-specific logging.
+        DO NOT override this method in subclasses. Override _run instead.
+        """
+        start_time = time.perf_counter()
+        
+        # Execute core unlearning logic
+        processed_model = self._run(model)
+        
+        end_time = time.perf_counter()
+        execution_time = end_time - start_time
+        
+        # Log to the strategy's specific file
+        self.log_metric(execution_time)
+        print(f"[{self.strategy_name}] Completed in {execution_time:.6f} seconds. Saved to {self.log_file}")
+        
+        return processed_model
+
+    def _run(self, model: nn.Module) -> nn.Module:
+        """Subclasses implement their core unlearning logic here."""
+        raise NotImplementedError