Nos recherches en neurosciences ont généré un ensemble de données EEG stockées au format .dat, présentant une structure complexe nécessitant une transformation. Le format de données se compose de trois parties distinctes :
- Première partie : en-tête d'information. Stocké sous un format similaire à un dictionnaire JSON, occupant une seule ligne avec plus de 500 colonnes. Lors de la lecture, ces données sont traitées comme des chaînes de caractères devant être converties en dictionnaire.
- Deuxième partie : corps des données. Séparées par des espaces, chaque ligne contient quatre valeurs de type flottant. Le nombre de lignes est variable. Après extraction, ces chaînes doivent être converties en liste de nombres flotants.
- Troisième partie : ligne de paramètres. Similaire à la première partie, située après le corps des données, pouvant occuper plusieurs lignes en raison d'un formatage imparfait, avec plus de 1000 colonnes.
Voici le processus détaillé pour chaque partie.
1. Préparation des bibliothèques et des répertoires :
import csv
import os
import json
repertoire_entree = 'D:\\Desktop\\donnees_eeg\\dat\\'
repertoire_sortie = 'D:\\Desktop\\donnees_eeg\\csv\\'
2. Lecture de l'en-tête d'information
Soit fichier_entree le fichier d'entrée. Nous ouvrons le fichier avec l'encodage utf-8-sig, et nous utiliserons la même méthode pour l'écriture. Nous utilisons readlines() pour lire le fichier ligne par ligne dans une liste.
fr = open(fichier_entree, encoding='utf-8-sig', errors='ignore')
lignes = fr.readlines()
# Lecture de l'en-tête
chaine_en_tete = lignes[0] # Extraction de la chaîne au format dictionnaire
dictionnaire_en_tete = json.loads(chaine_en_tete)
with open(file=fichier_sortie, mode='w', newline='', encoding='utf-8-sig') as f:
redacteur = csv.DictWriter(f, fieldnames=dictionnaire_en_tete.keys())
redacteur.writeheader()
redacteur.writerow(dictionnaire_en_tete)
Nous extrayons la première ligne et utilisons json.loads() pour convertir la chaîne en dictionnaire. Notez que cette chaîne respecte déjà le format dictionnaire JSON (par exemple : {"cle1": valeur1, "cle2": valeur2}).
Nous utilisons csv.DictWriter() pour écrire les données au format CSV dans un fichier. writer.writerow() permet d'écrire ligne par ligne dans le fichier de sortie. L'utilisation de with open simplifie la gestion de la fermeture du fichier.
3. Lecture du corps des données
Comme le nombre de lignes du corps des données est variable, nous définissons longueur_donnees comme le numéro de ligne de fin de la section de données (notez qu'il s'agit du numéro de ligne de fin, pas du nombre de lignes). Nous remarquons que chaque fichier contient une ligne d'en-tête, donc le corps des données commence réellement à la deuxième ligne. Nous avons adopté une approche pragmatique : le corps des données ne contient que 15 colonnes environ, tandis que les sections de paramètres et d'en-tête contiennent des centaines de colonnes, nous utilissons donc le nombre de colonnes comme critère de détection - plus de 100 indique une ligne de paramètre ou d'en-tête.
Nous avons identifié trois scénarios : une seule ligne de paramètres en fin de fichier (param_lignes=1, cas NORMAL), deux lignes ou plus de paramètres (param_lignes>=2, cas MULTI_PARAM), et l'absence de ligne de paramètres (param_lignes=0, cas SANS_PARAM). Nous utilisons estNormal pour enregistrer le scénario.
# Lecture de la section de données
# Après la section de données, il y a au maximum deux lignes de paramètres
# Par conséquent, nous devons calculer spécifiquement la longueur de la section de données
# Scénario normal
longueur_totale = len(lignes)
estNormal = "NORMAL"
# Calcul du nombre de lignes de paramètres
param_lignes = 0 # Initialisation du compteur
for indice in range(0, longueur_totale):
if len(lignes[indice]) > 100:
param_lignes += 1
# Détermination du type de données
# Notez qu'il y a au moins une ligne de paramètres, la première ligne
longueur_donnees = 0 # Initialisation de la longueur de la section de données
if param_lignes >= 3: # Vérification si l'avant-dernière ligne est un paramètre
longueur_donnees = longueur_totale - param_lignes + 1 # Notez que la première ligne est incluse, donc nous ajoutons 1
estNormal = "MULTI_PARAM"
elif param_lignes == 1:
longueur_donnees = longueur_totale
estNormal = "SANS_PARAM"
elif param_lignes == 2:
longueur_donnees = longueur_totale - param_lignes + 1
estNormal = "NORMAL"
for ligne in range(1, longueur_donnees):
ligne_donnees = lignes[ligne]
liste_chaine = ligne_donnees.split()
liste_float = list(map(float, liste_chaine))
with open(file=fichier_sortie, mode='a', newline='') as f:
redacteur = csv.writer(f)
redacteur.writerow(liste_float)
Dans les cas NORMAL et MULTI_PARAM, la ligne de fin du corps des données est : nombre total de lignes - nombre de lignes de paramètres et d'en-tête (param_lignes) + 1 (notez que l'index commence à 0). Dans le cas SANS_PARAM, la ligne de fin du corps des données est simplement la dernière ligne du fichier. Nous pourrions optimiser et fusionner cette partie du code, mais pour coller aux spécificités des données, nous l'avons maintenue en trois cas.
Les données sont séparées à l'aide de la fonction .split(), qui renvoie une liste.
4. Lecture des lignes de paramètres
Lors de la lecture des lignes de paramètres, bien que les données semblent déjà formatées en dictionnaire, elles sont en réalité de type chaîne et nécessitent une conversion. Nous utilisons json.loads() pour convertir la chaîne déjà formatée en dictionnaire. Notez que lors de l'écriture, le mode de la fonction open devrait être 'a'.
Dans le cas NORMAL, la ligne de paramètres en fin de fichier ne devrait contenir qu'une seule ligne, donc nous extrayons simplement la dernière ligne ; dans le cas MULTI_PARAM, notez l'initialisation du dictionnaire de stockage dict_param = {}, et nous parcourons toutes les lignes de paramètres (du numéro de ligne longueur_totale - param_lignes + 1 à longueur_totale - 1) pour les écrire, en utilisant .update() pour fusionner les dictionnaires ; dans le cas SANS_PARAM, nous sortons simplement de la fonction.
# Lecture de la section de paramètres
chaine_param = {} # Stockage des chaînes au format dictionnaire
dict_param = {} # Stockage des dictionnaires
if estNormal == "NORMAL":
chaine_param = lignes[longueur_totale - 1] # Extraction de la chaîne au format dictionnaire
dict_param = json.loads(chaine_param)
with open(file=fichier_sortie, mode='a', newline='', encoding='utf-8-sig') as f:
redacteur = csv.DictWriter(f, fieldnames=dict_param.keys())
redacteur.writeheader()
redacteur.writerow(dict_param)
elif estNormal == "MULTI_PARAM":
i = 0
dict_param = {}
for ligne in range(longueur_totale - param_lignes + 1, longueur_totale - 1):
chaine_param[i] = lignes[ligne]
dict_param[i] = json.loads(chaine_param[i])
dict_param.update(dict_param[i])
i += 1
with open(file=fichier_sortie, mode='a', newline='', encoding='utf-8-sig') as f:
redacteur = csv.DictWriter(f, fieldnames=dict_param.keys())
redacteur.writeheader()
redacteur.writerow(dict_param)
elif estNormal == "SANS_PARAM":
return True
- Code complet
Les sections ci-dessus sont combinées en une fonction dat_vers_csv(), avec ajout d'un traitement par lots pour les fichiers. Le code complet est le suivant :
import csv
import os
import json
repertoire_entree = 'D:\\Desktop\\donnees_eeg\\dat\\'
repertoire_sortie = 'D:\\Desktop\\donnees_eeg\\csv\\'
def dat_vers_csv(fichier_entree, fichier_sortie):
"""
Fonction principale de conversion .dat vers .csv
:param fichier_entree: nom complet du fichier d'entrée
:param fichier_sortie: nom complet du fichier de sortie
:return:
"""
fr = open(fichier_entree, encoding='utf-8-sig', errors='ignore')
lignes = fr.readlines()
# Lecture de l'en-tête
chaine_en_tete = lignes[0] # Extraction de la chaîne au format dictionnaire
dict_en_tete = json.loads(chaine_en_tete)
with open(file=fichier_sortie, mode='w', newline='', encoding='utf-8-sig') as f:
redacteur = csv.DictWriter(f, fieldnames=dict_en_tete.keys())
redacteur.writeheader()
redacteur.writerow(dict_en_tete)
# Lecture de la section de données
# Après la section de données, il y a au maximum deux lignes de paramètres
# Par conséquent, nous devons calculer spécifiquement la longueur de la section de données
longueur_totale = len(lignes)
estNormal = "NORMAL"
# Calcul du nombre de lignes de paramètres
param_lignes = 0 # Initialisation du compteur
for indice in range(0, longueur_totale):
if len(lignes[indice]) > 100:
param_lignes += 1
# Détermination du type de données
longueur_donnees = 0 # Initialisation de la longueur de la section de données
if param_lignes >= 3: # Vérification si l'avant-dernière ligne est un paramètre
longueur_donnees = longueur_totale - param_lignes + 1
estNormal = "MULTI_PARAM"
elif param_lignes == 1:
longueur_donnees = longueur_totale
estNormal = "SANS_PARAM"
elif param_lignes == 2:
longueur_donnees = longueur_totale - param_lignes + 1
estNormal = "NORMAL"
for ligne in range(1, longueur_donnees):
ligne_donnees = lignes[ligne]
liste_chaine = ligne_donnees.split()
liste_float = list(map(float, liste_chaine))
with open(file=fichier_sortie, mode='a', newline='') as f:
redacteur = csv.writer(f)
redacteur.writerow(liste_float)
# Lecture de la section de paramètres
chaine_param = {} # Stockage des chaînes au format dictionnaire
dict_param = {} # Stockage des dictionnaires
if estNormal == "NORMAL":
chaine_param = lignes[longueur_totale - 1] # Extraction de la chaîne au format dictionnaire
dict_param = json.loads(chaine_param)
with open(file=fichier_sortie, mode='a', newline='', encoding='utf-8-sig') as f:
redacteur = csv.DictWriter(f, fieldnames=dict_param.keys())
redacteur.writeheader()
redacteur.writerow(dict_param)
elif estNormal == "MULTI_PARAM":
i = 0
dict_param = {}
for ligne in range(longueur_totale - param_lignes + 1, longueur_totale - 1):
chaine_param[i] = lignes[ligne]
dict_param[i] = json.loads(chaine_param[i])
dict_param.update(dict_param[i])
i += 1
with open(file=fichier_sortie, mode='a', newline='', encoding='utf-8-sig') as f:
redacteur = csv.DictWriter(f, fieldnames=dict_param.keys())
redacteur.writeheader()
redacteur.writerow(dict_param)
elif estNormal == "SANS_PARAM":
return True
def traitement_par_fichiers(repertoire):
"""
Fonction de traitement par lots pour la conversion .dat vers .csv
:param repertoire: chemin du répertoire des fichiers
:return:
"""
liste_fichiers = os.listdir(repertoire)
print(liste_fichiers)
# fichier: nom du fichier
for fichier in liste_fichiers:
# Définition du chemin d'accès au fichier d'entrée
print(fichier)
fichier_entree = repertoire_entree + fichier
# Définition du nom du fichier de sortie
nom_fichier_sortie = fichier[:22] + '.csv'
# Définition du chemin d'accès au fichier de sortie
fichier_sortie = repertoire_sortie + nom_fichier_sortie
dat_vers_csv(fichier_entree, fichier_sortie)
traitement_par_fichiers(repertoire_entree)
L'exemple de fichier d'entrée .dat (les informations sensibles ont été supprimées, seules quelques données initiales de mode veille sont affichées) :
L'exemple de fichier de sortie .csv (seules quelques données sont affichées, les informations sensibles ne sont pas montrées) :
Cette solution a été développée pour répondre à un besoin spécifique de traitement de données EEG. L'approche peut être adaptée à d'autres formats de données similaires présentant une structure en trois parties distinctes.