# -*- coding: utf-8 -*-
"""
Bibliothèque pyspc du projet pyspc - IO - Octave - read
@author: m.charpentier-noyer
"""
import re
import pyspc.core.exception as _exception
from pyspc.core.serie import Serie
from pyspc.core.series import Series
from pyspc.data.octave.data import Octave_Data
from pyspc.convention.octave import RATIOS
from pyspc.convention.octave import OCTAVE_VARNAMES as OCT_VAR
def _compute_timestep(index):
if len(index) > 1:
return index[1] - index[0]
return None
def _identify_code(part):
if re.match(r'^[A-Z]?[0-9]{7,9}$', part): # Code PHyC
return 'Code PHyC', part
if re.match(r'^[A-Z]{2}[0-9]{1,6}$', part): # Code LAMEDO
return 'Code LAMEDO', part
return None, None
[docs]
def read_Octave_Data(filename: str = None, codes: list = None,
selected_headers: list = None, warning: bool = True
) -> object:
"""
Créer une instance Series à partir d'un csv exporté depuis Octave.
Parameters
----------
filename : str
Nom du fichier exporté depuis Octave.
codes : list
Liste des identifiants des stations à conserver.
Par défaut, toutes les stations sont retenues.
warning : bool
Imprimer les erreurs ?
selected_headers : list (optionnel)
Liste des entêtes à conserver.
Returns
-------
series : pyspc.core.series.Series
Collection de séries de données.
Warnings
--------
See Also
--------
pyspc.data.octave.Octave_Data
Examples
--------
>>> from pyspc.io.octave import read_octave_data
>>> import os
>>> f = 'test/data/data/octave/Octave_Multiseries.csv'
>>> series = read_octave_data(filename=f, selected_headers=None)
>>> series
*************************************
********** SERIES *******************
*************************************
* NOM DE LA COLLECTION = octave_data
* TYPE DE COLLECTION = obs
* NOMBRE DE SERIES = 2
* ----------------------------------
* SERIE #1
* - CODE = K0403010
* - VARNAME = QH
* - META = LO12844_Chambon-sur-Lignon [Le Lignon]_SerieType.Instantaneous_m3/s
* ----------------------------------
* SERIE #2
* - CODE = K0403020
* - VARNAME = QH
* - META = LO15760_Vastres - Pont Marie [Le Lignon]_SerieType.Instantaneous_m3/s
*************************************
"""
# -------------------------------------------------------------------------
# 0- Contrôles
# -------------------------------------------------------------------------
_exception.check_str(filename)
if codes is not None:
_exception.check_listlike(codes)
_exception.check_bool(warning)
# -------------------------------------------------------------------------
# 1- Lecture
# -------------------------------------------------------------------------
# Création du lecteur
reader = Octave_Data(filename=filename)
df = reader.read(selected_headers=selected_headers)
# -------------------------------------------------------------------------
# 2- Conversion en Series
# -------------------------------------------------------------------------
series = Series(datatype='obs', name='octave_data')
for c in df.columns:
d = df[c].to_frame()
try:
if not isinstance(c, str):
if warning:
_exception.Warning(
msg="Nom de colonne n'est pas une chaine de "
f"caractères : {c}")
continue
parts = c.split('_')
code, meta, detected_headers, varname = _set_cd_meta_dh_var(
c, parts)
unit = detected_headers.get('Unité')
unit_key = (varname, unit)
if unit_key in RATIOS:
ratio = RATIOS[unit_key]
if ratio != 1:
_exception.Warning(
msg="Changement d'unité pour normalisation car "
f"{(unit_key)}")
d = d * ratio
else:
_exception.Warning(
msg="Pas d'unité standardisée retrouvée")
meta = ' | '.join(meta)
if varname == 'HA':
varname_key = ('octave', ('HA', None))
# obligatoire car timestep de 24h calculé sinon
else:
timestep = _compute_timestep(d.index)
varname_key = ('octave', (varname, timestep))
_exception.raise_valueerror(varname_key not in OCT_VAR,
"Aucune grandeur trouvée")
varname = OCT_VAR[varname_key]
serie = Serie(d, code=code, provider='octave_data',
varname=varname)
series.add(serie=serie, meta=meta)
except Exception as e:
_exception.Warning(msg="Pas de code extrait de la colonne"
f" '{c}': {e}")
serie = Serie(d, code=None, provider='octave_data', varname=c)
series.add(serie=serie, meta=None)
return series
def _set_cd_meta_dh_var(c, parts):
"""Définir code, meta-données, entête et grandeur."""
code = None
varname = None
meta = []
detected_headers = {}
for part in parts:
header, value = _identify_code(part)
if header and value not in detected_headers:
detected_headers[header] = value
elif part in [v[1][0] for v in OCT_VAR.keys()]:
detected_headers['Grandeur'] = part
elif part in [v[1] for v in RATIOS.keys()]:
detected_headers['Unité'] = part
else:
meta.append(part)
code_phyC = detected_headers.get('Code PHyC')
if code_phyC in ["Z9999999", "000000000"]:
_exception.Warning(
msg="Code par défaut, il ne s'agit pas du réel "
f"code PHyC ({code_phyC})")
code_lamedo = detected_headers.get('Code LAMEDO')
if code_phyC and code_lamedo:
code = code_phyC
meta.append(code_lamedo)
elif code_phyC:
code = code_phyC
elif code_lamedo:
code = code_lamedo
else:
code = None
_exception.Warning(
msg=f"Pas de code extrait de la colonne '{c}'")
varname = detected_headers.get('Grandeur')
return code, meta, detected_headers, varname