#!/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
"""
import collections
import itertools
import os.path
import textwrap
import pyspc.core.exception as _exception
from pyspc.model.grp18.cal_config import read_multicfg
from pyspc.convention.grp22 import (
CAL_CONFIG_BVKEYS, CAL_CONFIG_MODKEYS, CAL_CONFIG_CALKEYS,
CAL_CONFIG_CONVERTERS, CAL_CONFIG_FORMAT, CAL_CONFIG_HEADER,
CAL_CONFIG_NAMES, CAL_CONFIG_COMPNAMES) # CAL_CONFIG_COMPMISS
GRP_Run = collections.namedtuple('GRP_Run', CAL_CONFIG_NAMES)
GRP_Run.__doc__ = """Run de GRP 2022
Attributes
----------
CODE : str
Identifiant du bassin
PDT : str
Pas de temps, au format ..J..H..M
NOM : str
Libellé bassin
SURFACE : float
Superficie du bassin en km²
RT : str
Référentiel Temporel (TU, HL ou HH)
DEB : str
Date début
FIN : str
Date fin
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
HOR1 : str
Horizon de calage #1, au format ..J..H..M
HOR2 : str
Horizon de calage #2, au format ..J..H..M
SC1 : float
Seuil de calage #1 en m3/s
SC2 : float
Seuil de calage #2 en m3/s
SV1 : float
Seuil #1 utilise pour l'analyse des résultat, en m3/s
SV2 : float
Seuil #2 utilise pour l'analyse des résultat, en m3/s
SV3 : float
Seuil #3 utilise pour l'analyse des résultat, en m3/s
NJ : int
Nombre de jours pour le tracé des hydrogrammes prevus
HC : str
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
"""
[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 2022 - 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_basinvalues(self):
"""
Liste des valeurs correspondant aux paramètres de calage de GRP
"""
return [tuple([r._asdict()[k] for k in CAL_CONFIG_BVKEYS])
for r in self]
[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 get_modelvalues(self):
"""
Liste des valeurs correspondant aux paramètres de calage de GRP
"""
return [tuple([r._asdict()[k] for k in CAL_CONFIG_MODKEYS])
for r in self]
[docs]
def read(self, encoding='iso-8859-15'):
"""
Lecture du fichier LISTE_BASSINS.DAT de GRP *Calage*
Parameters
----------
encoding : str
Encodage du fichier, 'iso-8859-15' par défaut
"""
self.clear()
header_len = CAL_CONFIG_HEADER.count('\n')
with open(self.filename, 'r', encoding=encoding) as f:
for _ in range(header_len):
f.readline()
for line in f.readlines():
info = line.split('!')[1:-1]
info.remove('##')
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 (ValueError, SyntaxError):
_exception.Warning(
__name__,
"définition de ligne incorrecte")
else:
self.append(run)
[docs]
def write(self, encoding='iso-8859-15', newline='\n'):
"""
Ecriture du fichier LISTE_BASSINS.DAT de GRP *Calage*
Parameters
----------
encoding : str
Encodage du fichier, 'iso-8859-15' par défaut
newline : str
Charactère de nouvelle ligne, '\n' par défaut
"""
with open(self.filename, 'w', encoding=encoding, newline=newline) 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')
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):
"""
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
Returns
-------
other_instance : GRP_Cfg
Nouvelle instance
See Also
--------
pyspc.convention.grp16.CAL_CONFIG_NAMES
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
if filenames is None:
filenames = {}
_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 "
f"par l'utilisateur. Il est défini ainsi: {output_filename}")
# ---------------------------------------------------------------------
# 1-1 Lecture des paramètres complémentaires
# ---------------------------------------------------------------------
other_config = collections.OrderedDict()
other_config.update(
load_config(names=CAL_CONFIG_NAMES, filenames=filenames))
# ---------------------------------------------------------------------
# 1-2 Lecture des paramètres complémentaires : cas HORx, SCx, SVx
# ---------------------------------------------------------------------
other_config.update(
load_config(names=CAL_CONFIG_COMPNAMES, filenames=filenames))
# ---------------------------------------------------------------------
# 2- Définition de la liste des runs :
# les originaux + les complémentaires
# ---------------------------------------------------------------------
other_runs = []
for run in self:
products, keys = set_keys_and_products(other_config, run)
if products:
products = list(itertools.product(*products))
for product in products:
fields = dict(zip(keys, product))
current_run = run._replace(**fields)
other_runs.append(current_run)
else:
other_runs.append(run)
# ---------------------------------------------------------------------
# 3- Définition de la configuration
# ---------------------------------------------------------------------
other_instance = GRP_Cfg(filename=output_filename)
other_instance.extend(other_runs) # other_instance is a list !
return other_instance
def load_config(names=None, filenames=None):
"""
Charger les listes des nouvelles valeurs par paramètre
Parameters
----------
names : list
Paramètres autorisés
filenames : dict
Dictionnaire des fichiers donnant les autres valeurs de paramètres
- clé: Nom du paramètre (parmi {})
- 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
"""
cfg = collections.OrderedDict()
if names is None or filenames is None:
return cfg
other_keys = [o for o in names if o in filenames]
for o in other_keys:
f = filenames[o]
try:
dtype = CAL_CONFIG_CONVERTERS[CAL_CONFIG_NAMES.index(o)]
except Exception:
dtype = CAL_CONFIG_CONVERTERS[CAL_CONFIG_NAMES.index(o + '1')]
cfg.setdefault(o, read_multicfg(f, dtype=dtype))
return cfg
def set_keys_and_products(other_config, run):
"""
Définir la liste des valeurs et la liste des paramètres
"""
run_dict = run._asdict()
code = run.CODE # run_dict['CODE']
ts = run.PDT # run_dict['PDT']
products = []
keys = []
for o in other_config.keys():
if (code, ts) not in other_config[o]:
continue
if o in CAL_CONFIG_COMPNAMES:
c = set_comp_values(other_config, o, code, ts, run_dict)
for x in c:
products.append(c[x])
keys.append(x)
else:
product = other_config[o][(code, ts)]
if run_dict[o] not in product:
product.insert(0, run_dict[o])
products.append(product)
keys.append(o)
return products, keys
def set_comp_values(other_config, o, code, ts, run_dict):
"""
Répartir les valeurs dans le cas des paramètres HORx, SCx, SVx
"""
cfg = {}
names = [x for x in CAL_CONFIG_NAMES if x.startswith(o)]
# m = CAL_CONFIG_COMPMISS[o]
product = other_config[o][(code, ts)]
for n in names[::-1]:
if run_dict[n] not in product:
product.insert(0, run_dict[n])
product = sorted(product)
for x, y in zip(itertools.cycle(names), product):
cfg.setdefault(x, [])
cfg[x].append(y)
return cfg