#!/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 2016 - Observations
"""
import collections
import itertools
import os.path
import textwrap
import pyspc.core.exception as _exception
from pyspc.convention.grp16 import (
CAL_CONFIG_CALKEYS, CAL_CONFIG_CONVERTERS, CAL_CONFIG_FORMAT,
CAL_CONFIG_HEADER, CAL_CONFIG_NAMES
)
GRP_Run = collections.namedtuple('GRP_Run', CAL_CONFIG_NAMES)
GRP_Run.__doc__ = """Run de GRP
Attributes
----------
NB : int
Numéro du bassin
CODE : str
Identifiant du bassin
NOM : str
Libellé bassin
ST : int
Sans module neige + Correction de Tangara
SR : int
Sans module neige + Correction des sorties par réseau de neurones
AT : int
Avec module neige + Correction de Tangara
AR : int
Avec module neige + Correction des sorties par réseau de neurones
SURFACE : float
Superficie du bassin en km2
RT : str
Référentiel Temporel (TU, HL ou HH)
HOR : int
Horizon de calage
SC : float
Seuil de calage en m3/s
DEB : str
Date debut
FIN : str
Date fin
SV : float
Seuil utilise pour l'analyse des résultat, en m3/s
NJ : int
Nombre de jours pour le tracé des hydrogrammes prevus
HC : int
Horizon des cheveux sur le tracé des hydrogrammes prevus
EC : int
Ecart en heure entre les cheveux des previsions (de 01 a 72 heures)
ECART : int
Ecart (en m3/s) entre prevision et observation
INC : int
0 ou 1, pour activer ou desactiver la generation d'abaques d'incertitudes
TGR : int
0 ou 1, pour tracer les eventuels resultats de TGR
"""
[docs]
class GRP_Cfg(list):
"""
Structure de données GRP Cfg : liste de runs de GRP
Attributes
----------
filename : str
Fichier **LISTE_BASSINS.DAT** de GRP *Calage*
..seealso:: GRP_Run
"""
[docs]
def __init__(self, filename=None):
"""
Initialisation de l'instance de la classe GRP_Cfg
Parameters
----------
filename : str
Fichier **LISTE_BASSINS.DAT** de GRP *Calage*
"""
super().__init__()
if filename is None:
raise ValueError('Nom de fichier inconnu')
self.filename = filename
def __str__(self):
"""
Afficher les méta-données de l'instance GRP_Cfg
"""
if self:
strdata = '\n'
for run in self:
strdata += ' ==> '
strdata += CAL_CONFIG_FORMAT.format(**run._asdict())
else:
strdata = ''
text = """
*************************************
*********** GRP 2016 - Config *******
*************************************
* NOM FICHIER = {filename}
* RUNS {strdata}
*************************************
"""
return text.format(filename=self.filename, strdata=strdata)
[docs]
@staticmethod
def check_run(run):
"""
Liste des valeurs correspondant aux paramètres de calage de GRP
"""
if not isinstance(run, GRP_Run):
raise ValueError("Le run n'est pas un GRP_Run")
[docs]
def get_calibrationvalues(self):
"""
Liste des valeurs correspondant aux paramètres de calage de GRP
"""
return [tuple([r._asdict()[k] for k in CAL_CONFIG_CALKEYS])
for r in self]
[docs]
def read(self):
"""
Lecture du fichier LISTE_BASSINS.DAT de GRP *Calage*
"""
self.clear()
header_len = CAL_CONFIG_HEADER.count('\n')
with open(self.filename, 'r', encoding='utf-8') as f:
for _ in range(header_len):
f.readline()
for line in f.readlines():
info = line.split('!')[1:-1]
info.remove('##')
if len(info) != len(CAL_CONFIG_NAMES):
_exception.Warning(
__name__,
"format de ligne incorrect")
continue
try:
run = {n: d(i.strip())
for n, d, i in zip(CAL_CONFIG_NAMES,
CAL_CONFIG_CONVERTERS,
info)}
run = GRP_Run(**run)
except SyntaxError:
_exception.Warning(
__name__,
"définition de ligne incorrecte")
else:
self.append(run)
[docs]
def write(self):
"""
Ecriture du fichier LISTE_BASSINS.DAT de GRP *Calage*
"""
with open(self.filename, 'w', encoding='utf-8', newline='\r\n') as f:
f.write(CAL_CONFIG_HEADER)
for run in self:
try:
self.check_run(run)
except ValueError as ve:
raise ValueError("Le run n'est pas un GRP_Run") from ve
run = run._asdict()
# Pour limiter le nom à 30 caractères
run['NOM'] = textwrap.shorten(run['NOM'], 30)
run['NOM'] = run['NOM'].replace('é', 'e')\
.replace('è', 'e')\
.replace('ê', 'e')\
.replace('ë', 'e')\
.replace('à', 'a')\
.replace('â', 'a')\
.replace('ô', 'o')\
.replace('ù', 'u')\
.replace('ç', 'c')
if run['NB'] >= 100:
_exception.Warning(
__name__,
f"NB du run trop grand : {run['NB']} >= 100")
try:
f.write(CAL_CONFIG_FORMAT.format(**run))
except ValueError:
_exception.Warning(
__name__,
"Incohérence de format lors de l'écriture "
f"du run\n{run}")
[docs]
def product(self, output_filename=None, filenames=None, only_2digits=True):
"""
Créer une nouvelle instance à partir de l'instance courante
et des listes de paramètres fournies dans les fichiers
Parameters
----------
output_filename : str
Fichier de la nouvelle instance
filenames : dict
Dictionnaire des fichiers donnant les autres valeurs de paramètres
- clé: Nom du paramètre
- valeur: Chemin du fichier de type csv
Les fichiers sont au format csv
code1;valeur1,1;valeur1,2;...;valeur1,N
code2;valeur2,1;valeur2,2;...;valeur2,N
Other Parameters
----------------
only_2digits : bool
limiter le paramètre 'NB' d'un run à 2 chiffres
Returns
-------
other_instance : GRP_Cfg
Nouvelle instance
See Also
--------
pyspc.convention.grp16.CAL_CONFIG_NAMES
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
# if filenames is None:
# filenames = {}
filenames = _exception.set_default(filenames, default={})
_exception.raise_valueerror(
not isinstance(filenames, dict),
"Les noms de fichiers source ne sont pas définis "
"à l'aide d'un dictionnaire"
)
_exception.raise_valueerror(
len(self) == 0,
"L'instance ne contient aucun run"
)
# if output_filename is None:
# output_filename = os.path.splitext(
# self.filename)[0] + '_product.DAT'
# _exception.Warning(
# None,
# "Le chemin associé à la nouvelle instance n'est pas défini "
# "par l'utilisateur. Il est défini ainsi: {}"
# "".format(output_filename))
output_filename = _exception.set_default(
output_filename,
default=os.path.splitext(self.filename)[0] + '_product.DAT',
text="Le chemin associé à la nouvelle instance n'est pas défini "
f"par l'utilisateur. Il est défini ainsi: {output_filename}",
warning=True)
# ---------------------------------------------------------------------
# 1- Lecture des paramètres complémentaires
# ---------------------------------------------------------------------
other_config = collections.OrderedDict()
other_keys = [o for o in CAL_CONFIG_NAMES if o in filenames]
for o in other_keys:
f = filenames[o]
dtype = CAL_CONFIG_CONVERTERS[CAL_CONFIG_NAMES.index(o)]
other_config.setdefault(o, read_multicfg(f, dtype=dtype))
# ---------------------------------------------------------------------
# 2- Définition de la liste des runs :
# les originaux + les complémentaires
# ---------------------------------------------------------------------
other_runs = []
other_counter = 0
for run in self:
run_dict = run._asdict()
code = run.CODE # run['CODE']
products = []
keys = []
for o in other_config.keys():
if code in other_config[o]:
product = other_config[o][code]
if run_dict[o] not in product:
product.insert(0, run_dict[o])
products.append(product)
keys.append(o)
if products:
products = list(itertools.product(*products))
for product in products:
fields = dict(zip(keys, product))
other_counter += 1
fields.update({'NB': other_counter})
current_run = run._replace(**fields)
if only_2digits:
if current_run.NB < 100:
other_runs.append(current_run)
else:
_exception.Warning(
__name__,
f"NB du run trop grand : {run.NB} >= 100")
else:
other_runs.append(current_run)
else:
other_counter += 1
current_run = run._replace(NB=other_counter)
if only_2digits:
if current_run.NB < 100:
other_runs.append(current_run)
else:
_exception.Warning(
__name__,
f"NB du run trop grand : {run.NB} >= 100")
else:
other_runs.append(current_run)
# ---------------------------------------------------------------------
# 3- Définition de la configuration
# ---------------------------------------------------------------------
other_instance = GRP_Cfg(filename=output_filename)
other_instance.extend(other_runs)
return other_instance
def read_multicfg(filename, dtype=None):
"""
Lire les fichiers contenant les valeurs supplémentaires (SC, HC, SV...)
Parameters
----------
filename : str
Chemin du fichier
dtype : function
Fonction de conversion. Exemples: 'int', 'float', 'bool'
Returns
-------
cfg : dict
Dictionnaire des valeurs supplémentaires, dont la clé est définie
par l'identifiant (1e colonne)
"""
cfg = {}
with open(filename, 'r', encoding='utf-8', newline='\n') as f:
for line in f.readlines():
info = line.replace('\n', '').split(';')
code = info.pop(0)
if dtype is not None:
info = [dtype(i) for i in info if i != '']
cfg.setdefault(code, info)
if cfg:
return cfg
return None