Tinsae/Finetuning
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added reports and params

Browse Source
This commit is contained in:
Tinsae
2026-07-03 13:31:43 +02:00
parent 524991dc4e
commit 61da187012
33 changed files with 4872 additions and 755 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
sets/Casia.py
View File

@@ -1,167 +0,0 @@
from torchvision import datasets, transforms
from torch.utils.data import Dataset, DataLoader, Subset
import torch
import numpy as np
import os
# train set transform
def train_transform(res):
return transforms.Compose([
transforms.Resize((res, res)),
transforms.RandomHorizontalFlip(p=0.5),
transforms.ColorJitter(
brightness=0.2,
contrast=0.2,
saturation=0.1
),
transforms.ToTensor(),
transforms.Normalize(
mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]
)
])
# test set transform
def test_transform(res):
return transforms.Compose([
transforms.Resize((res, res)),
transforms.ToTensor(),
transforms.Normalize(
mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]
)
])
# Load data using ImageFolder for CASIA-WebFace
'''
def get_set():
# This will check local cache first, then download if missing
print("Checking for CASIA-WebFace dataset...")
path = kagglehub.dataset_download("debarghamitraroy/casia-webface")
# Kagglehub often downloads a nested structure (e.g., path/casia-webface/casia-webface)
# We need the folder that directly contains the identity subfolders
# We'll check if there's a 'casia-webface' subfolder inside the downloaded path
sub_path = os.path.join(path, "casia-webface")
final_path = sub_path if os.path.exists(sub_path) else path
print(f"Loading dataset from: {final_path}")
return datasets.ImageFolder(
root=final_path,
transform=None
)'''
# Load data using ImageFolder for your UNPACKED images
def get_set():
# This must point to the folder created by Extractor.py
# NOT the kagglehub cache path
final_path = os.path.abspath("./data/casia-set")
if not os.path.exists(final_path):
raise FileNotFoundError(
f"Unpacked dataset not found at {final_path}. "
"Please run Extractor.py first!"
)
print(f"Loading unpacked CASIA dataset from: {final_path}")
return datasets.ImageFolder(
root=final_path,
transform=None
)
def get_ids_and_counts(dataset):
# ImageFolder stores labels in .targets
targets = torch.tensor(dataset.targets)
return torch.unique(
input = targets,
return_counts=True
)
def select_ids(dataset, sample_size, class_size):
ids, counts = get_ids_and_counts(dataset=dataset)
eligible_mask = counts >= sample_size
eligible_ids = ids[eligible_mask].numpy()
if len(eligible_ids) < class_size:
raise ValueError(
f"Only found {len(eligible_ids)} identities with {sample_size}+ images."
)
return np.random.choice(eligible_ids, class_size, replace=False)
def select_balanced_ids(dataset, class_size):
ids, counts = get_ids_and_counts(dataset=dataset)
sorted_indices = torch.argsort(counts, descending=True)
top_ids = ids[sorted_indices][:class_size].numpy()
return np.array(top_ids, dtype=int)
def get_indices(dataset, identities, split_at):
train_indices = []
test_indices = []
# We convert to numpy for faster searching with np.where
all_targets = np.array(dataset.targets)
for person_id in identities:
# Get all indices for this specific person
indices = np.where(all_targets == person_id)[0]
# Shuffle the indices for this person
np.random.shuffle(indices)
# Split data based on your split_at value
train_indices.extend(indices[:split_at])
test_indices.extend(indices[split_at:])
return train_indices, test_indices
# optional function to get max amount of samples per class
def select_top_ids(dataset, class_size):
ids, counts = get_ids_and_counts(dataset=dataset)
# sort by number of images (descending)
sorted_indices = torch.argsort(counts, descending=True)
top_ids = ids[sorted_indices][:class_size].numpy()
return np.array(top_ids, dtype=int)
def get_forget_retain_loaders(dataset: Dataset, forget_class_idx: int, batch_size: int = 32) -> tuple[DataLoader, DataLoader]:
"""
Splits an IdentitySubset or standard Dataset into forget and retain sets
based on a remapped target class index.
"""
# 1. Safely extract targets whether it's a standard dataset or a Subset wrapper
if hasattr(dataset, 'targets'):
targets = dataset.targets
elif hasattr(dataset, 'identity'): # Raw CelebA support
targets = dataset.identity
else:
# If it's an IdentitySubset or standard Subset, extract mapped targets sequentially
# This guarantees we get the 0 -> (n-1) remapped labels
targets = [dataset[i][1] for i in range(len(dataset))]
if not isinstance(targets, torch.Tensor):
targets = torch.tensor(targets)
# 2. Generate mask indices local to this subset
forget_indices = torch.where(targets == forget_class_idx)[0].tolist()
retain_indices = torch.where(targets != forget_class_idx)[0].tolist()
# 3. Create PyTorch Subsets
forget_subset = Subset(dataset, forget_indices)
retain_subset = Subset(dataset, retain_indices)
# 4. Wrap into clean DataLoaders
forget_loader = DataLoader(forget_subset, batch_size=batch_size, shuffle=False)
retain_loader = DataLoader(retain_subset, batch_size=batch_size, shuffle=True)
print(f"[Data Split] Local Class {forget_class_idx}: {len(forget_subset)} samples | Remaining Classes: {len(retain_subset)} samples.")
return forget_loader, retain_loader

