Introduction aux Données Multimodales pour la Sécurité
L'automatisation de la détection d'incendies nécessite souvent une combinaison de capteurs pour garantir la fiabilité dans des environnements complexes. Un ensmeble de données spécialisé couplant imagerie visible et infrarouge permet d'affiner les modèles de vision par ordinateur. Ce jeu de données comprend plus de 1300 paires d'images synchronisées, offrant une couverture variée incluant feux forestiers, bâtiments et expériences en laboratoire.
Les caractéristiques techniques incluent une résolution typique entre 640x480 et 1024x768 pixels. Les annotations sont réalisées sous forme de masques binaires où le blanc indique la zone du feu. Le volume total se situe autour de 800 Mo, compatible avec les frameworks standards comme TensorFlow ou PyTorch.
Architecture du Projet
Pour assurer une maintenance efficace, l'organisation des fichiers suit une structure modulaire :
project_root/
│── config.yaml # Configuration des paramètres
│── scripts/
│ ├── prepare_data.py # Script de préparation
│ └── infer.py # Exécution des prédictions
│── models/
│ └── net_v2.py # Implémentation neuronale
│── datasets/
│ ├── loader.py # Gestion des chargeurs
│ └── transforms.py # Normalisation et ajustement
└── assets/
├── vis_frames/ # Images caméra standard
├── ir_frames/ # Images thermiques
└── ann_masks/ # Annotations ground truth
Gestion du Flux de Données
Le chargement des données est centralisé dans un module dédié qui gère les chemins absolus et applique les transformations nécessaires avant l'alimentation du réseau. Voici une implémentation utilisant les utilitaires PyTorch :
# datasets/loader.py
import os
from torch.utils.data import Dataset, DataLoader
from PIL import Image
import numpy as np
import torchvision.transforms as T
class DoubleSpectralFireSet(Dataset):
def __init__(self, racine_vis, racine_ir, racine_mask, transforms=None):
"""
Initialisation du dataset multi-canaux.
:param racine_vis: Dossier contenant les images RGB
:param racine_ir: Dossier contenant les images Luminance
:param racine_mask: Dossier contenant les masques PNG
"""
self.champs_vis = sorted([os.path.join(racine_vis, f)
for f in os.listdir(racine_vis)])
self.champs_ir = sorted([os.path.join(racine_ir, f)
for f in os.listdir(racine_ir)])
self.champs_ann = sorted([os.path.join(racine_mask, f)
for f in os.listdir(racine_mask)])
assert len(self.champs_vis) == len(self.champs_ir) == len(self.champs_ann), \
"Nombre d'images non aligné entre les sources"
self.transforms = transforms
def __len__(self):
return len(self.champs_vis)
def __getitem__(self, idx):
img_v_path = self.champs_vis[idx]
img_i_path = self.champs_ir[idx]
msk_path = self.champs_ann[idx]
ch_v = Image.open(img_v_path).convert('RGB')
ch_i = Image.open(img_i_path).convert('L')
msk_img = Image.open(msk_path).convert('L')
if self.transforms:
ch_v = self.transforms(ch_v)
ch_i = self.transforms(ch_i)
# Conversion masque : binarisation stricte
msk_arr = np.array(msk_img)
msk_tensor = torch.from_numpy(msk_arr > 127).float()
msk_tensor.unsqueeze_(0) # Ajout dimension channel
return {
'vis': ch_v,
'ir': ch_i,
'gt': msk_tensor
}
def charger_donnees(path_v, path_ir, path_m, batch_size=4, nb_workers=2):
trans_f = T.Compose([
T.Resize((384, 256)),
T.ToTensor(),
T.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
ds = DoubleSpectralFireSet(path_v, path_ir, path_m, trans_f)
dl = DataLoader(ds, batch_size=batch_size, shuffle=True, num_workers=nb_workers, drop_last=True)
return dl
Conception du Réseau Neuronal
L'architecture choisie utilise un flux bifurqué pour traiter séparément les canaux thermqiues et optiques avant leur fusion. L'utilisation de Batch Normalization aide à la convergence lors de l'entraînement sur ce type de données hétérogènes.
# models/net_v2.py
import torch
import torch.nn as nn
class ArborescenceU_Fusion(nn.Module):
def __init__(self):
super().__init__()
# Encodeur Visuel (3 canaux)
self.enc_1_v = nn.Sequential(
nn.Conv2d(3, 32, 3, padding=1),
nn.BatchNorm2d(32), nn.ReLU(inplace=True),
nn.Conv2d(32, 32, 3, padding=1),
nn.BatchNorm2d(32), nn.ReLU(inplace=True)
)
# Encodeur Thermique (1 canal)
self.enc_1_ir = nn.Sequential(
nn.Conv2d(1, 32, 3, padding=1),
nn.BatchNorm2d(32), nn.ReLU(inplace=True),
nn.Conv2d(32, 32, 3, padding=1),
nn.BatchNorm2d(32), nn.ReLU(inplace=True)
)
# Bloc de Fusion Central
self.fuse_conv = nn.Conv2d(64, 64, 3, padding=1)
self.dec_block = nn.Sequential(
nn.Conv2d(64, 32, 1),
nn.ReLU(inplace=True),
nn.Conv2d(32, 1, 1),
nn.Sigmoid()
)
def forward(self, x_vis, x_ir):
emb_v = self.enc_1_v(x_vis)
emb_ir = self.enc_1_ir(x_ir)
# Concaténation des cartes de caractéristiques
fused_features = torch.cat([emb_v, emb_ir], dim=1)
out = self.fuse_conv(fused_features)
return self.dec_block(out)
def charger_modele():
return ArborescenceU_Fusion()
Orchestration de l'Entraînement
La boucle principale gère les étapes de gradient et sauvegarde定期检查s périodiques de l'état du modèle. Le calcul de perte utilise une fonction adaptée aux sorties probabilités binaires.
# scripts/train_main.py
import torch
import torch.optim as optim
from torch.utils.tensorboard import SummaryWriter
from models.net_v2 import charger_modele
from datasets.loader import charger_donnees
import torch.nn as nn
def execute_entrainement():
writer = SummaryWriter('logs/exp_segmentation')
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
modele = charger_modele().to(device)
# Chemins définis selon structure projet
loader_train = charger_donnees('./assets/vis_frames',
'./assets/ir_frames',
'./assets/ann_masks')
optimiseur = optim.Adam(modele.parameters(), lr=1e-3)
critere_perte = nn.BCELoss()
nombre_epochs = 30
for ep in range(nombre_epochs):
modele.train()
somme_perdu = 0.0
for lot in loader_train:
v = lot['vis'].to(device)
i = lot['ir'].to(device)
m = lot['gt'].to(device)
optimiseur.zero_grad()
pred = modele(v, i)
p_val = critere_perte(pred, m)
p_val.backward()
optimiseur.step()
somme_perdu += p_val.item()
moyenne_loss = somme_perdu / len(loader_train)
print(f'Epoch {ep+1} | Loss: {moyenne_loss:.4f}')
writer.add_scalar('Train_Loss', moyenne_loss, ep)
torch.save(modele.state_dict(), f'models/poids_ep{ep+1}.pth')
if __name__ == '__main__':
execute_entrainement()
Validation et Inférence
Après entraînement, le script de démonstration charge les poids et affiche la superposition du résultat prédit sur les entrées originales.
# scripts/infer.py
from models.net_v2 import charger_modele
from datasets.loader import DoubleSpectralFireSet
import torch
from PIL import Image
import matplotlib.pyplot as plt
import os
def verifier_resultat():
poids_fichier = 'models/poids_ep30.pth'
modeles = charger_modele()
modeles.load_state_dict(torch.load(poids_fichier))
modeles.eval()
# Sélection d'un échantillon arbitraire
id_echantillon = 10
# Reconstruction chemin manuel pour démo rapide
vis_src = f'./assets/vis_frames/{list(os.listdir("./assets/vis_frames"))[id_echantillon]}'
ir_src = f'./assets/ir_frames/{list(os.listdir("./assets/ir_frames"))[id_echantillon]}'
t_modif = T.Compose([T.Resize((384, 256)), T.ToTensor()])
img_vis = Image.open(vis_src).convert('RGB')
img_ir = Image.open(ir_src).convert('L')
inp_v = t_modif(img_vis).unsqueeze(0).to('cuda')
inp_i = t_modif(img_ir).unsqueeze(0).to('cuda')
with torch.no_grad():
out_map = modeles(inp_v, inp_i)
map_binaire = (out_map > 0.5).float()
# Visualisation
fig, axes = plt.subplots(1, 3, figsize=(10, 3))
axes[0].imshow(img_vis)
axes[0].set_title("Source Visible")
axes[0].axis('off')
axes[1].imshow(img_ir, cmap='gray')
axes[1].set_title("Source Thermique")
axes[1].axis('off')
axes[2].imshow(map_binaire.squeeze().cpu().numpy(), cmap='gray')
axes[2].set_title("Masque Déduit")
axes[2].axis('off')
plt.tight_layout()
plt.show()