#!/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 - GRP version 2016 - write
"""
from datetime import datetime as dt
import numpy as np
import os.path
from pyspc.convention.grp16 import (
CAL_DATA_HEADERS,
RT_ARCHIVE_VARNAMES, RT_ARCHIVE_BASENAMES,
RT_DATA_OBSFILEPREFIX, RT_DATA_LINEPREFIX, RT_DATA_SCENFORMAT)
from pyspc.model.grp16 import GRP_Data, GRPRT_Archive, GRPRT_Data
from pyspc.core.convention import EXTERNAL_VARNAMES
from pyspc.core.series import Series
import pyspc.core.exception as _exception
[docs]
def write_GRP16(series=None, dirname='.', datatype=None, how='replace'):
"""
Créer des fichiers de données GRP16 à partir d'une instance Series
Parameters
----------
series : pyspc.core.series.Series
Collection de séries de données
datatype : str
Type de données
Voir pyspc.convention.grp16.DATATYPES
dirname : str
Répertoire de destination
Other Parameters
----------------
how : str
Option d'écriture, si datatype = 'grp16_cal_data'
- replace : écraser si un fichier existe déjà (défaut)
- fillna : mettre à jour le fichier existant
(seulement les NaN et les valeurs aux instants non existants)
- overwrite : mettre à jour le fichier existant
(toutes les valeurs, y compris les non-NaN)
Returns
-------
filenames : list
Noms des fichiers écrits
"""
# -------------------------------------------------------------------------
# 0- Contrôles
# -------------------------------------------------------------------------
_exception.check_str(dirname)
_exception.raise_valueerror(
not isinstance(series, Series),
"'series' doit être une collection 'Series'"
)
_exception.raise_valueerror(
how not in ['replace', 'fillna', 'overwrite'],
"Option d'écriture incorrecte"
)
assoc = {v: k[1] for k, v in EXTERNAL_VARNAMES.items() if k[0] == 'GRP16'}
# -------------------------------------------------------------------------
# 1.1- CALAGE - Cas de données d'observation
# -------------------------------------------------------------------------
if datatype == 'grp16_cal_data':
return _grp16_cal_data(series=series, dirname=dirname,
datatype=datatype, how=how, assoc=assoc)
# -------------------------------------------------------------------------
# 2.1- TEMPS-REEL - Cas de données d'observation
# TEMPS-REEL - Cas de scénarios météorologiques
# -------------------------------------------------------------------------
if datatype == 'grp16_rt_data':
return _grp16_rt_data(series=series, dirname=dirname, assoc=assoc)
# -------------------------------------------------------------------------
# 2.2- TEMPS-REEL - Cas des archives
# -------------------------------------------------------------------------
if datatype == 'grp16_rt_archive':
return _grp16_rt_archive(series=series, dirname=dirname, assoc=assoc)
# -------------------------------------------------------------------------
# 3- Cas inconnu
# -------------------------------------------------------------------------
raise NotImplementedError
def _grp16_cal_data(series=None, dirname=None, datatype=None,
how=None, assoc=None):
"""
CALAGE - Cas de données d'observation
"""
from pyspc.io.grp16.reader import read_GRP16
filenames = []
for key, serie in series.items():
# Ignorer les simulations et prévisions
if key[2] is not None:
continue
# Définir station, variable et nom du fichier
station = key[0]
try:
varname = assoc[key[1]]
except KeyError:
continue
filename = os.path.join(
dirname,
GRP_Data.join_basename(station=station, varname=varname))
# Charger les données existantes si la série les complète
if os.path.exists(filename) and how in ['fillna', 'overwrite']:
try:
other = read_GRP16(
datatype=datatype, filename=filename)[key]
if how == 'fillna':
other.update(serie, overwrite=False)
else:
other.update(serie, overwrite=True)
except (KeyError, ValueError):
pass
else:
serie = other
# Construire le dataframe
df = serie.data_frame
df.index.name = CAL_DATA_HEADERS['index']
df.columns = [CAL_DATA_HEADERS[varname]]
writer = GRP_Data(filename=filename)
writer.write(data=df)
filenames.append(filename)
return filenames
def _grp16_rt_archive(series=None, dirname=None, assoc=None):
"""
TEMPS-REEL - Cas des archives
"""
filenames = []
for serie in series.values():
# -----------------------------------------------------------------
# 1.1 - Fusion des séries + reformattage du DataFrame
# -----------------------------------------------------------------
df = serie.data_frame
loc = serie.code
df.columns = [loc]
df.columns.name = 'Code'
df.index.name = 'Date (TU)'
varname = f'{assoc[serie.spc_varname]}V'
if varname not in RT_ARCHIVE_VARNAMES:
continue
basename = RT_ARCHIVE_BASENAMES[varname]
# -----------------------------------------------------------------
# 1.2 - Export
# -----------------------------------------------------------------
filename = os.path.join(dirname, loc)
if not os.path.exists(filename):
os.makedirs(filename)
filename = os.path.join(filename, basename)
writer = GRPRT_Archive(filename=filename)
writer.write(df)
filenames.append(filename)
return filenames
def _grp16_rt_data(series=None, dirname=None, assoc=None):
"""
TEMPS-REEL - Cas de données d'observation
TEMPS-REEL - Cas de scénarios météorologiques
"""
filenames = []
# -------------------------------------------------------------------------
# 1- TEMPS-REEL - Cas de données d'observation
# -------------------------------------------------------------------------
if series.datatype == 'obs':
cases = _grp16_rt_data_obskeys(series=series, assoc=assoc)
for varname, keys in cases.items():
# -----------------------------------------------------------------
# 1.1 - Fusion des séries + reformattage du DataFrame
# -----------------------------------------------------------------
df = series.concat(keys)
df = _grp16_rt_data_format_df(df=df, varname=varname)
# -----------------------------------------------------------------
# 1.2 - Export
# -----------------------------------------------------------------
filename = os.path.join(
dirname,
f'{RT_DATA_OBSFILEPREFIX[varname]}_{series.name}.txt')
writer = GRPRT_Data(filename=filename)
writer.write(df)
filenames.append(filename)
return filenames
# -------------------------------------------------------------------------
# 2- TEMPS-REEL - Cas de scénarios météorologiques
# -------------------------------------------------------------------------
if series.datatype == 'fcst':
# -----------------------------------------------------------------
# 2.1 - Sélection des scenarios et grandeurs météorologiques
# -----------------------------------------------------------------
cases = _grp16_rt_data_fcstkeys(series=series, assoc=assoc)
for case in cases:
sv = case[0]
scen = case[1]
v = assoc[sv]
# -----------------------------------------------------------------
# 2.2 - Fusion des séries + reformattage du DataFrame
# -----------------------------------------------------------------
keys = cases[case]
df = series.concat(keys)
df = _grp16_rt_data_format_df(df=df, varname=v)
# -----------------------------------------------------------------
# 2.3 - Export
# -----------------------------------------------------------------
filename = os.path.join(
dirname, RT_DATA_SCENFORMAT[v].format(
f"{scen:0>3s}"[-3:], series.name))
writer = GRPRT_Data(filename=filename)
writer.write(df)
filenames.append(filename)
return filenames
# -------------------------------------------------------------------------
# 3- Cas inconnu
# -------------------------------------------------------------------------
raise NotImplementedError
def _grp16_rt_data_obskeys(series=None, assoc=None):
"""Clés par cas d'export - Observation hydrométéo"""
cases = {assoc[sv]: [key
for key in series if key[1] == sv and key[2] is None]
for sv in set(series.varnames) if sv in assoc}
return cases
def _grp16_rt_data_fcstkeys(series=None, assoc=None):
"""Clés par cas d'export - Prévision météo"""
scens = sorted(list({s[2] for s in series.meta}))
varnames = sorted(list({v
for v in series.varnames
if v in assoc
if assoc[v] in RT_DATA_SCENFORMAT}))
cases = {(v, s): [key for key in series
if key[1] == v and key[2][2] == s]
for v in varnames for s in scens}
return cases
def _grp16_rt_data_format_df(df=None, varname=None):
"""
Formater le DataFrame
"""
prefix = RT_DATA_LINEPREFIX[varname]
df.columns = [c[0] for c in df.columns]
# Reformater l'index
df = df.stack(0)
df.index.rename('DATE', level=0, inplace=True)
df.index.rename('CODE', level=1, inplace=True)
# Pour trier par POSTE croissante puis par DATE croissante
df.sort_index(level=['CODE', 'DATE'], inplace=True)
# Effacer l'index : ('DATE', 'POSTE') -> (0, 1, ... N)
df = df.reset_index(level=['DATE', 'CODE'])
# Changer le nom de la colonne des valeurs
df.columns = ['DATE', 'CODE', 'VALUE']
# Formatter la date en 2 colonnes (Jour, Heure)
df['DATE'] = df['DATE'].map(
lambda x: dt.strftime(x, '%Y%m%d %H:%M'))
new = df['DATE'].str.split(" ", expand=True)
df['DATE'] = new[0]
df['HOUR'] = new[1]
# Ré-arrangement des colonnes dans le bon ordre
# + ajout de la colonne superflue
df = df.reindex(
columns=['PREFIX', 'CODE', 'DATE', 'HOUR', 'VALUE', None])
df = df.fillna(value={'PREFIX': prefix, None: np.nan})
return df