#!/usr/bin/python3
# -*- coding: utf-8 -*-
########################################################################
#
# This file is part of python module <pyspc>.
# Copyright (C) 2013-2021 R. Marty
# (renaud.marty@developpement-durable.gouv.fr)
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program (see COPYING.txt).
# If not, see <http://www.gnu.org/licenses/>.
#
########################################################################
"""
Modélisations hydrologiques - GRP version 2022 - Observations
"""
from datetime import timedelta as td
import os.path
import pandas as pnd
import pyspc.core.exception as _exception
from pyspc.convention.grp22 import CAL_DATA_HEADERS, CAL_DATA_VARNAMES
from pyspc.core.timeutil import dtfmt, dtheader
from pyspc.model.grp22.tdelta import td2str, str2td
[docs]
class GRP_Data():
"""
Structure de données GRP Data (GRP *Calage*)
Attributes
----------
filename : str
Nom du fichier de données
station : str
Nom de la station
varname : str
Nom de la variable
timestep : str, None
Pas de temps de la variable
tdelta : timedelta, None
Pas de temps de la variable
"""
[docs]
def __init__(self, filename=None):
"""
Initialisation de l'instance de la classe GRP_Data
Parameters
----------
filename : str
Nom du fichier de données
"""
self.filename = filename
if self.filename is not None:
info = self.split_basename(self.filename)
self.station = info[0]
self.varname = info[1]
self.timestep = info[2]
self._check_varname(varname=self.varname)
else:
self.station = None
self.varname = None
self.timestep = None
self.tdelta = str2td(self.timestep)
def __str__(self):
"""
Afficher les méta-données de l'instance GRP_Data
"""
text = """
*************************************
*********** GRP 2022 - Data *********
*************************************
* NOM FICHIER = {filename}
* CODE STATION = {station}
* NOM VARIABLE = {varname}
* PAS DE TEMPS = {timestep}
* PAS DE TEMPS = {tdelta}
*************************************
"""
return text.format(**vars(self))
def _check_dataframe(self, df=None, strict=False):
"""
Contrôler le tableau de données est correctement formaté
Parameters
----------
df : pandas.DataFrame
Tableau des données d'observation de GRP 2022
strict : bool
Lecture avec vérification de la variable (défaut: False)
Raises
------
ValueError
Si la variable n'est pas reconnue par pyspc
"""
_exception.raise_valueerror(
df.index.name != dtheader(self.tdelta),
f"Entête de colonne mal-formatée : {df.index.name} != {dtheader(self.tdelta)}",
strict
)
_exception.raise_valueerror(
len([c for c in df.columns if c != CAL_DATA_HEADERS[self.varname]])
!= 0,
f"Entête de colonne mal-formatée : {df.columns}", strict)
def _check_varname(self, varname=None):
"""
Contrôler la variable
Parameters
----------
varname : str
Nom de la variable.
Raises
------
ValueError
Si la variable n'est pas reconnue par pyspc
"""
if varname not in self.get_varnames():
raise ValueError("Variable mal renseignée")
[docs]
def read(self, strict=False):
"""
Lecture du fichier de données GRP Data (GRP *Calage*)
Parameters
----------
strict : bool
Lecture avec vérification de la variable (défaut: False)
Returns
-------
pandas.DataFrame
Tableau des données d'observation de GRP 2022
Examples
--------
>>> from pyspc.model.grp22 import GRP_Data
Cas d'une grandeur sans pas de temps
>>> f = 'data/model/grp22/cal/RH10585x_Q.txt'
>>> reader = GRP_Data(filename=f)
>>> df = reader.read()
>>> df
Q(m3/s)
AAAAMMJJHHMM
2007-01-18 19:00:00 6.56
2007-01-18 19:30:00 7.92
2007-01-18 22:00:00 13.60
2007-01-18 22:03:00 13.70
2007-01-18 23:00:00 15.80
2007-01-18 23:30:00 16.70
2007-01-19 00:00:00 17.30
2007-01-19 01:00:00 19.20
2007-01-19 01:30:00 19.80
2007-01-19 02:45:00 20.40
2007-01-19 03:24:00 21.50
2007-01-19 04:00:00 23.00
2007-01-19 04:30:00 23.60
2007-01-19 05:30:00 23.40
2007-01-19 06:00:00 23.60
2007-01-19 06:30:00 24.70
2007-01-19 07:30:00 28.60
2007-01-19 07:45:00 30.60
2007-01-19 08:30:00 35.10
2007-01-19 10:00:00 36.60
2007-01-19 11:30:00 35.90
2007-01-19 12:15:00 32.10
2007-01-19 12:49:00 28.60
2007-01-19 13:30:00 26.70
2007-01-19 14:15:00 25.30
2007-01-19 15:30:00 23.50
2007-01-19 16:30:00 22.80
2007-01-19 18:00:00 20.80
2007-01-19 18:30:00 20.40
2007-01-19 19:30:00 19.20
2007-01-19 20:00:00 19.30
2007-01-19 20:30:00 19.10
2007-01-19 21:00:00 18.60
2007-01-19 22:00:00 18.80
2007-01-19 23:00:00 17.90
Cas d'une grandeur avec pas de temps
>>> f = 'data/model/grp22/cal/90065003_P_00J01H00M.txt'
>>> reader = GRP_Data(filename=f)
>>> df = reader.read()
>>> df
P(mm)
AAAAMMJJHH
2007-01-18 12:00:00 0.8
2007-01-18 13:00:00 0.6
2007-01-18 14:00:00 1.0
2007-01-18 15:00:00 4.8
2007-01-18 16:00:00 7.2
2007-01-18 17:00:00 8.0
2007-01-18 18:00:00 11.2
2007-01-18 19:00:00 7.4
2007-01-18 20:00:00 7.6
2007-01-18 21:00:00 6.8
2007-01-18 22:00:00 8.8
2007-01-18 23:00:00 7.6
2007-01-19 00:00:00 5.8
2007-01-19 01:00:00 7.8
2007-01-19 02:00:00 8.2
2007-01-19 03:00:00 5.2
2007-01-19 04:00:00 6.6
2007-01-19 05:00:00 7.2
2007-01-19 06:00:00 9.0
2007-01-19 07:00:00 6.0
2007-01-19 08:00:00 5.0
2007-01-19 09:00:00 3.6
2007-01-19 10:00:00 2.8
2007-01-19 11:00:00 2.0
2007-01-19 12:00:00 1.8
"""
df = pnd.read_csv(
self.filename,
sep=';',
header=0,
index_col=0,
na_values=[-99.9, -999.999],
keep_default_na=True,
parse_dates=True,
date_format=dtfmt(self.tdelta)
)
self._check_dataframe(df=df, strict=strict)
return df
[docs]
def write(self, data=None, dirname='.', strict=False):
"""
Ecrire le fichier de données GRP Data (GRP *Calage*)
Parameters
----------
data : pandas.DataFrame
Tableau des données d'observation de GRP 2022
dirname : str
Répertoire local
strict : bool
Lecture avec vérification de la variable (défaut: False)
"""
_exception.check_dataframe(data)
self._check_dataframe(df=data, strict=strict)
if self.filename is None:
self.filename = self.join_basename(
station=self.station, varname=self.varname,
timestep=self.timestep)
self.filename = os.path.join(dirname, self.filename)
return data.to_csv(
self.filename,
sep=';',
float_format='%.3f',
index=True,
date_format=dtfmt(self.tdelta),
na_rep='-99.999',
lineterminator='\n'
)
[docs]
@classmethod
def get_varnames(cls):
"""
Définir le nom de la variable
"""
return sorted(CAL_DATA_VARNAMES)
[docs]
@staticmethod
def split_basename(filename=None):
"""
Extraire les informations depuis le nom du fichier
de données GRP Data (GRP *Calage*)
Parameters
----------
filename : str
Fichier de données GRP Data (GRP *Calage*)
Returns
-------
station : str
Identifiant de la station
varname : str
Nom de la variable
timestep : str, None
Pas de temps de la variable
Examples
--------
>>> from pyspc.model.grp22 import GRP_Data
Cas d'une grandeur sans pas de temps
>>> f = 'data/model/grp22/cal/RH10585x_Q.txt'
>>> [station, varname, timestep] = GRP_Data.split_basename(filename=f)
>>> station
RH10585
>>> varname
Q
>>> timestep
None
Cas d'une grandeur avec pas de temps
>>> f = 'data/model/grp22/cal/90065003_P_00J01H00M.txt'
>>> [station, varname, timestep] = GRP_Data.split_basename(filename=f)
>>> station
90065003
>>> varname
P
>>> timestep
00J01H00M
"""
if filename is None:
return None, None, None
basename = os.path.splitext(os.path.basename(filename))[0]
try:
info = basename.split('_')
station = info.pop(0)
varname = info.pop(0)
except (IndexError, ValueError) as ive:
raise ValueError("Le nom de fichier ne respecte pas le "
"nommage de GRP") from ive
try:
timestep = info.pop(0)
except IndexError:
timestep = None
return station, varname, timestep
[docs]
@staticmethod
def join_basename(station=None, varname=None, timestep=None):
"""
Définir le nom du fichier de données GRP Data (GRP *Calage*)
à partir des informations
Parameters
----------
station : str
Identifiant de la station
varname : str
Nom de la variable
timestep : timedelta, str, None
Pas de temps de la variable
Returns
-------
filename : str
Fichier de données GRP Data (GRP *Calage*)
"""
if station is None or varname is None:
raise ValueError('Définition incorrecte des arguments')
if timestep is None:
return f'{station}_{varname}.txt'
if isinstance(timestep, td):
timestep = td2str(timestep)
_exception.check_str(timestep)
return f'{station}_{varname}_{timestep}.txt'