Code source de pyspc.data.octave.data
# -*- coding: utf-8 -*-
"""
Données exportées (en colonnes) depuis Octave - Données observées/calculées.
Type de données lisibles :
Débit
Hauteur
HU/SIM (depuis LAMEDO)
pluvio
@author: m.charpentier-noyer
"""
import pandas as pd
import re
from pyspc.convention.octave import HEADERS as HD
[docs]
class Octave_Data:
"""Structure de données exportées depuis Octave (données en colonnes)."""
[docs]
def __init__(self, filename=None):
"""
Initialisation de l'instance de la classe Octave_Data.
Parameters
----------
filename : str
Fichier à lire
"""
self.filename = filename
self.data = None
self.headers_info = {}
def __str__(self):
"""Afficher les méta-données de l'instance Octave_Data."""
text = """
****************************************
*********** OCTAVE - DATA **************
****************************************
* NOM FICHIER = {filename}
****************************************
"""
return text.format(**vars(self))
[docs]
def detect_headers(self):
"""Détecte la présence et l'emplacement des lignes clés (HEADERS)."""
with open(self.filename, "r", encoding="utf-8") as file:
lines = file.readlines()
self.headers_info.clear() # Réinitialiser avant analyse
for i, line in enumerate(lines):
for key in HD:
if line.startswith(key):
# Stocke la position de l'en-tête
self.headers_info[key] = i
return self.headers_info
[docs]
def extract_headers(self, selected_headers=None):
"""
Construit dynamiquement les noms de colonnes.
En fonction des lignes d'entête disponibles.
"""
if selected_headers is None:
# Prend tous les en-têtes détectés
selected_headers = list(self.headers_info.keys())
# Vérification des en-têtes valides
valid_headers = [h for h in selected_headers if h in self.headers_info]
if not valid_headers:
raise ValueError(
"Aucun en-tête valide fourni dans selected_headers.")
with open(self.filename, "r", encoding="utf-8") as file:
lines = file.readlines()
headers_data = {h: lines[self.headers_info[h]].strip().split(";")[1:]
for h in valid_headers
if self.headers_info[h] < len(lines)}
new_headers = []
# Itération sur les colonnes
for i in range(len(headers_data.get(valid_headers[0], []))):
parts = [headers_data[h][i]
for h in valid_headers
if i < len(headers_data[h]) and headers_data[h][i]]
new_headers.append("_".join(parts))
if not new_headers:
new_headers = ['Code_LAMEDO']
# Vérification des doublons
duplicates = {header for header in new_headers
if new_headers.count(header) > 1}
if duplicates:
print("Avertissement : Les en-têtes suivants sont dupliqués : "
f"{', '.join(duplicates)}")
return new_headers
[docs]
def detect_header_line(self):
"""
Trouve la première ligne contenant une date.
Pour identifier où commencent les données.
"""
date_pattern = re.compile(r"\b\d{2}/\d{2}/\d{4}\b")
with open(self.filename, "r", encoding="utf-8") as file:
for i, line in enumerate(file):
if date_pattern.search(line):
return i - 1 # Retourne la ligne juste avant la date
raise ValueError(
"Impossible de détecter la ligne d'en-tête contenant une date.")
[docs]
def read(self, selected_headers: object = None) -> object:
"""
Lecture du fichier de données Octave avec sélection des en-têtes.
Examples
--------
>>> from pyspc.data.octave.data import Octave_Data
>>> f = 'data/data/octave/Octave_Multiseries.csv'
>>> reader = Octave_Data(filename=f)
>>> content = reader.read(
... selected_headers=["Code PHyC", "Grandeur", "Unité"])
>>> content
K0403010_Q_m3/s K0403020_Q_L/s
Date
2024-10-16 12:00:00 19.2536 11.7147
2024-10-16 13:00:00 26.3926 13.9217
2024-10-16 14:00:00 33.9680 15.5506
2024-10-16 15:00:00 41.0813 14.1870
2024-10-16 16:00:00 44.5392 12.0414
2024-10-16 17:00:00 44.4443 10.6963
2024-10-16 18:00:00 40.2565 9.7104
2024-10-16 19:00:00 36.4109 9.4937
2024-10-16 20:00:00 34.0360 9.8263
2024-10-16 21:00:00 34.4067 10.5495
2024-10-16 22:00:00 35.1157 12.9860
2024-10-16 23:00:00 46.2170 15.1426
2024-10-17 00:00:00 58.1265 18.7350
2024-10-17 01:00:00 73.4513 34.7830
2024-10-17 02:00:00 95.9750 53.0143
2024-10-17 03:00:00 144.7872 49.6830
2024-10-17 04:00:00 211.3700 56.2290
2024-10-17 05:00:00 330.7380 66.9773
2024-10-17 06:00:00 395.4790 74.0440
2024-10-17 07:00:00 481.5055 125.9880
2024-10-17 08:00:00 493.4020 193.9770
2024-10-17 09:00:00 749.4350 188.8860
2024-10-17 10:00:00 737.5685 120.4700
2024-10-17 11:00:00 485.2950 91.3350
2024-10-17 12:00:00 324.6220 66.2190
2024-10-17 13:00:00 241.1575 49.9410
2024-10-17 14:00:00 185.9420 42.7760
2024-10-17 15:00:00 145.3445 100.6320
2024-10-17 16:00:00 142.1030 82.1240
2024-10-17 17:00:00 298.6280 61.0660
2024-10-17 18:00:00 212.4260 61.9435
2024-10-17 19:00:00 160.6010 63.6200
2024-10-17 20:00:00 150.6060 52.1080
2024-10-17 21:00:00 152.4625 54.7443
2024-10-17 22:00:00 151.6190 59.0720
2024-10-17 23:00:00 173.2800 42.7044
2024-10-18 00:00:00 156.1705 31.3290
2024-10-18 01:00:00 124.1383 25.1735
2024-10-18 02:00:00 103.0916 24.6830
2024-10-18 03:00:00 89.6756 23.0458
2024-10-18 04:00:00 83.1818 19.6820
2024-10-18 05:00:00 78.9327 17.5213
2024-10-18 06:00:00 73.4926 17.7100
"""
try:
self.detect_headers()
header_line = self.detect_header_line()
new_headers = self.extract_headers(selected_headers)
df = pd.read_csv(
self.filename,
sep=';',
skiprows=header_line,
index_col=0,
decimal='.',
na_values=['', 'NA', 'null']
)
# df.index = pd.to_datetime(df.index)
df.index = pd.to_datetime(df.index, dayfirst=True)
df.index.name = "Date"
df.columns = new_headers
return df
except Exception as e:
print(f"Erreur lors de la lecture du fichier : {e}")
return None