#!/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 - Sandre - write
"""
from datetime import datetime as dt
import pandas as pnd
import re
from pyspc.convention.sandre import ASSOC, RATIO_UNITS, TRENDS
from pyspc.data.sandre import Sandre
from pyspc.core.parameter import Parameter
from pyspc.core.keyseries import str2tuple, tuple2str
from pyspc.core.series import Series
import pyspc.core.exception as _exception
[docs]
def write_Sandre(series=None, datatype=None, filename=None,
runtime=None, sender=None, user=None, target=None):
"""
Ecrire un fichier xml Sandre à partir d'une collections de séries
Parameters
----------
series : pyspc.core.series.Series
Collection de séries de données
dataype : str
Type de données
Voir pyspc.data.sandre.Sandre.get_types
filename : str
Nom du fichier xml à écrire
runtime : datetime
Date de production de la donnée/prévision.
Si non défini, la date est dt.utcnow()
sender : str
Identifiant de l'emetteur
user : str
Identifiant du contact
target : str
Identifiant du destinataire
Returns
-------
filename : str
Nom du fichier xml écrit
"""
# -------------------------------------------------------------------------
# 0- Contrôles
# -------------------------------------------------------------------------
_exception.check_str(filename)
_exception.raise_valueerror(
not isinstance(series, Series),
"'series' doit être une collection 'Series'"
)
writer = Sandre(filename=filename)
Sandre.check_dtype(datatype)
assoc = {v: k for k, v in ASSOC.items()}
trends = {v: k for k, v in TRENDS.items()}
# -------------------------------------------------------------------------
# 1- Eléments communs
# -------------------------------------------------------------------------
# Date de production
if runtime is None:
runtime = dt.utcnow()
# Pas de temps de cumul
p = Parameter(series.varnames[0]).timestep
if p is None:
p = 0
# Concaténer les séries de données
df = series.concat()
# Grandeur
# df.columns = [(c[0], assoc[c[1][0]], c[2]) for c in df.columns]
df.columns = set_df_columns(df.columns, assoc)
# -------------------------------------------------------------------------
# 2- Cas des données d'observation
# -------------------------------------------------------------------------
if datatype.startswith('data_obs'):
# Inclure le nom de la simulation dans l'identifiant du lieu
df.columns = pnd.MultiIndex.from_tuples(
[str2tuple(tuple2str(c), forceobs=True) for c in df.columns],
names=['Location', 'Varname', 'Meta'])
# Retirer le 3e niveau dans le MultiIndex
df = df.droplevel('Meta', axis=1)
# Appliquer le changement d'unité
for c in df.columns:
df[c] = df[c] / RATIO_UNITS[c[1]]
return writer.write(
data=df, datatype=datatype, runtime=runtime, timedelta=p,
sender=sender, user=user, target=target)
# -------------------------------------------------------------------------
# 3- Cas des données de prévision
# -------------------------------------------------------------------------
if datatype.startswith('data_fcst'):
# Eclater le tuple des méta-données
df.columns = pnd.MultiIndex.from_tuples(
[tuple([c[0], c[1], *c[2][:-1], trends[c[2][-1]]])
for c in df.columns],
names=['Location', 'Varname',
'Runtime', 'Model', 'Scenario', 'Prob'])
# Appliquer le changement d'unité
for c in df.columns:
df[c] = df[c] / RATIO_UNITS[c[1]]
return writer.write(
data=df, datatype=datatype, runtime=runtime, timedelta=p,
sender=sender, user=user, target=target)
# -------------------------------------------------------------------------
# 4- Erreur
# -------------------------------------------------------------------------
raise NotImplementedError(f"Type d'export '{datatype}' non implémenté")
def set_df_columns(cols, assoc):
"""
Définir les nouvelles entêtes de colonnes
"""
new_cols = []
for c in cols:
v = assoc[c[1][0]]
s = c[0]
if v == 'H' and len(s) != 10:
s2 = f'{s:0<10s}'
_exception.Warning(
None,
f"L'identifiant {s} ne correspond pas à une station hydro. "
f"Il est forcé à 10 caractères: {s2}")
s = s2
elif v == 'RR' and not s.isdigit():
x = re.sub(r"\D", "", s) # pylint: disable=unused-variable
s2 = f'{x:0<8s}'
_exception.Warning(
None,
f"L'identifiant {s} ne correspond pas à un site météo. "
f"Il est forcé à {s2}")
s = s2
new_cols.append((s, v, c[2]))
return new_cols