#!/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/>.
#
########################################################################
"""
Objets natifs et convention de pyspc - Format CSV multi-colonne
"""
import os.path
import pandas as pnd
from pyspc.core.keyseries import str2tuple, tuple2str
import pyspc.core.exception as _exception
from pyspc.core.convention import SPC_VARNAMES
from pyspc.core.parameter import Parameter
from pyspc.core.timeutil import dtfmt, dtheader
[docs]
class PyspcFile():
"""
Structure de données des csv multi-colonnes
Attributes
----------
filename : str
Nom du fichier de données
station : str
Nom de la station
varname : str
Nom de la variable
param : Parameter
Grandeur
tdelta : timedelta, None
Pas de temps de la variable
simulation : str, None
Nom de la simulation
runtime : str, None
Instant de production de la prévision
model : str, None
Modèle de prévision
scen : str, None
Scénario de prévision
uncert : str, None
Incertitude de prévision
"""
[docs]
def __init__(self, filename=None):
"""
Initialisation de l'instance de la classe PyspcFile
Parameters
----------
filename : str
Nom du fichier de données
"""
self.filename = filename
info = list(self.split_basename(filename=self.filename))
self.station = info.pop(0)
self.varname = info.pop(0)
self.simulation = info.pop(0)
self.runtime = info.pop(0)
self.model = info.pop(0)
self.scen = info.pop(0)
self.uncert = info.pop(0)
if self.varname in self.get_varnames():
self.param = Parameter(varname=self.varname)
self.tdelta = self.param.timestep
else:
self.param = None
self.tdelta = None
def __str__(self):
"""
Afficher les méta-données de l'instance GRP_Data
"""
text = """
*************************************
*********** CORE CSV FILE ***********
*************************************
* NOM FICHIER = {filename}
* LIEU = {station}
* GRANDEUR = {varname}
* PAS DE TEMPS = {tdelta}
* SIMULATION = {simulation}
* INSTANT PRODUCTION = {runtime}
* MODELE = {model}
* SCENARIO = {scen}
* INCERTITUDE = {uncert}
*************************************
"""
return text.format(**vars(self))
def _check_dataframe(self, df=None, strict=False):
"""
Contrôler le tableau de données est correctement formaté
Parameters
----------
df : pandas.DataFrame
Tableau des données
strict : bool
Lecture avec vérification de la variable (défaut: False)
Raises
------
ValueError
Si la variable n'est pas reconnue par pyspc
"""
_exception.raise_valueerror(
df.index.name != dtheader(self.tdelta),
f"Entête de colonne mal-formatée : {df.index.name} "
f"!= {dtheader(self.tdelta)}",
strict
)
def _check_varname(self, varname=None):
"""
Contrôler la variable
Parameters
----------
varname : str
Nom de la variable.
Raises
------
ValueError
Si la variable n'est pas reconnue par pyspc
"""
if varname not in self.get_varnames():
raise ValueError("Variable mal renseignée")
[docs]
def read(self, strict=False):
"""
Lecture du fichier csv multi-colonnes
Parameters
----------
strict : bool
Lecture avec vérification de la variable (défaut: False)
Returns
-------
pandas.DataFrame
Tableau des données
Examples
--------
>>> from pyspc.core.csvfile import PyspcFile
Cas d'une grandeur en minutes
>>> f = 'data/core/csv/K1341810_QI.txt'
>>> reader = PyspcFile(filename=f)
>>> df = reader.read()
>>> df
K1341810_QI
AAAAMMJJHHMM
2020-03-05 12:00:00 87.4
2020-03-05 13:00:00 90.6
2020-03-05 16:00:00 107.0
2020-03-05 17:35:00 117.0
2020-03-05 18:20:00 121.0
2020-03-05 18:50:00 124.0
2020-03-05 21:55:00 139.0
2020-03-05 23:25:00 146.0
2020-03-06 00:10:00 151.0
2020-03-06 00:22:00 151.0
2020-03-06 01:45:00 160.0
2020-03-06 03:20:00 170.0
2020-03-06 10:30:00 205.0
2020-03-06 13:00:00 213.0
2020-03-06 14:05:00 217.0
2020-03-06 15:20:00 219.0
2020-03-06 16:00:00 220.0
2020-03-06 16:20:00 219.0
2020-03-06 16:50:00 220.0
2020-03-06 17:45:00 220.0
2020-03-06 18:25:00 221.0
2020-03-06 19:10:00 221.0
2020-03-06 22:00:00 219.0
2020-03-06 22:15:00 220.0
2020-03-06 23:35:00 219.0
2020-03-07 06:15:00 243.0
2020-03-07 07:10:00 244.0
2020-03-07 08:05:00 244.0
2020-03-07 11:20:00 238.0
2020-03-07 12:00:00 236.0
>>> f = 'data/core/csv/43111002_P6m.txt'
>>> reader = PyspcFile(filename=f)
>>> df = reader.read()
>>> df
43111002_P6m
AAAAMMJJHHMM
2017-06-13 17:00:00 0.0
2017-06-13 17:06:00 1.0
2017-06-13 17:12:00 0.2
2017-06-13 17:18:00 0.4
2017-06-13 17:24:00 0.4
2017-06-13 17:30:00 0.4
2017-06-13 17:36:00 2.2
2017-06-13 17:42:00 11.7
2017-06-13 17:48:00 13.3
2017-06-13 17:54:00 8.2
2017-06-13 18:00:00 6.3
2017-06-13 18:06:00 13.4
2017-06-13 18:12:00 14.2
2017-06-13 18:18:00 12.9
2017-06-13 18:24:00 21.5
2017-06-13 18:30:00 14.1
2017-06-13 18:36:00 17.0
2017-06-13 18:42:00 17.0
2017-06-13 18:48:00 8.0
2017-06-13 18:54:00 2.8
2017-06-13 19:00:00 1.8
2017-06-13 19:06:00 6.1
2017-06-13 19:12:00 11.2
2017-06-13 19:18:00 6.6
2017-06-13 19:24:00 3.5
2017-06-13 19:30:00 1.6
2017-06-13 19:36:00 0.4
2017-06-13 19:42:00 0.6
2017-06-13 19:48:00 5.0
2017-06-13 19:54:00 1.8
Cas d'une grandeur en heures
>>> f = 'data/core/csv/K1321810_mohys_QH.txt'
>>> reader = PyspcFile(filename=f)
>>> df = reader.read()
>>> df
K1321810_mohys_QH
AAAAMMJJHH
2018-01-03 12:00:00 118.9
2018-01-03 13:00:00 120.3
2018-01-03 14:00:00 121.4
2018-01-03 15:00:00 122.2
2018-01-03 16:00:00 122.4
2018-01-03 17:00:00 122.2
2018-01-03 18:00:00 121.6
2018-01-03 19:00:00 120.7
2018-01-03 20:00:00 119.7
2018-01-03 21:00:00 118.8
2018-01-03 22:00:00 118.3
2018-01-03 23:00:00 118.8
2018-01-04 00:00:00 120.5
2018-01-04 01:00:00 123.9
2018-01-04 02:00:00 129.5
2018-01-04 03:00:00 137.2
2018-01-04 04:00:00 147.3
2018-01-04 05:00:00 159.8
2018-01-04 06:00:00 172.7
2018-01-04 07:00:00 187.1
2018-01-04 08:00:00 200.7
2018-01-04 09:00:00 213.4
2018-01-04 10:00:00 224.8
2018-01-04 11:00:00 235.8
2018-01-04 12:00:00 245.6
2018-01-04 13:00:00 254.7
2018-01-04 14:00:00 266.9
2018-01-04 15:00:00 273.6
2018-01-04 16:00:00 277.8
2018-01-04 17:00:00 280.1
2018-01-04 18:00:00 280.9
2018-01-04 19:00:00 280.7
2018-01-04 20:00:00 280.2
2018-01-04 21:00:00 280.0
2018-01-04 22:00:00 279.9
2018-01-04 23:00:00 280.1
2018-01-05 00:00:00 280.4
2018-01-05 01:00:00 280.7
2018-01-05 02:00:00 280.9
2018-01-05 03:00:00 280.9
2018-01-05 04:00:00 280.6
2018-01-05 05:00:00 279.9
2018-01-05 06:00:00 278.7
2018-01-05 07:00:00 277.1
2018-01-05 08:00:00 275.5
2018-01-05 09:00:00 273.9
2018-01-05 10:00:00 272.4
2018-01-05 11:00:00 271.0
2018-01-05 12:00:00 269.8
Cas d'une grandeur en jours
>>> f = 'data/core/csv/41005_202006120526_BP_PJ.txt'
>>> reader = PyspcFile(filename=f)
>>> df = reader.read()
>>> df
MoyInf MoySup LocInf LocSup
AAAAMMJJ
2020-06-12 100.0 130.0 200.0 200.0
2020-06-13 7.0 15.0 NaN NaN
2020-06-14 0.0 0.0 NaN NaN
Cas d'une grandeur en mois
>>> f = 'data/core/csv/K1341810_QM.txt'
>>> reader = PyspcFile(filename=f)
>>> df = reader.read()
>>> df
K1341810_QM
AAAAMM
2017-08-01 1.4
2017-09-01 2.1
2017-10-01 1.4
2017-11-01 5.5
2017-12-01 53.6
2018-01-01 156.0
2018-02-01 59.8
2018-03-01 60.9
2018-04-01 27.8
2018-05-01 9.5
2018-06-01 11.1
2018-07-01 2.6
2018-08-01 1.1
"""
df = pnd.read_csv(
self.filename,
sep=';',
header=0,
index_col=0,
na_values=[-9.99, -99.9, -999.999, ''],
keep_default_na=True,
parse_dates=True,
date_format=dtfmt(self.tdelta),
# date_parser=lambda x: dt.strptime(x, dtfmt(self.tdelta))
)
self._check_dataframe(df=df, strict=strict)
return df
[docs]
def write(self, data=None, dirname='.', strict=False):
"""
Ecrire le fichier de données GRP Data (GRP *Calage*)
Parameters
----------
data : pandas.DataFrame
Tableau des données d'observation de GRP 2016
dirname : str
Répertoire local
strict : bool
Lecture avec vérification de la variable (défaut: False)
"""
_exception.check_dataframe(data)
self._check_dataframe(df=data, strict=strict)
if self.filename is None:
raise NotImplementedError
self.filename = os.path.join(dirname, self.filename)
return data.to_csv(
self.filename,
sep=';',
float_format='%.3f',
index=True,
date_format=dtfmt(self.tdelta),
lineterminator='\n'
)
[docs]
@classmethod
def get_varnames(cls):
"""
Définir le nom de la variable
"""
return sorted(list(SPC_VARNAMES.keys()))
[docs]
@staticmethod
def split_basename(filename=None, sep=None,
forceobs=False, forcesim=False):
"""
Extraire les informations depuis le nom du fichier
csv multi-colonnes
Parameters
----------
filename : str
Fichier de données csv multi-colonnes
sep : str
Séparateur des champs. Défaut: '_'
forceobs : bool
Forcer la conversion en tant que série d'observation.
Défaut: False
forcesim : bool
Forcer la conversion en tant que série de simulation.
Défaut: False
L'option forceobs a la préséance sur forcesim
Returns
-------
station : str
Identifiant de la station
varname : str
Nom de la variable
simulation : str, None
Nom de la simulation
runtime : str, None
Instant de production de la prévision
model : str, None
Modèle de prévision
scen : str, None
Scénario de prévision
uncert : str, None
Incertitude de prévision
See Also
--------
pyspc.core.keyseries.str2tuple
Examples
--------
>>> from datetime import datetime as dt
>>> from pyspc.core.csvfile import PyspcFile
Cas d'une série d'obserbation
>>> f = 'data/core/csv/K1321810_QH'
>>> info = PyspcFile.split_basename(filename=f)
>>> info
('K1321810', 'QH', None, None, None, None, None)
Cas d'une série de simulation
>>> f = 'data/core/csv/K1321810_mohys_QH'
>>> info = PyspcFile.split_basename(filename=f)
>>> info
('K1321810', 'QH', 'mohys', None, None, None, None)
Cas d'une série de prévision
>>> f = 'data/core/csv/K0253030_2020061200_GR6J_QJ'
>>> info = PyspcFile.split_basename(filename=f)
>>> info
('K0253030', 'QJ', None, dt(2020, 6, 12, 0, 0), 'GR6J', None, None)
"""
if filename is None:
return None, None, None, None, None, None, None
basename = os.path.splitext(os.path.basename(filename))[0]
key = str2tuple(
s=basename, sep=sep, forceobs=forceobs, forcesim=forcesim)
# CAS OBSERVATION
if key[-1] is None:
return key[0], key[1], None, None, None, None, None
# CAS SIMULATION
if isinstance(key[-1], str):
return key[0], key[1], key[2], None, None, None, None
# CAS PREVISION
if isinstance(key[-1], tuple):
return (key[0], key[1], None,
key[2][0], key[2][1], key[2][2], key[2][3])
# CAS INCORRECT
raise ValueError("Nom de fichier incorrect")
[docs]
@staticmethod
def join_basename(station=None, varname=None, simulation=None,
runtime=None, model=None, scen=None, uncert=None):
"""
Définir le nom du fichier de données GRP Data (GRP *Calage*)
à partir des informations
Parameters
----------
station : str
Identifiant de la station
varname : str
Nom de la variable
simulation : str, None
Nom de la simulation
runtime : str, None
Instant de production de la prévision
model : str, None
Modèle de prévision
scen : str, None
Scénario de prévision
uncert : str, None
Incertitude de prévision
Returns
-------
filename : str
Fichier de données GRP Data (GRP *Calage*)
"""
# CAS OBSERVATION
if simulation is None and runtime is None:
return f'{tuple2str((station, varname, None))}.txt'
# CAS SIMULATION
if simulation is not None:
return f'{tuple2str((station, varname, simulation))}.txt'
# CAS PREVISION
p = tuple2str((station, varname, (runtime, model, scen, uncert)))
return f"{p}.txt"