21
sets/CasiaFace.py
View File

@@ -1,21 +0,0 @@
import os
from torchvision import datasets
from torch.utils.data import Dataset
import torch
from .Data import Data
class CasiaSet(Data):
def __init__(self, resolution: int = 224, sample_size = 190):
super().__init__(resolution = resolution, sample_size = sample_size)
def get_set(self) -> Data:
path_str = "./datasets/casia-set"
path = os.path.abspath(path_str)
if not os.path.exists(path):
raise FileNotFoundError(f"Unpacked dataset missing at {self.final_path}. Run Extractor.py first!")
print(f"Loading unpacked CASIA dataset from: {self.final_path}")
set = datasets.ImageFolder(root=path, transform=None)
# we set the target here
self.target = torch.tensor(set.targets)
return set

20
sets/CelebA.py
View File

@@ -1,20 +0,0 @@
from torchvision import datasets
from torch.utils.data import Dataset
import torch
from .Data import Data
class CelebA(Data):
def __init__(self, resolution: int = 224, sample_size = 30):
super().__init__(resolution, sample_size = sample_size)
def get_set(self):
set = datasets.CelebA(
root = "../data",
split='all',
target_type='identity',
download=False,
transform=None
)
# set the target first
self.target = set.identity
return set

15
sets/Data.py
View File

