#!/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 2020 - Temps-réel archive
"""
from datetime import datetime as dt
import functools
import os.path
import pandas as pnd
from pyspc.convention.grp20 import (
RT_ARCHIVE_DATEFORMAT, RT_ARCHIVE_DATESHORTFORMAT,
RT_ARCHIVE_VARNAMES, RT_ARCHIVE_HEADERS)
import pyspc.core.exception as _exception
def date_parser(y, x):
"""
Analyseur des dates des fichiers Archives de GRP Temps-Réel
Parameters
----------
y : int
Année
x : str
Date sous forme de texte mmjjHHMM
Returns
-------
datetime.datetime
Date de la donnée
"""
return dt.strptime(f'{y:04d}{x}', RT_ARCHIVE_DATEFORMAT)
[docs]
class GRPRT_Archive():
"""
Structure de données GRPRT Archive (GRP *Temps Réel*)
- PV_jjJhhHmmM_aaaa.DAT
- QV_aaaa.DAT
Attributes
----------
filename : str
Nom du fichier Archve de GRP *Temps-Réel*
varname : str
Variable
timestep : str, None
Pas de temps de la variable
year : int
Année de l'archive
"""
[docs]
def __init__(self, filename=None):
"""
Initialisation de l'instance de la classe GRPRT_Archive
Parameters
----------
filename : str
Nom du fichier Archve de GRP *Temps-Réel*
"""
self.filename = filename
if filename is not None:
infos = self.split_basename(filename=self.filename)
self.varname = infos[0]
self.timestep = infos[1]
self.year = infos[2]
self._check_varname(varname=self.varname)
def __str__(self):
"""
Afficher les méta-données de l'instance GRPRT_Archive
"""
text = """
*************************************
*********** GRP 2020 - RT Archive ***
*************************************
* NOM FICHIER = {filename}
* NOM VARIABLE = {varname}
* PAS DE TEMPS = {timestep}
* ANNEE = {year}
*************************************
"""
return text.format(**vars(self))
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):
"""
Lecture du fichier GRPRT_Archive
Returns
-------
pandas.DataFrame
Tableau des données d'archives de GRP Temps-Réel
Examples
--------
>>> from pyspc.model.grp20 import GRPRT_Archive
>>> f = 'data/model/grp20/rt/PV_00J01H00M_2007.DAT'
>>> reader = GRP_Data(filename=f)
>>> reader
*************************************
*********** GRP 2020 - RT Archive ***
*************************************
* NOM FICHIER = data/model/grp20/rt/PV_00J01H00M_2007.DAT
* NOM VARIABLE = PV
* PAS DE TEMPS = 00J01H00M
* ANNEE = 2007
*************************************
>>> df = reader.read()
>>> df
90065003
Date(TU)
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
2007-01-19 13:00:00 2.2
2007-01-19 14:00:00 1.6
2007-01-19 15:00:00 3.2
2007-01-19 16:00:00 1.8
2007-01-19 17:00:00 1.2
2007-01-19 18:00:00 0.8
>>> f = 'data/model/grp20/rt/QV_2007.DAT'
>>> reader = GRP_Data(filename=f)
>>> reader
*************************************
*********** GRP 2020 - RT Archive ***
*************************************
* NOM FICHIER = data/model/grp20/rt/QV_2007.DAT
* NOM VARIABLE = QV
* PAS DE TEMPS = None
* ANNEE = 2007
*************************************
>>> df = reader.read()
>>> df
RH10585x
Date(TU)
2007-01-18 18:00:00 3.950000
2007-01-18 18:08:00 4.190000
2007-01-18 18:30:00 5.110000
2007-01-18 18:54:00 6.210000
2007-01-18 19:00:00 6.560000
2007-01-18 19:30:00 7.920000
2007-01-18 22:00:00 13.600000
2007-01-18 22:03:00 13.700000
2007-01-18 23:00:00 15.800000
2007-01-18 23:30:00 16.700000
2007-01-19 00:00:00 17.300000
2007-01-19 01:00:00 19.200000
2007-01-19 01:30:00 19.800000
2007-01-19 02:45:00 20.400000
2007-01-19 03:24:00 21.500000
2007-01-19 04:00:00 23.000000
2007-01-19 04:30:00 23.600000
2007-01-19 05:30:00 23.400000
2007-01-19 06:00:00 23.600000
2007-01-19 06:30:00 24.700000
2007-01-19 07:30:00 28.600000
2007-01-19 07:45:00 30.600000
2007-01-19 08:30:00 35.100000
2007-01-19 10:00:00 36.600000
2007-01-19 11:30:00 35.900000
2007-01-19 12:15:00 32.099998
2007-01-19 12:49:00 28.600000
2007-01-19 13:30:00 26.700000
2007-01-19 14:15:00 25.300000
2007-01-19 15:30:00 23.500000
2007-01-19 16:30:00 22.800000
2007-01-19 18:00:00 20.800000
"""
with open(self.filename, 'r', encoding='iso-8859-1') as f:
f.readline()
infos = f.readline().strip()
station = infos.split(' ')[-1]
# Lecture des données
df = pnd.read_csv(
self.filename,
encoding='iso-8859-1',
sep=';',
header=0,
skiprows=4,
index_col=0,
converters={'Date(TU)': str, 'Débit (L/s) ': float,
'Pluie (mm) ': float, 'Temp. (°C) ': float},
decimal=','
)
df.index = df.index.map(functools.partial(date_parser, self.year))
# Nettoyer les intitulés des colonnes
df.columns = [station]
df.index.name = df.index.name.strip()
# Unités
if self.varname.startswith('Q'):
try:
df = df.map(lambda x: x / 1000.)
except TypeError:
_exception.Warning(
None,
"Impossible d'appliquer le changement d'unité. "
" Veuillez vérifier les entêtes de colonnes comme suit. "
"'QV': 'Débit (L/s) ', "
"'PV': 'Pluie (mm) ', "
"'TV': 'Temp. (°C) '.")
return df
[docs]
def write(self, data=None, dirname='.', varname=None, timestep=None,
year=None):
"""
Ecriture du fichier GRPRT_Archive
Parameters
----------
data : pandas.DataFrame
Tableau des données annuelles
varname : str
Nom de la variable
timestep : str, None
Pas de temps de la variable
year : int
Année de l'archive
"""
_exception.check_dataframe(data)
_exception.raise_valueerror(
len(data.columns) != 1,
"Incompatible avec un DataFrame avec plusieurs colonnes."
)
mni = min(data.index)
mxi = max(data.index)
_exception.raise_valueerror(
mni < dt(year, 1, 1) or mni >= dt(year+1, 1, 1)
or mxi < dt(year, 1, 1) or mxi >= dt(year+1, 1, 1),
"Incompatibilité entre les données et l'année."
)
_exception.check_str(dirname)
self.filename = os.path.join(
dirname,
self.join_basename(varname=varname, timestep=timestep, year=year)
)
self.varname = varname
self._check_varname(varname=self.varname)
self.timestep = timestep
self.year = year
station = data.columns[0]
header = RT_ARCHIVE_HEADERS[self.varname]
col = f"{header:12s}"
data.columns = [col]
data.index = [x.strftime(RT_ARCHIVE_DATESHORTFORMAT)
for x in data.index]
if self.varname.startswith('Q'):
try:
data = data.map(lambda x: x * 1000.)
except ValueError:
pass
data[col] = data[col].apply(lambda x: f"{x:12.4f}")
data.index.name = 'Date(TU)'
with open(self.filename, 'w', encoding='iso-8859-1',
newline='\r\n') as fopen:
fopen.write("#----------------------------------\n")
fopen.write(f"Fichier de données {self.year} de {header} pour la station {station}\n")
fopen.write("MMJJHHMM;FFFFFFF.FFFF\n")
fopen.write("#----------------------------------\n")
return data.to_csv(self.filename, mode='a', sep=';',
encoding='iso-8859-1')
[docs]
@classmethod
def get_varnames(cls):
"""
Définir le nom de la variable
"""
return sorted(RT_ARCHIVE_VARNAMES)
[docs]
@staticmethod
def split_basename(filename=None):
"""
Extraire les informations depuis le nom du fichier
de données GRP Archive (GRP *Temps-Réel*)
Parameters
----------
filename : str
Fichier de données GRP Archive (GRP *Temps-Réel*)
Returns
-------
varname : str
Nom de la variable
timestep : str, None
Pas de temps de la variable
year : int
Année de l'archive
Examples
--------
>>> from pyspc.model.grp20 import GRPRT_Archive
Cas d'une grandeur sans pas de temps
>>> f = 'data/model/grp20/rt/QV_2007.DAT'
>>> varname, timestep, year = GRPRT_Archive.split_basename(filename=f)
>>> varname
QV
>>> timestep
None
>>> year
2007
Cas d'une grandeur avec un pas de temps
>>> f = 'data/model/grp20/rt/PV_00J01H00M_2007.DAT'
>>> varname, timestep, year = GRPRT_Archive.split_basename(filename=f)
>>> varname
PV
>>> timestep
00J01H00
>>> year
2007
"""
if filename is None:
return None, None, None
infos = os.path.splitext(os.path.basename(filename))[0].split('_')
varname = infos.pop(0)
year = int(float(infos.pop(-1)))
try:
timestep = infos.pop(0)
except IndexError:
timestep = None
return varname, timestep, year
[docs]
@staticmethod
def join_basename(varname=None, timestep=None, year=None):
"""
Définir le nom du fichier de données GRP RT Archive
à partir des informations
Parameters
----------
varname : str
Nom de la variable
timestep : str, None
Pas de temps de la variable
year : int
Année de l'archive
Returns
-------
filename : str
Fichier de données GRP Archive (GRP *Temps-Réel*)
Examples
"""
_exception.check_str(varname)
_exception.check_int(year)
if timestep is None:
return f"{varname}_{year}.DAT"
_exception.check_str(timestep)
return f"{varname}_{timestep}_{year}.DAT"