#!/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/>.
#
########################################################################
"""
Bibliothèque pyspc du projet pyspc - IO - PLATHYNES - write
"""
import os.path
from pyspc.convention.plathynes import DATA_EXT, RATIOS
import pyspc.core.exception as _exception
from pyspc.core.parameter import Parameter
from pyspc.core.series import Series
from pyspc.model.plathynes import Data
[docs]
def write_Plathynes(series=None, dirname=None, suffix=None, event=None,
injections=None):
"""
Ecrire des fichiers d'observation (.mgr, .mqo, .mqi) pour PLATHYNES
Parameters
----------
dirname : str
Répertoire de destination
suffix : dict
Association entre les stations hydrométriques (grandeur: Q)
et les suffixes des noms de fichier
Other Parameters
----------------
event : str
Identifiant de l'événement.
Si non défini, reprend le nom de la collection
injections : dict
Indiquer si une série (clé=code de la série)
est une injection pour PLATHYNES (valeur: True/False)
Returns
-------
filenames : list
Fichiers créés
"""
# -------------------------------------------------------------------------
# 0- Contrôles
# -------------------------------------------------------------------------
_exception.raise_valueerror(
not isinstance(series, Series),
"'series' doit être une collection 'Series'"
)
_exception.check_str(dirname)
_exception.check_str(event)
if suffix is None:
suffix = {}
_exception.check_dict(suffix)
if injections is None:
injections = {}
_exception.check_dict(injections)
# -------------------------------------------------------------------------
# 1- Initialisation
# -------------------------------------------------------------------------
keys = {v: [k for k in series.keys() if k[1] == v]
for v in series.varnames}
filenames = []
data_ext = {v: k for k, v in DATA_EXT.items()}
# -------------------------------------------------------------------------
# 2- Séries par grandeur
# -------------------------------------------------------------------------
for varname in keys:
subkeys = keys[varname]
# ---------------------------------------------------------------------
# 2.1- pluviométrie
# ---------------------------------------------------------------------
if varname.startswith('P'):
basename = '_'.join([event, 'RRobs'])
filename = os.path.join(
dirname, f"{basename}{data_ext[('P', False)]}")
writer = Data(filename=filename)
meta = {
'Type de donnees': 'PLUVIO',
'Station': 'PROJECT_SET',
'Pas de temps': Parameter(varname=varname).timestep,
'Facteur multiplicatif': 1,
'locs': {series[k].code: {'x': series[k].location.x,
'y': series[k].location.y,
'code': series[k].location.code}
for k in subkeys}
}
df = series.concat(subkeys)
df.columns = [c[0] for c in df.columns]
df = df * 1 / RATIOS.get(writer.varname, 1) # mm ===> 1/10 mm
filenames.append(writer.write(data=df, meta=meta))
# ---------------------------------------------------------------------
# 2.2- débitmétrie
# ---------------------------------------------------------------------
if varname.startswith('Q'):
for key in subkeys:
serie = series[key]
code = serie.location.code
inj = injections.get(code, False)
ext = data_ext[('Q', inj)]
basename = '_'.join([event, suffix.get(code, code)])
filename = os.path.join(dirname, f'{basename}{ext}')
writer = Data(filename=filename)
meta = {'loc': code, 'loc0': code,
'x': serie.location.x, 'y': serie.location.y,
'obj': 'Qobs', 'value': '0.000'}
df = series[key].data_frame
df.columns = [serie.location.code]
filenames.append(writer.write(data=df, meta=meta))
return filenames