#!/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 - Bassin
"""
import collections
import os.path
from pyspc.convention.grp20 import CAL_BASIN_LINESEP, CAL_BASIN_HEADERS
[docs]
class GRP_Basin(collections.OrderedDict):
"""
Structure du fichier Bassin de GRP *Calage*
Attributes
----------
filename : str
Nom du fichier Bassin de GRP *Calage*
"""
[docs]
def __init__(self, filename=None):
"""
Initialisation de l'instance de la classe GRP_Data
Parameters
----------
filename : str
Nom du fichier Bassin de GRP *Calage*
"""
super().__init__()
self.filename = filename
if self.filename is not None:
info = self.split_basename(self.filename)
self.location = info[0]
self.timestep = info[1]
else:
self.location = None
self.timestep = None
def __str__(self):
"""
Afficher les méta-données de l'instance GRP_Basin
"""
text = """
*************************************
*********** GRP 2020 - Basin ********
*************************************
* NOM FICHIER = {filename}
* LIEU MODELE = {location}
* PAS DE TEMPS = {timestep}
* INFORMATIONS = {content}
*************************************
"""
content = ""
for k, v in self.items():
if isinstance(v, str):
content += f'\n * + {k} = {v}'
elif isinstance(v, (dict, collections.OrderedDict)):
content += f'\n * + {k}'.format(k)
for k2, v2 in v.items():
content += f'\n * + {k2} = {v2}'
return text.format(filename=self.filename, location=self.location,
timestep=self.timestep, content=content)
[docs]
def read(self):
"""
Lecture du fichier Bassin de GRP *Calage*
Examples
--------
>>> from pyspc.model.grp20.cal_basin import GRP_Basin
>>> f = 'data/model/grp20/cal/RH10585x_00J01H00M.DAT'
>>> basin = GRP_Basin(filename=f)
>>> basin.read()
>>> basin
*************************************
*********** GRP 2020 - Basin ********
*************************************
* NOM FICHIER = data/model/grp20/cal/RH10585x_00J01H00M.DAT
* LIEU MODELE = RH10585x
* PAS DE TEMPS = 00J01H00M
* INFORMATIONS =
* + E
* + RH10585x = {'n': 'Code et ponderation poste ETP 1', 'w': 1.0}
* + P
* + 90052002 = {'n': 'Code, ponderation et pas de temps poste pluvio 1', 'w': 0.8, 't': '00J01H00M'}
* + 90065003 = {'n': 'Code, ponderation et pas de temps poste pluvio 2', 'w': 0.2, 't': '00J01H00M'}
* + T
* + 90035001 = {'n': 'Code, ponderation et altitude du poste temperature 1', 'w': 1.0, 'z': 401.0}
* + 90052002 = {'n': 'Code, ponderation et altitude du poste temperature 2', 'w': 1.0, 'z': 473.0}
* + 90065003 = {'n': 'Code, ponderation et altitude du poste temperature 3', 'w': 1.0, 'z': 1153.0}
* + L
* + t = 01J00H00M
* + n = Duree seuil pour les lacunes (format nnJnnHnnM)
*************************************
"""
content = {}
with open(self.filename, 'r', encoding='iso-8859-15') as f:
for line in f:
# Une ligne de commentaire commence par "#" ou "!"
if line.startswith("!") or line.startswith("#"):
continue
# Balise
tag = line[0]
# Ajout de la balise
content.setdefault(tag, collections.OrderedDict())
# Découpage de la ligne
x = line.split("!", 1)[0][2:].split(" ")
x = list(filter(None, x))
code = x.pop(0) # Id Station
# Nom de l'entité
try:
name = line.split("!")[1].strip()
except IndexError:
name = ''
# Spécificités N/L
if tag in ['N']:
content[tag] = float(code)
continue
if tag in ['L']:
content[tag] = {'t': code, 'n': name}
continue
# Ajout de l'entité
content[tag].setdefault(code, {})
content[tag][code]['n'] = name
# Pondération de l'entité
try:
w = float(x.pop(0)) # Poids Station
except IndexError:
pass
else:
content[tag][code]['w'] = w
# 3e élément : z (si T), pdt (si P)
try:
t = x.pop(0)
except IndexError:
pass
else:
if tag == 'P':
content[tag][code]['t'] = t
elif tag == 'T':
content[tag][code]['z'] = float(t) # Altitude
self.update(content)
[docs]
def write(self):
"""
Ecriture du fichier Bassin de GRP *Calage*
"""
with open(self.filename, 'w',
encoding='iso-8859-15', newline="\r\n") as f:
# Première ligne
f.write(CAL_BASIN_LINESEP)
# Boucle sur les étiquettes
for tag in self.keys():
# Entête de l'étiquette
f.write(CAL_BASIN_HEADERS[tag])
if tag in ['N']:
f.write(f'{tag:1s}')
f.write(f' {self[tag]:9.2f}')
f.write(f' {"":<98s} !')
f.write('\n')
f.write(CAL_BASIN_LINESEP)
continue
if tag in ['L']:
f.write(f'{tag:1s}')
f.write(f" {self[tag]['t']:9s}")
f.write(f' {"":<17s}')
f.write(f"! {self[tag]['n']:<79s} !")
f.write('\n')
f.write(CAL_BASIN_LINESEP)
continue
# Boucle sur les entités de l'étiquette
for code in self[tag].keys():
# Etiquette
f.write(f'{tag:1s}')
# Station
f.write(f' {code:8s}')
# Pondération
f.write(f" {self[tag][code]['w']:9.2f}")
# Altitude - Nom
if tag == 'T':
f.write(f" {self[tag][code]['z']:7.2f}")
f.write(f" ! {self[tag][code]['n']:<79s} !")
elif tag == 'P':
f.write(f" {self[tag][code]['t']:9s}")
f.write(f" ! {self[tag][code]['n']:<77s} !")
else:
f.write(f" ! {self[tag][code]['n']:<87s} !")
f.write('\n')
f.write(CAL_BASIN_LINESEP)
[docs]
@staticmethod
def split_basename(filename=None):
"""
Extraire les informations depuis le nom du fichier
de données GRP Basin (GRP *Calage*)
Parameters
----------
filename : str
Fichier de données GRP Basin (GRP *Calage*)
Returns
-------
location : str
Identifiant de la station
timestep : str
Pas de temps du modèle
Examples
--------
>>> from pyspc.model.grp20 import GRP_Basin
>>> f = 'data/model/grp20/cal/RH10585x_00J01H00M.DAT'
>>> [location, timestep] = GRP_Basin.split_basename(filename=f)
>>> location
RH10585x
>>> varname
00J01H00M
"""
if filename is None:
return None, None
basename = os.path.splitext(os.path.basename(filename))[0]
try:
[location, timestep] = basename.split('_')
except ValueError as ve:
raise ValueError("Le nom de fichier ne respecte pas le "
"nommage de GRP") from ve
return (location, timestep)
[docs]
@staticmethod
def join_basename(location=None, timestep=None):
"""
Définir le nom du fichier de données GRP Basin (GRP *Calage*)
à partir des informations
Parameters
----------
location : str
Identifiant de la station
timestep : str
Pas de temps du modèle
Returns
-------
filename : str
Fichier de données GRP Basin (GRP *Calage*)
"""
if location is None or timestep is None:
raise ValueError('Définition incorrecte des arguments')
return f'{location}_{timestep}.DAT'