#!/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 - Série de données
"""
from datetime import datetime as dt
import numpy as np
import pandas as pnd
from pandas.tseries.frequencies import to_offset
import pytz
import pyspc.core.exception as _exception
from pyspc.core.common import BasicSerie
from pyspc.core._serie import (
BinaryOperator, # méthodes des opérateurs binaires
Combining, # méthodes de mélanges, mises-à-jour
Computation, # méthodes de calculs
Copy, # méthodes de copie
Event, # méthodes de détermination d'événements
Modeling, # méthodes de modélisation
Plotting, # méthodes de figures
Reindexing, # méthodes de re-indexation temporelle
Scaling, # méthodes de changement d'échelle temporelle
TimeSerie, # méthodes de traitement temporel
)
[docs]
class Serie(BasicSerie, BinaryOperator, Combining, Computation, Copy, Event,
Modeling, Plotting, Reindexing, Scaling, TimeSerie):
"""
Structure d'une Série de données
Attributes
----------
data_frame : pnd.DataFrame
Série temporelle
code : str
Lieu de la série
location : Location
Lieu de la série
parameter : Parameter
Grandeur de la série
varname : str
Grandeur de la série
spc_varname : str
Grandeur de la série selon la convention de pyspc
long_varname : str
Intitulé de la grandeur
timestep : timedelta
Pas de temps de la grandeur
timeunits : str
Unité de temps de la grandeur
units : str
Unité de la grandeur
dtfmt : str
Format de la date
np_dtype : str
Type de données de la grandeur
missing : float/int
Valeur manquante 'missing'
timezone : pytz
Fuseau horaire
warning : bool
Afficher les avertissements au lieu de lever des erreurs
défaut: True
firstdt : datetime
Première date de la série de données
lastdt : datetime
Dernière date de la série de données
length : int
Profondeur temporelle de la série de données
See Also
--------
pyspc.core.common.BasicSerie
pyspc.core.parameter.Parameter
pyspc.core.location.Location
pyspc.core.provider.Provider
"""
[docs]
def __init__(self, datval, code=None, provider=None, varname=None,
fill=True, missing=None, timezone=pytz.utc, strict=False,
warning=True):
"""
Initialiser l'instance de la classe Serie
Parameters
----------
datval : list, pnd.DataFrame
Ensemble de dates et de valeurs
- ['aaaamm[jj[hh[MM]]]', 'valeur'] avec '.' comme sép. décimal
- [datetime(aaaa, mm, jj, hh, MM), valeur]
- Série de données sous forme de DataFrame
code : str
Lieu de la série
provider : str, Provider
Producteur de la série
varname : str, Parameter
Grandeur de la série
fill : bool
Etendre la période et remplir avec valeur manquante ? défaut: True
missing : float/int
Valeur manquante 'missing'
overwrite : bool
Remplacer les données existantes ?
timezone : pytz
Fuseau horaire
strict : bool
Contrôler la cohérence entre le pas de temps des données
et le pas de temps de la grandeur. Défaut: False
warning : bool
Afficher les avertissements au lieu de lever des erreurs
défaut: True
"""
# =====================================================================
# Définition du lieu (code, location) : voir property/setter
# Définition du paramètre
# =====================================================================
super().__init__(code=code, varname=varname, provider=provider,
missing=missing)
# self.code = code
# self.parameter = (varname, provider)
# self.missing = missing
# =====================================================================
# Autres méta-données
# =====================================================================
self._fill = fill
self._timezone = timezone
self._warning = warning
# =====================================================================
# Création du pandas.DataFrame
# =====================================================================
# Créer le dataframe
self.data_frame = self._create_dframe(datval)
if self.data_frame is not None:
# Traitement des données manquantes dans la série
for m in self.missing:
try:
self.data_frame[self.data_frame == m] = np.nan
except ValueError:
pass
# Enlever les instants dupliqués
self.data_frame = \
self.data_frame[~self.data_frame.index.duplicated(keep='last')]
if self.timestep is not None and fill and strict: #
try:
itd = pnd.infer_freq(self.data_frame.index)
except (TypeError, ValueError):
pass
else:
_exception.raise_valueerror(
pnd.to_timedelta(to_offset(itd)) == self.timestep,
f"Le pas de temps de la grandeur {self.spc_varname} "
f"'{self.timestep}' ne correspond pas au pas de temps "
"de la série de donnée "
f"'{pnd.to_timedelta(to_offset(itd))}'"
)
# Re-index pour affecter la valeur manquante
# aux pas de temps manquants
# + tri chronologique
if fill and not self.spc_varname.endswith('I'):
self.reindex()
# Si pas de reindex, tri chronologique
else:
self.data_frame.sort_index(inplace=True)
# Nom de la colonne des données : (lieu, grandeur)
self.data_frame.columns = [(self.code, self.spc_varname)]
# =====================================================================
# Définition de la période temporelle
# =====================================================================
self._length = len(self.data_frame.index)
self._firstdt = self.data_frame.index[0].to_pydatetime()
self._lastdt = self.data_frame.index[-1].to_pydatetime()
else:
self._length = None
self._firstdt = None
self._lastdt = None
def __str__(self):
"""
Afficher les méta-données de l'instance Serie
"""
text = """
*************************************
*********** SERIE *******************
*************************************
* NOM VARIABLE SPC = {_spc_varname}
* INTITULE VARIABLE = {_long_varname}
* IDENTIFIANT = {_code}
* FOURNISSEUR = {_provider}
* NOM VARIABLE = {_varname}
* UNITE = {_units}
* VALEUR MANQUANTE = {_missing}
* SERIE CONTINUE = {_fill}
* PAS DE TEMPS = {_timestep}
* UNITE DE TEMPS = {_timeunits}
* FUSEAU HORAIRE = {_timezone}
* PROFONDEUR SERIE = {_length}
* PREMIER PAS DE TEMPS = {_firstdt}
* DERNIER PAS DE TEMPS = {_lastdt}
*************************************
"""
return text.format(**vars(self))
@property
def data_frame(self):
"""Contenu temporel de la série"""
return self._data_frame
@data_frame.setter
def data_frame(self, df):
"""Définir le contenu temporel de la série"""
if isinstance(df, pnd.DataFrame):
self._data_frame = df
else:
self._data_frame = None
@property
def df(self):
"""Contenu temporel de la série. Alias de self.data_frame"""
return self._data_frame
@df.setter
def df(self, df):
"""Définir le contenu temporel de la série"""
if isinstance(df, pnd.DataFrame):
self._data_frame = df
else:
self._data_frame = None
@BasicSerie.code.setter
def code(self, code):
# ---------------------------------------------------------------------
# Application du setter du Parent
# ---------------------------------------------------------------------
BasicSerie.code.fset(self, code)
# ---------------------------------------------------------------------
# Propagation de la modification dans le data_frame
# ---------------------------------------------------------------------
try:
self.data_frame.columns = [(self.code, self.spc_varname)]
except AttributeError:
pass
@property
def fill(self):
"""Données continues"""
return self._fill
@property
def warning(self):
"""Afficher les avertissements"""
return self._warning
def _create_dframe(self, datval):
"""
Créer le dataframe
Parameters
----------
datval : list, pnd.DataFrame
Ensemble de dates et de valeurs
- ['aaaamm[jj[hh[MM]]]', 'valeur'] avec '.' comme sép. décimal
- [datetime(aaaa, mm, jj, hh, MM), valeur]
- Série de données sous forme de DataFrame
Returns
-------
data_frame : pnd.DataFrame
Série temporelle
"""
# si <datval> est une liste non-vide
if isinstance(datval, list) and len(datval) > 0:
dates = [dt.strptime(x[0], self.dtfmt)
if not isinstance(x[0], dt) else x[0]
for x in datval]
values = [x[1] if x[1] not in self.missing else np.nan
for x in datval]
values = np.array(values, dtype=self.np_dtype)
return pnd.DataFrame(values, index=dates)
if isinstance(datval, pnd.DataFrame):
return datval
if isinstance(datval, pnd.Series):
return datval.to_frame()
_exception.raise_valueerror(
True,
'La série de donnée est mal formatée pour être convertie',
self.warning
)
return None
@property
def firstdt(self):
"""Première date de la série de données"""
return self._firstdt
@property
def lastdt(self):
"""Dernière date de la série de données"""
return self._lastdt
@property
def length(self):
"""Profondeur temporelle de la série de données"""
return self._length
@property
def _constructor_expanddim(self):
"""Importer Series pour éviter des imports circulaires"""
from pyspc.core.series import Series
return Series