Code source de pyspc.core.samples

#!/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 - Statistiques."""
from collections import namedtuple
import itertools
from operator import attrgetter
import os.path
import matplotlib.pyplot as mplt
import matplotlib.lines as mlines
import pandas as pnd

import pyspc.core.exception as _exception
from pyspc.core.common import BasicSerie, BasicDict
from pyspc.statistics.freq import from_rank, to_period, to_ugumbel
from pyspc.statistics.period import RETURN_PERIODS, to_freq
from pyspc.plotting.colors import COLORS, TABCOLORS
from pyspc.plotting.markers import MARKERS

# %% Item
# %%% SampleItem
SampleItem = namedtuple(
    'SampleElement',
    ['date', 'value', 'excluded'],
    # Remplir de droite à gauche, donc seulement pour excluded
    defaults=[False]
)
"""Élement d'un échantillon statistique."""


# %% List of items
[docs] class Sample(BasicSerie): """ Structure d'un échantillon statistique. Attributes ---------- df : pnd.DataFrame Vue sous forme de tableau 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 units : str Unité de la grandeur dtfmt : str Format de la date np_dtype : str Type de données de la grandeur 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 expanded_columns : bool Extension des colonnes demandées? (ajout: rang, fréq, temps retour) name : str Libellé de l'échantillon items : list of SampleItem Valeurs de l'échantillon See Also -------- pyspc.core.samples.SampleItem pyspc.core.common.BasicSerie """
[docs] def __init__(self, items=None, name='sample', code=None, provider=None, varname=None): """ Initialise l'instance de la classe Sample. Parameters ---------- items : list of SampleItem Items de l'échanitllon name : str Libellé de l'échantillon. Par défaut: 'sample' code : str Lieu de l'échantillon provider : str, Provider Producteur de l'échantillon varname : str, Parameter Grandeur de l'échantillon See Also -------- pyspc.core.samples.SampleItem pyspc.core.common.BasicSerie """ # ===================================================================== # Initialisation # ===================================================================== super().__init__(code=code, varname=varname, provider=provider) self.name = name # ===================================================================== # Autres méta-données # ===================================================================== self.infer_params = None # ===================================================================== # Création du contenu # ===================================================================== if items is not None: if isinstance(items, SampleItem): items = [items] self.items = items else: self.items = [] self.df_view()
def __str__(self): """Afficher les méta-données de l'instance Sample.""" strinfer = '' if self.infer_params is not None: raise NotImplementedError text = """ ************************************* *********** SAMPLE ****************** ************************************* * NOM ECHANTILLON = {_name} * NOM VARIABLE SPC = {_spc_varname} * INTITULE VARIABLE = {_long_varname} * IDENTIFIANT = {_code} * FOURNISSEUR = {_provider} * NOM VARIABLE = {_varname} * TAILLE ECHANTILLON = {_length} * PREMIERE DATE = {_firstdt} * DERNIERE DATE = {_lastdt} * ECHANTILLON \n{_df}\n {strinfer} ************************************* """ return text.format(**vars(self), strinfer=strinfer) @property def name(self): """Nom de la collection.""" return self._name @name.setter def name(self, name): """Définir le nom de la collection.""" if isinstance(name, str): self._name = name @property def items(self): """Liste des items de l'échantillon.""" return self._items @items.setter def items(self, items): """Définir la liste des items de l'échantillon.""" if _exception.check_listlike(items): self._items = [i for i in items if isinstance(i, SampleItem)] else: self._items = [] @property def df(self): """Contenu de l'échantillon.""" return self._df @df.setter def df(self, df): """Définir le contenu de l'échantillon.""" if isinstance(df, pnd.DataFrame): self._df = df if df.empty: self._firstdt = None self._lastdt = None else: self._firstdt = min(df['date']).to_pydatetime() self._lastdt = max(df['date']).to_pydatetime() self._length = len(df.index) if self.infer_params is not None: self.infer(params=self.infer_params) else: self._df = None @property def firstdt(self): """Première date de l'échantillon.""" return self._firstdt @property def lastdt(self): """Dernière date de l'échantillon.""" return self._lastdt @property def length(self): """Profondeur temporelle de l'échantillon.""" return self._length
[docs] def append(self, item=None): """ Ajouter un élément dans l'échantillon. Parameter --------- item : pyspc.core.samples.SampleItem Item à ajouter See Also -------- pyspc.core.samples.Sample.extend """ if isinstance(item, SampleItem): self.items.append(item) self.df_view()
[docs] def df_view(self): """Créer la vue sous forme de pandas.DataFrame.""" df = pnd.DataFrame.from_records(self.items, columns=SampleItem._fields) self.df = df
[docs] def extend(self, items=None): """ Ajouter plusieurs éléments dans l'échantillon. Parameter --------- item : list of pyspc.core.samples.SampleItem Items à ajouter See Also -------- pyspc.core.samples.Sample.append """ if _exception.check_listlike(items): self.items.extend([i for i in items if isinstance(i, SampleItem)]) self.df_view()
[docs] def infer(self, params=None): """ Réaliser une inférence empirique sur les données. Parameters ---------- params : dict Paramètres d'inférence empirique. Voir les arguments des différentes fonctions utilisées et listées ci-après. Notes ----- L'inférence conduit à ajouter des informations à l'échantillon. Examples -------- >>> sample.df date value excluded 0 1994-11-05 17:29:00 1030.0 True 1 1996-01-24 02:57:00 491.0 False 2 1996-11-13 09:05:00 2060.0 False 3 1997-12-19 22:01:00 280.0 False 4 1999-05-18 12:30:00 1010.0 False 5 1999-10-21 15:00:00 313.0 False 6 2000-10-14 09:36:00 527.0 False 7 2001-10-20 23:13:00 1010.0 False 8 2002-11-25 09:10:00 936.0 False 9 2003-12-02 15:10:00 1740.0 False 10 2004-11-04 23:40:00 367.0 False 11 2006-04-10 22:50:00 110.0 False 12 2006-11-18 10:10:00 146.0 False 13 2008-05-29 19:30:00 257.0 False 14 2008-11-02 10:40:00 2750.0 False 15 2010-06-16 19:30:00 226.0 False 16 2010-11-01 03:50:00 503.0 False 17 2011-11-05 08:10:00 557.0 False 18 2013-05-19 03:30:00 537.0 False 19 2014-01-20 05:30:00 377.0 False 20 2014-11-05 00:40:00 620.0 False 21 2016-04-07 01:30:00 88.4 False 22 2016-11-23 02:30:00 1030.0 False 23 2018-05-16 04:35:00 227.0 False 24 2018-11-10 01:05:00 293.0 False 25 2019-11-23 17:40:00 1260.0 False 26 2021-05-11 09:35:00 539.0 False 27 2021-12-29 16:20:00 117.0 False 28 2023-05-14 03:20:00 117.0 False 29 2024-03-10 08:47:30 818.0 False >>> params = { ... 'freq': {'method': 'Hazen'}, ... 'period': {'highflow': True, 'asint': False}, ... } >>> sample.infer(params=params) >>> sample.df date value excluded rank freq period ugumbel 0 2016-04-07 01:30:00 88.4 False 1.0 0.017 1.017 1.401 1 2006-04-10 22:50:00 110.0 False 2.0 0.051 1.054 1.085 2 2021-12-29 16:20:00 117.0 False 3.0 0.086 1.094 0.896 3 2023-05-14 03:20:00 117.0 False 4.0 0.120 1.137 0.748 4 2006-11-18 10:10:00 146.0 False 5.0 0.155 1.183 0.622 5 2010-06-16 19:30:00 226.0 False 6.0 0.189 1.234 0.508 6 2018-05-16 04:35:00 227.0 False 7.0 0.224 1.288 0.402 7 2008-05-29 19:30:00 257.0 False 8.0 0.258 1.348 0.301 8 1997-12-19 22:01:00 280.0 False 9.0 0.293 1.414 0.204 9 2018-11-10 01:05:00 293.0 False 10.0 0.327 1.487 0.109 10 1999-10-21 15:00:00 313.0 False 11.0 0.362 1.567 0.015 11 2004-11-04 23:40:00 367.0 False 12.0 0.396 1.657 0.078 12 2014-01-20 05:30:00 377.0 False 13.0 0.431 1.757 0.172 13 1996-01-24 02:57:00 491.0 False 14.0 0.465 1.870 0.268 14 2010-11-01 03:50:00 503.0 False 15.0 0.500 2.000 0.366 15 2000-10-14 09:36:00 527.0 False 16.0 0.534 2.148 0.467 16 2013-05-19 03:30:00 537.0 False 17.0 0.568 2.320 0.572 17 2021-05-11 09:35:00 539.0 False 18.0 0.603 2.521 0.683 18 2011-11-05 08:10:00 557.0 False 19.0 0.637 2.761 0.799 19 2014-11-05 00:40:00 620.0 False 20.0 0.672 3.052 0.924 20 2024-03-10 08:47:30 818.0 False 21.0 0.706 3.411 1.058 21 2002-11-25 09:10:00 936.0 False 22.0 0.741 3.866 1.206 22 1999-05-18 12:30:00 1010.0 False 23.0 0.775 4.461 1.371 23 2001-10-20 23:13:00 1010.0 False 24.0 0.810 5.272 1.559 24 1994-11-05 17:29:00 1030.0 True NaN NaN NaN NaN 25 2016-11-23 02:30:00 1030.0 False 25.0 0.844 6.444 1.780 26 2019-11-23 17:40:00 1260.0 False 26.0 0.879 8.285 2.050 27 2003-12-02 15:10:00 1740.0 False 27.0 0.913 11.600 2.406 28 1996-11-13 09:05:00 2060.0 False 28.0 0.948 19.333 2.935 29 2008-11-02 10:40:00 2750.0 False 29.0 0.982 58.000 4.051 See Also -------- pyspc.statistics.freq.RANK2FREQ pyspc.statistics.freq.from_rank pyspc.statistics.freq.to_period pyspc.statistics.freq.to_ugumbel """ self.df_view() if params is None: params = {} # RANG params['rank'] = True self.sort(by='value', reverse=False) self.df['rank'] = self.df.value[self.df.excluded == 0].rank( ascending=True, method='first') # FREQUENCE if 'freq' not in params: params['freq'] = {'method': 'Hazen'} n = self.df['rank'].max() self.df['freq'] = self.df['rank'].apply( from_rank, args=(n,), **params['freq'], check=False) # TEMPS DE RETOUR if 'period' not in params: params['period'] = {'highflow': True, 'asint': False} self.df['period'] = self.df['freq'].apply( to_period, **params['period'], check=False) # U GUMBEL params['ugumbel'] = True self.df['ugumbel'] = self.df['freq'].apply(to_ugumbel) self.infer_params = params
[docs] def sort(self, by=None, reverse=None): """ Trier les items par ordre chronologique, ou valeurs. Parameters ---------- by : str Choix du tri parmi ['date', 'value']. reverse : bool Tri par ordre décroissant. Par défaut: False. Notes ----- Un tri par rang revient à trier par valeur croissante. """ if by is None: by = 'date' if reverse is None: reverse = False try: self.items = sorted( self.items, key=attrgetter(by), reverse=reverse) except AttributeError: _exception.Warning( msg=f"Impossible de trier l'échantillon par '{by}'. " f"Veuillez choisir un champ parmi {SampleItem._fields}") self.df_view()
[docs] def to_Hydroportail(self, dirname=None): """ Export au format Hydroportail. Parameters ---------- dirname : str Répertoire d'export Returns ------- filenames : dict Fichiers écrits renvoyés sous la forme de dictionnaire. Clé = clé de l'échantillon, valeur = nom du fichier Examples -------- >>> sample ************************************* *********** SAMPLE ****************** ************************************* * NOM ECHANTILLON = test_export_hydroportail * NOM VARIABLE SPC = QI * INTITULE VARIABLE = Débit instantané * IDENTIFIANT = K0550010 * FOURNISSEUR = Provider(name=None) * NOM VARIABLE = QI * TAILLE ECHANTILLON = 30 * PREMIERE DATE = 1994-11-05 17:29:00 * DERNIERE DATE = 2024-03-10 08:47:30 ************************************* >>> filename = sample.to_Hydroportail(dirname='data') >>> filename 'data/Q-X_None_K0550010_test_export_hydroportail_Echantillon.csv' """ from pyspc.io.hydroportail import write_Hydroportail return write_Hydroportail( data=self, datatype='hp_sample', dirname=dirname)
[docs] @classmethod def from_dve(cls, dates=None, values=None, exclusions=None, name=None, code=None, varname=None, provider=None): """ Créer un échantillon à partir de dates, valeurs et exclusions. Parameters ---------- dates : list Dates des éléments de l'échantillon values : list Valeurs des éléments de l'échantillon exclusions : list Indication s'il faut exclure la valeur dans les analyses Other Parameters ---------------- name : str Libellé de l'échantillon. Par défaut: 'sample' code : str Lieu de l'échantillon provider : str, Provider Producteur de l'échantillon varname : str, Parameter Grandeur de l'échantillon Returns ------- sample : pyspc.core.samples.Sample Échantillon """ # =================================================================== # 0- CONTROLE # =================================================================== _exception.check_listlike(dates) _exception.check_listlike(values) if exclusions is None: exclusions = [False] * len(dates) _exception.raise_valueerror( len(dates) != len(values), "Les dates et les valeurs ne contiennent pas le même nombre " "d'éléments" ) _exception.raise_valueerror( len(dates) != len(exclusions), "Les dates et les exclusions ne contiennent pas le même nombre " "d'éléments" ) # =================================================================== # 1- CREATION ECHANTILLON # =================================================================== items = [SampleItem(d, v, e) for d, v, e in zip(dates, values, exclusions)] sample = Sample( items=items, name=name, code=code, varname=varname, provider=provider) return sample
# %% Dict of List of items
[docs] class Samples(BasicDict): """ Structure d'une collection d'échantillons statistiques. Attributes ---------- datatype : str Type de la collection name : str Nom de la collection. Par défaut: 'samples' See Also -------- pyspc.core.samples.Sample pyspc.core.common.BasicDict """
[docs] def __init__(self, datatype=None, name='samples'): """ Initialise l'instance de la classe Samples. Parameters ---------- datatype : str Type de la collection name : str Nom de la collection. Par défaut: 'samples' See Also -------- pyspc.core.samples.Sample pyspc.core.common.BasicDict """ super().__init__(datatype=datatype, name=name)
# self.check_datatype(datatype) def __str__(self): """Afficher les méta-données de l'instance Samples.""" strseries = '' if len(self.keys()) > 0: counter = 0 for key in self.keys(): counter += 1 strseries += '\n * ----------------------------------' strseries += f"\n * SAMPLE #{counter} : {key}" text = """ ************************************* ********** SAMPLES ****************** ************************************* * NOM DE LA COLLECTION = {_name} * TYPE DE COLLECTION = {_datatype} * NOMBRE DE SERIES = {length} {strseries} ************************************* """ return text.format(**vars(self), length=len(self.keys()), strseries=strseries)
[docs] def update(self, other, overwrite=True): """ Ajouter des éléments d'une autre instance. Parameters ---------- other : pyspc.core.samples.Samples Collection d'échantillons statistiques overwrite : bool Écraser la donnée existante ? défaut: False """ _exception.raise_valueerror(not isinstance(other, Samples)) for k, v in other: self.add(v, k, overwrite=overwrite)
[docs] def add(self, sample=None, key=None, overwrite=False): """ Ajouter un échantillon dans la collection. Parameters ---------- sample : pyspc.core.samples.Sample Échantillon statistique. key : str, None Clé d'identification de l'échantillon. Si non défini, la clé sera sample.name. overwrite : bool Écraser la donnée existante ? défaut: False """ # --------------------------------------------------------------------- # 0- Contrôles # --------------------------------------------------------------------- _exception.raise_valueerror( not isinstance(sample, Sample), 'Sample incorrecte' ) if key is None: key = sample.name _exception.raise_valueerror(not isinstance(key, str), 'Clé incorrecte') # --------------------------------------------------------------------- # 1- Ajout # --------------------------------------------------------------------- if overwrite: self[key] = sample else: self.setdefault(key, sample)
[docs] def extend(self, samples=None, overwrite=False): """ Alimenter la collection à partir d'une autre collection. Parameters ---------- samples : pyspc.core.samples.Samples Collection d'échantillons statistiques overwrite : bool Écraser la donnée existante ? défaut: False """ # --------------------------------------------------------------------- # 0- Contrôles # --------------------------------------------------------------------- _exception.raise_valueerror( not isinstance(samples, type(self)), 'Collection de Sample incorrecte' ) # --------------------------------------------------------------------- # 1- Ajout # --------------------------------------------------------------------- for key, sample in samples.items(): self.add(key=key, sample=sample, overwrite=overwrite)
[docs] def infer(self, params=None): """ Trier les échantillons par ordre chronologique, ou valeurs. Parameters ---------- params : dict Paramètres d'inférence empirique. Voir les arguments des différentes fonctions utilisées et listées ci-après. See Also -------- pyspc.core.samples.Sample.infer """ for key in self.keys(): self[key].infer(params=params)
[docs] def sort(self, by=None, reverse=None): """ Trier les échantillons par ordre chronologique, ou valeurs. Parameters ---------- by : str Choix du tri parmi ['date', 'value']. reverse : bool Tri par ordre décroissant. Par défaut: False. Notes ----- Un tri par rang revient à trier par valeur croissante. See Also -------- pyspc.core.samples.Sample.sort """ for key in self.keys(): self[key].sort(by=by, reverse=reverse)
[docs] def plot_gumbel_paper(self, config=None, filename=None, dirname=None): """ Tracer une figure similaire à un papier de Gumbel. Parameters ---------- dirname : str Répertoire d'enregistrement de la figure. filename : str Nom du fichier à enregister. Si non défini, le nom de fichier est la concaténation de dirname et de l'attribut name de la collection. config : Config, dict, filename Configuration de la figure et des courbes. Les clés des options des courbes correspondent au keyseries. Voir aussi pyspc.core.keyseries Les valeurs correspondent aux éléments à définir: - color - marker - markersize """ # --------------------------------------------------------------------- # 0- Contrôles # --------------------------------------------------------------------- dpi = 300 if filename is None: filename = self.name + '.png' if dirname is None: dirname = '.' filename = os.path.join(dirname, filename) iter_colors = itertools.cycle( TABCOLORS) if len(self) <= len( TABCOLORS) else itertools.cycle(COLORS) iter_markers = itertools.cycle(MARKERS) # --------------------------------------------------------------------- # 2- Création de la figure et des zones graphiques # --------------------------------------------------------------------- fig, ax, ax2 = _plot_create_fig(dpi) # --------------------------------------------------------------------- # 3- Création des points # --------------------------------------------------------------------- handles = [] for key, sample in self.items(): color = config.get(key, {}).get('color', next(iter_colors)) marker = config.get(key, {}).get('marker', next(iter_markers)) markersize = config.get(key, {}).get('maerkersize', 20) if sample.infer_params is None: sample.infer() # move dots on top of line : zorder=2.5 (Line2D a un zorder de 2) ax.scatter(sample.df['ugumbel'], sample.df['value'], c=color, marker=marker, s=markersize, zorder=2.5) handles.append( mlines.Line2D([], [], color=color, marker=marker, linestyle='None', markersize=5, label=sample.name) ) # --------------------------------------------------------------------- # 4- AXE PRINCIPAL (incluant la légende) # --------------------------------------------------------------------- _plot_custom_ax(ax, handles) # ------------------------------------ # 5- AXE SECONDAIRE (période de retour) # ------------------------------------ _plot_custom_ax2(ax2, ax) # --------------------------------------------------------------------- # 6- Titre # --------------------------------------------------------------------- fig.suptitle("Inférence empirique", fontsize=16) # --------------------------------------------------------------------- # 7- Enregistrement de la figure # --------------------------------------------------------------------- fig.savefig(filename, dpi=dpi) mplt.close(fig) fig = None return filename
[docs] def to_Hydroportail(self, dirname=None): """ Export au format Hydroportail. Parameters ---------- dirname : str Répertoire d'export Returns ------- filenames : dict Fichiers écrits renvoyés sous la forme de dictionnaire. Clé = clé de l'échantillon, valeur = nom du fichier Examples -------- >>> samples ************************************* ********** SAMPLES ****************** ************************************* * NOM DE LA COLLECTION = test_sort * TYPE DE COLLECTION = None * NOMBRE DE SERIES = 1 * ---------------------------------- * SAMPLE #1 : Bas ************************************* >>> filenames = samples.to_Hydroportail(dirname='data') >>> filenames {'Bas': 'data/Q-X_None_K0550010_Bas_Echantillon.csv'} """ return {k: s.to_Hydroportail(dirname=dirname) for k, s in self.items()}
[docs] @classmethod def from_dve(cls, dve=None, name='samples', datatype=None): """ Créer un échantillon à partir de dates, valeurs et exclusions. Parameters ---------- dve : dict Données à insérer en tant qu'échantillons statistiques. datatype : str Type de la collection name : str Nom de la collection. Par défaut: 'samples' Returns ------- samples : pyspc.core.samples.Samples Échantillons Notes ----- ``dve`` est un dictionnaire où la clé sera la clé de l'échantillon dans la collection créée et où la valeur est elle-même un dictionnaire : - 'dates' : liste des dates - 'values' : liste des valeurs - 'exclusions' : liste des exclusions - 'name' : nom de l'échantillon - 'code' : identifiant du lieu - 'varname' : grandeur physique - 'provider' : fournisseur de la donnée See Also -------- pyspc.core.samples.Sample.from_dve """ samples = Samples(name=name, datatype=datatype) for key, content in dve.items(): sample = Sample.from_dve(**content) samples.add(sample=sample, key=key) return samples
# %% Plotting functions common to pyspc.core.statitics def _plot_create_fig(dpi): """Créer la figure Papier de Gumbel.""" fig = mplt.figure(dpi=dpi) ax = fig.add_axes([0.08, 0.19, 0.90, 0.70]) ax2 = ax.twiny() return fig, ax, ax2 def _plot_custom_ax(ax, handles): """Configure l'axe principal (variable de Gumbel).""" ax.set_ylabel("$m^3/s$", fontsize=6) ax.set_xlabel("Variable de Gumbel", fontsize=6) ax.tick_params(axis='both', which='major', labelsize=6) ax.grid(True, color='#E1E1E1') ax.legend(handles=handles, loc='upper left', fontsize=6, bbox_to_anchor=(0.01, 0.99), fancybox=True, shadow=True) def _plot_custom_ax2(ax2, ax): """Configure l'axe secondaire (période de retour).""" # Move twinned axis ticks and label from top to bottom ax2.xaxis.set_ticks_position("bottom") ax2.xaxis.set_label_position("bottom") # Offset the twin axis below the host ax2.spines["bottom"].set_position(("axes", -0.15)) # Turn on the frame for the twin axis, but then hide all # but the bottom spine ax2.set_frame_on(True) ax2.patch.set_visible(False) for sp in ax2.spines.values(): sp.set_visible(False) ax2.spines["bottom"].set_visible(True) ax2.spines["bottom"].set_color('tab:grey') ax2.set_xticks([to_ugumbel(to_freq(tr)) for tr in RETURN_PERIODS]) ax2.set_xticklabels([str(tr) for tr in RETURN_PERIODS]) ax2.set_xlabel("Période de retour", fontsize=6) ax2.xaxis.label.set_color('tab:grey') ax2.tick_params( axis='both', which='major', labelsize=6, colors='tab:grey') ax2.set_xlim(ax.get_xlim())