@@ -40,10 +40,11 @@ def test_transform(res):
)
])
# Load data with 'identity' as target and transform it
# Load data
def get_set(set_name:Set_Name):
return fetch_celeb_a() if set_name == Set_Name.CELEBA else fetch_casia_faces()
# celebA
def fetch_celeb_a():
return datasets.CelebA(
root='./data',
@@ -53,6 +54,7 @@ def fetch_celeb_a():
transform=None
)
# CASIA-WebFaces
def fetch_casia_faces():
# location of the data (path relative to project root)
final_path = os.path.abspath("./data/casia-set")
@@ -191,12 +193,12 @@ def vertical_split(dataset, batch_size,num_classes):
50% of each class goes to the Retain Set, 50% goes to the Forget Set.
"""
# 1. Group dataset indices by their respective ground-truth classes
# Dataset indices by their respective ground-truth classes
class_to_indices = {c: [] for c in range(num_classes)}
print(" [Vertical Split] Tracking class indices across the combined dataset...")
for idx in range(len(dataset)):
# Extract the label cleanly from the underlying dataset structure
# Extract labels
_, label = dataset[idx]
if label in class_to_indices:
class_to_indices[label].append(idx)
@@ -204,14 +206,14 @@ def vertical_split(dataset, batch_size,num_classes):
retain_indices = []
forget_indices = []
# 2. Slice each class identity vertically (exactly 50/50)
# Slice each class identity vertically (exactly 50/50)
for c, indices in class_to_indices.items():
if len(indices) < 2:
print(f" Warning: Class {c} has fewer than 2 samples. Cannot split vertically.")
retain_indices.extend(indices)
continue
# Deterministic shuffle per class to ensure honest distribution before splitting
# Suffle to ensure honest distribution before splitting
np.random.shuffle(indices)
mid = len(indices) // 2
@@ -220,11 +222,10 @@ def vertical_split(dataset, batch_size,num_classes):
print(f" Vertical split complete: Retain Index Size = {len(retain_indices)} | Forget Index Size = {len(forget_indices)}")
# 3. Construct lightweight PyTorch Subsets using our sliced index maps
# Subsets using our sliced index maps
retain_subset = Subset(dataset, retain_indices)
forget_subset = Subset(dataset, forget_indices)
# 4. Return pristine, shuffled DataLoaders mirroring your environment's batch specifications
retain_loader = DataLoader(retain_subset, batch_size=batch_size, shuffle=True)
forget_loader = DataLoader(forget_subset, batch_size=batch_size, shuffle=True)

50
sets/DataAnalyser.py Normal file
View File

@@ -0,0 +1,50 @@
#from Data import *
from sets.Casia import *
'''
Because the size of samples per class had the biggest impact
on training outcome, I decided to check the maximum amount of data
I can get from a class.
The highest I can get is
Rank | Identity ID |Count
-----------------------------------
1 | 3782 | 35
2 | 2820 | 35
3 | 3227 | 35
4 | 3745 | 34
5 | 3699 | 34
6 | 8968 | 32
7 | 9152 | 32
8 | 9256 | 32
9 | 2114 | 31
... | ... | ...
17 | 4126 | 31
18 | 3185 | 30
... | ... | ...
50 | 3186 | 30
as can be seen, 3 classes have 35, 2 have 34, 3 have 32 and the rest have 30.
'''
def print_top_identity_stats(dataset, top_n=50):
# we get data
ids, counts = get_ids_and_counts(dataset)
# sort in descending order
sorted_counts, sorted_indices = torch.sort(counts, descending=True)
# coresponding sorted ids
sorted_ids = ids[sorted_indices]
# 4. Slice the first 'top_n' and print
print(f"{'Rank':<8} | {'Identity ID':<12} | {'Image Count':<12}")
print("-" * 35)
for i in range(top_n):
identity_id = sorted_ids[i].item()
count = sorted_counts[i].item()
print(f"{i+1:<8} | {identity_id:<12} | {count:<12}")
# Usage:
dataset = get_set()
print_top_identity_stats(dataset, 50)

174
sets/Data_OOP.py
View File

@@ -1,174 +0,0 @@
import torch
import numpy as np
from abc import ABC, abstractmethod
from torchvision import transforms, datasets
from torch.utils.data import Dataset, DataLoader, Subset
class Data(ABC):
"""
Handles image pipelines, identity filtering, indexing, and unlearning splits.
"""
def __init__(self, res: int = 224, sample_size = 30, class_size = 20):
self.res = res
self.sample_size = sample_size
self.class_size = class_size
self.target = None # will have to be set in get_set()
def train_transform(self):
return transforms.Compose([
# ResNet expects 224 x 224 res
# Inception expects 299 x 299
transforms.Resize((self.res, self.res)),
transforms.RandomHorizontalFlip(p=0.5),
transforms.ColorJitter(
brightness=0.2,
contrast=0.2,
saturation=0.1
),
transforms.ToTensor(),
transforms.Normalize(
mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]
)
])
def test_transform(self):
return transforms.Compose([
transforms.Resize((self.res, self.res)),
transforms.ToTensor(),
transforms.Normalize(
mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]
)
])
@abstractmethod
def get_set(self)-> datasets.TorchDataset:
"""Loads and returns the raw underlying PyTorch Dataset instance."""
pass
def get_targets(self) -> torch.Tensor:
return self.target
def get_ids_and_counts(self) -> tuple[torch.Tensor, torch.Tensor]:
if self.target is None:
raise ValueError ("This should be called after the 'target' variable has been set.")
return torch.unique(
self.target,
return_counts=True
)
def select_ids(self) -> np.ndarray:
ids, counts = self.get_ids_and_counts()
eligible_mask = counts >= self.sample_size
eligible_ids = ids[eligible_mask].numpy()
if len(eligible_ids) < self.class_size:
raise ValueError(
f"Only found {len(eligible_ids)} identities with {self.sample_size}+ images."
)
return np.random.choice(eligible_ids, self.class_size, replace=False)
# Function to get max amount of samples per class
def select_top_ids(self) -> np.ndarray:
ids, counts = self.get_ids_and_counts()
# sort by number of images (descending)
sorted_indices = torch.argsort(counts, descending=True)
top_ids = ids[sorted_indices][:self.class_size].numpy()
return np.array(top_ids, dtype=int)
def get_indices(self, identities: np.ndarray, split_at: int, max_size: int = None) -> tuple[list, list]:
'''train_indices = []
test_indices = []
max_size = self.sample_size if max_size is None else max_size
# Pull raw target tensor array using concrete implementation rules
all_targets = np.array(self.get_targets().cpu())
np.random.seed(42)
for person_id in identities:
indices = np.where(all_targets == person_id)[0]
np.random.shuffle(indices)
# Constrain total sample tracking size if requested (e.g. CelebA ceiling)
current_pool = indices[:max_size] if max_size else indices
if split_at >= len(current_pool):
raise ValueError(f"Split point ({split_at}) exceeds slice size ({len(current_pool)}) for class {person_id}.")
train_indices.extend(current_pool[:split_at])
test_indices.extend(current_pool[split_at:])
return train_indices, test_indices'''
if split_at >= self.sample_size: # debug safety
raise ValueError(f"Split point ({split_at}) must be less than total size ({self.sample_size}).")
train_indices = []
test_indices = []
#training_sample = int(sample_size * training_ratio)
np.random.seed(42)
target = self.get_targets()
for person_id in identities:
# Get all indices for this specific person
indices = torch.where(target == person_id)[0].numpy()
# Shuffle the indices for this person
np.random.shuffle(indices)
# split data to testing and training
train_indices.extend(indices[:split_at])
test_indices.extend(indices[split_at:self.sample_size])
return train_indices, test_indices
@staticmethod
def get_unlearn_loaders(
dataset: Dataset,
forget_class_idx: int,
batch_size: int = 32
) -> tuple[DataLoader, DataLoader]:
"""Splits an IdentitySubset into forget/retain parts based on local class index."""
if hasattr(dataset, 'targets'):
targets = dataset.targets
elif hasattr(dataset, 'identity'):
targets = dataset.identity
else:
targets = [dataset[i][1] for i in range(len(dataset))]
if not isinstance(targets, torch.Tensor):
targets = torch.tensor(targets)
forget_indices = torch.where(targets == forget_class_idx)[0].tolist()
retain_indices = torch.where(targets != forget_class_idx)[0].tolist()
forget_subset = Subset(dataset, forget_indices)
retain_subset = Subset(dataset, retain_indices)
forget_loader = DataLoader(forget_subset, batch_size=batch_size, shuffle=False)
retain_loader = DataLoader(retain_subset, batch_size=batch_size, shuffle=True)
print(f"[Data Split] Local Class {forget_class_idx}: {len(forget_subset)} samples | Remaining Classes: {len(retain_subset)} samples.")
return forget_loader, retain_loader
@staticmethod
def getDataSet(set:SetType, sample_size):
# some test
if set == SetType.CASIA:
from sets.CasiaFace import CasiaFace
return CasiaFace(sample_size = sample_size)
if set == SetType.CELEBA:
from sets.CelebA import CelebA
return CelebA(sample_size=sample_size)
from enum import Enum, auto
class SetType(Enum):
CASIA = auto()
CELEBA = auto()