File size: 7,420 Bytes
961cf0c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 | from sklearn.model_selection import train_test_split
from torch.utils.data import Dataset, DataLoader
from sklearn.preprocessing import OneHotEncoder, LabelEncoder, StandardScaler
import pandas as pd
import numpy as np
from PIL import Image
from torchvision import transforms
import albumentations as A
from albumentations.pytorch import ToTensorV2
import torch
import os
import pickle
import cv2
class SkinLesionDataset(Dataset):
def __init__(self, metadata_file:str, img_dir:str, bert_model_name="one-hot-encoder", size:tuple=(224,224),
drop_nan:bool=False, random_undersampling:bool=False,
image_encoder:str="resnet-50", is_train:bool=True):
# Store parameters
self.metadata_file = metadata_file
self.img_dir = img_dir
self.size = size
self.bert_model_name=bert_model_name
self.is_to_drop_nan = drop_nan
self.random_undersampling = random_undersampling
self.image_encoder = image_encoder
self.is_train = is_train
self.targets = None
self.normalization = ([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
self.transform = self.load_transforms()
# Load metadata and process
self.metadata = self.load_metadata()
self.features, self.labels, self.targets = self.one_hot_encoding()
def __len__(self):
return len(self.metadata)
def __getitem__(self, idx):
image_name = self.metadata.iloc[idx]['img_id']
img_path = os.path.abspath(os.path.join(self.img_dir, image_name))
try:
with Image.open(img_path) as img:
image = img.convert("RGB")
image = np.array(image)
except Exception as e:
print(f"[Erro] Não foi possível abrir imagem com PIL: {img_path} — {e}")
raise FileNotFoundError(f"Imagem inválida: {img_path}")
if self.transform:
image = self.transform(image=image)['image']
metadata = torch.tensor(self.features[idx], dtype=torch.float32)
label = torch.tensor(self.labels[idx], dtype=torch.long)
return image_name, image, metadata, label
def load_transforms(self):
"""
Define as transformações de imagem para treino/validação.
- Treino:
* Resize fixo
* Rotate moderado (±45°) – implementação própria do Albumentations (cv2), sem skimage.AffineTransform
* Flips horizontal/vertical
* Blur, dropout e variação de cor
* Normalização + ToTensorV2
- Val/Test:
* Apenas Resize + Normalize + ToTensorV2
"""
if self.is_train:
return A.Compose([
# Ajuste de tamanho base
A.Resize(self.size[0], self.size[1]),
# Geométricas SEGURAS (sem Affine / ShiftScaleRotate)
A.Rotate(
limit=45,
border_mode=cv2.BORDER_REFLECT,
p=0.5
),
# Flips
A.HorizontalFlip(p=0.5),
A.VerticalFlip(p=0.2),
# Blur
A.GaussianBlur(sigma_limit=(0, 2.0), p=0.25),
# Dropout leve (oclusões pequenas)
A.CoarseDropout(
max_holes=5,
max_height=8,
max_width=8,
p=0.15
),
# Variações de cor/iluminação
A.HueSaturationValue(
hue_shift_limit=10,
sat_shift_limit=15,
val_shift_limit=10,
p=0.25
),
A.RandomBrightnessContrast(p=0.25),
# Normalização + tensor
A.Normalize(mean=self.normalization[0], std=self.normalization[1]),
ToTensorV2(),
])
else:
# Validação / teste: sem augmentations fortes
return A.Compose([
A.Resize(self.size[0], self.size[1]),
A.Normalize(mean=self.normalization[0], std=self.normalization[1]),
ToTensorV2(),
])
def load_metadata(self):
# Carregar o CSV
metadata = pd.read_csv(self.metadata_file).fillna("EMPTY").replace(" ", "EMPTY").replace(" ", "EMPTY").\
replace("NÃO ENCONTRADO", "EMPTY").replace("BRASIL","BRAZIL")
# Verificar se deve descartar linhas com NaN
if self.is_to_drop_nan:
metadata = metadata.dropna().reset_index(drop=True)
return metadata
def one_hot_encoding(self):
dataset_features = self.metadata.drop(
columns=['patient_id', 'lesion_id', 'img_id', 'biopsed', 'diagnostic']
)
# Colunas numéricas fixas
numerical_cols = ['age', 'diameter_1', 'diameter_2']
categorical_cols = [col for col in dataset_features.columns if col not in numerical_cols]
# Converter categóricas
dataset_features[categorical_cols] = dataset_features[categorical_cols].astype(str)
# Forçar numérico nas colunas numéricas, substituindo inválidos por NaN
dataset_features[numerical_cols] = dataset_features[numerical_cols].apply(
pd.to_numeric, errors="coerce"
)
# Preencher valores faltantes (NaN gerados acima) com -1
dataset_features[numerical_cols] = dataset_features[numerical_cols].fillna(-1)
# Caminho base
base_dir = os.path.join("./data", "preprocess_data")
os.makedirs(base_dir, exist_ok=True)
# OneHotEncoder
ohe_path = os.path.join(base_dir, "ohe_pad_20.pickle")
if os.path.exists(ohe_path):
with open(ohe_path, "rb") as f:
ohe = pickle.load(f)
categorical_data = ohe.transform(dataset_features[categorical_cols])
else:
ohe = OneHotEncoder(sparse_output=False, handle_unknown='ignore')
categorical_data = ohe.fit_transform(dataset_features[categorical_cols])
with open(ohe_path, "wb") as f:
pickle.dump(ohe, f)
# StandardScaler
scaler_path = os.path.join(base_dir, "scaler_pad_20.pickle")
if os.path.exists(scaler_path):
with open(scaler_path, "rb") as f:
scaler = pickle.load(f)
numerical_data = scaler.transform(dataset_features[numerical_cols])
else:
scaler = StandardScaler()
numerical_data = scaler.fit_transform(dataset_features[numerical_cols])
with open(scaler_path, "wb") as f:
pickle.dump(scaler, f)
# Concatenar dados
processed_data = np.hstack((categorical_data, numerical_data))
# Labels
labels = self.metadata['diagnostic'].values
le_path = os.path.join(base_dir, "label_encoder_pad_20.pickle")
if os.path.exists(le_path):
with open(le_path, "rb") as f:
label_encoder = pickle.load(f)
encoded_labels = label_encoder.transform(labels)
else:
label_encoder = LabelEncoder()
encoded_labels = label_encoder.fit_transform(labels)
with open(le_path, "wb") as f:
pickle.dump(label_encoder, f)
return processed_data, encoded_labels, self.metadata['diagnostic'].unique()
|