#!/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 - prv (Otamin, Scores) - write
"""
import pandas as pnd
import pyspc.core.exception as _exception
from pyspc.convention.prv import DATATYPES
from pyspc.core.convention import EXTERNAL_VARNAMES
from pyspc.core.series import Series
from pyspc.model.otamin16 import RT_Data as PRV_otamin16
from pyspc.model.otamin16.rt_data import date_formatter
from pyspc.model.otamin18 import RT_Data as PRV_otamin18
from pyspc.verification.scores import Data as PRV_scores
[docs]
def write_prv(series=None, datatype=None, filename=None):
"""
Ecrire un fichier prv (Otamin, Scores)
Parameters
----------
series : pyspc.core.series.Series
Collection de séries de données
datatype : str
Type du fichier de données
filename : str
Nom du fichier prv à écrire
Returns
-------
filename : str
Nom du fichier prv écrit
"""
# -------------------------------------------------------------------------
# 0- Contrôles
# -------------------------------------------------------------------------
_exception.check_str(filename)
_exception.raise_valueerror(
datatype not in DATATYPES,
f"Type de données prv '{datatype}' incorrect")
_exception.raise_valueerror(
not isinstance(series, Series),
"'series' doit être une collection 'Series'"
)
# -------------------------------------------------------------------------
# 1- Ecrivain
# -------------------------------------------------------------------------
writer = None
subtype = datatype.split('_')[-1]
if datatype.startswith('otamin16'):
writer = PRV_otamin16(filename=filename, datatype=subtype)
elif datatype.startswith('otamin18'):
writer = PRV_otamin18(filename=filename, datatype=subtype)
elif datatype.startswith('scores'):
writer = PRV_scores(filename=filename, datatype=subtype)
# -------------------------------------------------------------------------
# 2- Conversion - CAS OBSERVATION / SIMULATION
# -------------------------------------------------------------------------
if subtype in ['obs', 'sim']:
return _prv_obssim(series=series, subtype=subtype, writer=writer)
# -------------------------------------------------------------------------
# 3- Conversion - CAS PREVISION / TENDANCE
# -------------------------------------------------------------------------
if subtype in ['fcst', 'trend']:
return _prv_fcsttrend(series=series, datatype=datatype,
subtype=subtype, writer=writer)
# -------------------------------------------------------------------------
# 4- Erreur
# -------------------------------------------------------------------------
raise NotImplementedError(f"Type d'export '{datatype}' non implémenté")
def _prv_obssim(series=None, subtype=None, writer=None):
"""
CAS OBSERVATION / SIMULATION
"""
if subtype == 'obs':
keys = [k for k in series.keys() if k[2] is None]
nok = [k for k in series.keys() if k not in keys]
if len(nok) > 0:
_exception.Warning(
None,
"Les séries suivantes ne sont pas des observations. Elles "
f"ne sont donc pas exportées.\n{nok}")
else:
keys = [k for k in series.keys() if isinstance(k[2], str)]
nok = [k for k in series.keys() if k not in keys]
if len(nok) > 0:
_exception.Warning(
None,
"Les séries suivantes ne sont pas des simulations. Elles "
f"ne sont donc pas exportées.\n{nok}")
df = series.concat(keys=keys)
cols = pnd.MultiIndex.from_tuples(
[(c[0], _prv_varname(c[1]), 'obs' if c[2] is None else c[2])
for c in df.columns],
names=('Stations', 'Grandeurs', 'IdSeries')
)
df.columns = cols
df.index.name = None
writer.write(data=df)
return writer.filename
def _prv_fcsttrend(series=None, datatype=None, subtype=None, writer=None):
"""
CAS PREVISION / TENDANCE
"""
if datatype.startswith('scores'):
subtype = 'trend'
df = series.concat()
cols = pnd.MultiIndex.from_tuples(
[(c[0],
_prv_varname(c[1]),
date_formatter(c[2][0]),
c[2][1],
c[2][2],
_prv_prob(c[2][3], subtype),
_prv_idseries(c[2][1], c[2][2], c[2][3], subtype))
for c in df.columns],
names=('Stations', 'Grandeurs', '# DtDerObs', '# Modeles',
'# Scenarios', '# Probas', 'IdSeries')
)
if subtype != 'trend':
cols = cols.droplevel(level='# Probas')
df.columns = cols
df.index.name = None
writer.write(data=df)
return writer.filename
def _prv_idseries(model, scen, prob, subtype):
"""
Définir l'IdSeries
"""
txt = "{model}_{scen}"
if subtype == 'trend':
txt += "_{prob}"
prob = _prv_prob(prob, subtype)
return txt.format(model=model, scen=scen, prob=prob)
def _prv_prob(prob, subtype):
"""
Définir Prob
"""
if subtype == 'trend' and prob is None:
return '-1'
return prob
def _prv_varname(varname):
"""
Définir la grandeur dans la convention prv
"""
for k, v in EXTERNAL_VARNAMES.items():
if k[0] != 'prv':
continue
if varname.startswith(v[0]):
return k[1][0]
# return k[1]
raise ValueError("Grandeur '{varname}' incompatible avec le prv")