#!/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
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.core.samples import Sample
from pyspc.core.samples import (
_plot_create_fig, _plot_custom_ax, _plot_custom_ax2)
from pyspc.statistics.freq import to_ugumbel
from pyspc.statistics.period import to_freq
from pyspc.plotting.colors import COLORS, TABCOLORS
from pyspc.plotting.markers import MARKERS
# %% Item
# %%% StatItem
StatItem = namedtuple(
'StatItem',
['return_period', 'value', 'value_low', 'value_high'],
# Remplir de droite à gauche, donc seulement pour value_low et value_high
defaults=[None, None]
)
"""Élement d'un ajustement statistique."""
# %%% GradexItem
GradexItem = namedtuple(
'GradexItem',
['mm', 'timestep', 'ratio', 'pivot', 'area'],
defaults=[None, None, None, None, None]
)
"""Élement d'un ajustement Gradex sur les précipitations."""
# %% List of items
[docs]
class Stat(BasicSerie):
"""
Structure d'un résultat d'un ajustement 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
name : str
Libellé de l'échantillon
method : str
Méthode d'ajustement
size : int
Taille de l'échantillon
date_range : tuple
Période temporelle de l'échantillon
gradex : GradexItem
Informations du gradex sur la pluviométrie
coverage : int
Couverture de l'intervalle d'incertitude, entre 0 et 100.
sample : Sample
Echantillon à l'origine de l'ajustement
items : list of StatItem
Valeurs de l'ajustement
See Also
--------
pyspc.core.statistics.StatItem
pyspc.core.statistics.GradexItem
pyspc.core.statistics.Sample
pyspc.core.common.BasicSerie
"""
[docs]
def __init__(self, items=None, name='stat',
method=None, length=None, date_range=None, gradex=None,
coverage=None, sample=None,
code=None, provider=None, varname=None):
"""
Initialise l'instance de la classe Sample.
Parameters
----------
name : str
Libellé de l'échantillon
method : str
Méthode d'ajustement
length : int
Taille de l'échantillon
date_range : tuple
Période temporelle de l'échantillon
gradex : GradexItem
Informations du gradex sur la pluviométrie
coverage : int
Couverture de l'intervalle d'incertitude, entre 0 et 100.
sample : Sample
Echantillon à l'origine de l'ajustement. Si fourni, les valeurs de
*length* et *date_range* proviennent de *sample*.
items : list of StatItem
Valeurs de l'ajustement
code : str
Lieu de l'échantillon
provider : str, Provider
Producteur de l'échantillon
varname : str, Parameter
Grandeur de l'échantillon
See Also
--------
pyspc.core.statistics.StatItem
pyspc.core.statistics.GradexItem
pyspc.core.statistics.Sample
pyspc.core.common.BasicSerie
"""
# =====================================================================
# Initialisation
# =====================================================================
super().__init__(code=code, varname=varname, provider=provider)
self.name = name
# =====================================================================
# Autres méta-données
# =====================================================================
self._method = method
self._gradex = gradex if isinstance(gradex, GradexItem) else None
self._coverage = coverage
self._sample = sample if isinstance(sample, Sample) else None
if isinstance(self.sample, Sample):
self._length = self.sample.length
self._date_range = (self.sample.firstdt, self.sample.lastdt)
else:
self._length = length
self._date_range = date_range
# =====================================================================
# Création du contenu
# =====================================================================
if items is not None:
if isinstance(items, StatItem):
items = [items]
self.items = items
else:
self.items = []
self.df_view()
def __str__(self):
"""Afficher les méta-données de l'instance Stat."""
text = """
*************************************
************* STAT ******************
*************************************
* NOM ECHANTILLON = {_name}
* NOM VARIABLE SPC = {_spc_varname}
* INTITULE VARIABLE = {_long_varname}
* IDENTIFIANT = {_code}
* FOURNISSEUR = {_provider}
* NOM VARIABLE = {_varname}
* TAILLE ECHANTILLON = {_length}
* PERIODE ECHANTILLON = {_date_range}
* GRADEX PLUVIOMETRIQUE= {_gradex}
* COUVERTURE INCERT. = {_coverage}
* METHODE AJUSTEMENT = {_method}
* AJUSTEMENT \n{_df}
*************************************
"""
return text.format(**vars(self))
@property
def coverage(self):
"""Couverture de l'intervalle d'incertitude."""
return self._coverage
@property
def date_range(self):
"""Période temporelle de l'échantillon."""
return self._date_range
@property
def gradex(self):
"""Gradex."""
return self._gradex
@property
def length(self):
"""Profondeur temporelle de l'échantillon."""
return self._length
@property
def method(self):
"""Nom de la méthode d'ajustement."""
return self._method
@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
else:
self._name = None
@property
def sample(self):
"""Echantillon."""
return self._sample
@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, StatItem)]
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
else:
self._df = None
[docs]
def append(self, item=None):
"""
Ajouter un élément dans l'ajustement.
Parameters
----------
item : pyspc.core.statistics.StatItem
Item à ajouter
See Also
--------
pyspc.core.statistics.Stat.extend
"""
if isinstance(item, StatItem):
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=StatItem._fields)
self.df = df
[docs]
def extend(self, items=None):
"""
Ajouter plusieurs éléments dans l'ajustement.
Parameters
----------
item : list of pyspc.core.statistics.StatItem
Item à ajouter
See Also
--------
pyspc.core.statistics.Sample.append
"""
if _exception.check_listlike(items):
self.items.extend([i for i in items if isinstance(i, StatItem)])
self.df_view()
[docs]
@classmethod
def from_records(cls, values=None, return_periods=None,
values_low=None, values_high=None,
name=None,
method=None, length=None, date_range=None, gradex=None,
coverage=None, sample=None,
code=None, varname=None, provider=None):
"""
Créer un ajustements à partir de valeurs, périodes de retour.
Parameters
----------
values : list
Valeurs des éléments de l'ajustement
return_periods : list
Temps de retour des éléments de l'ajustement
values_low : list
Valeurs basses des éléments de l'ajustement
values_high : list
Valeurs hautes des éléments de l'ajustement
Other Parameters
----------------
name : str
Libellé de l'échantillon. Par défaut: 'sample'
method : str
Méthode d'ajustement
length : int
Taille de l'échantillon
date_range : tuple
Période temporelle de l'échantillon
gradex : GradexItem
Informations du gradex sur la pluviométrie
coverage : int
Couverture de l'intervalle d'incertitude, entre 0 et 100.
sample : Sample
Echantillon à l'origine de l'ajustement. Si fourni, les valeurs de
*length* et *date_range* proviennent de *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.statistics.Stat
Résultat d'ajustement
"""
# ===================================================================
# 0- CONTROLE
# ===================================================================
_exception.check_listlike(values)
_exception.check_listlike(return_periods)
if values_low is not None:
_exception.check_listlike(values_low)
else:
values_low = [None] * len(values)
if values_high is not None:
_exception.check_listlike(values_high)
else:
values_high = [None] * len(values)
_exception.raise_valueerror(
len(return_periods) != len(values),
"Les périodes de retour et les valeurs ne contiennent pas le même"
" nombre d'éléments"
)
_exception.raise_valueerror(
len(values_low) != len(values),
"Les valeurs et les valeurs basses ne contiennent pas le même"
" nombre d'éléments"
)
_exception.raise_valueerror(
len(values_high) != len(values),
"Les valeurs et les valeurs hautes ne contiennent pas le même"
" nombre d'éléments"
)
# ===================================================================
# 1- CREATION ECHANTILLON
# ===================================================================
items = [StatItem(rp, v, vl, vh)
for rp, v, vl, vh in zip(return_periods, values,
values_low, values_high)]
stat = Stat(
items=items, name=name,
method=method, length=length, date_range=date_range, gradex=gradex,
coverage=coverage, sample=sample,
code=code, varname=varname, provider=provider)
return stat
# %% Dict of List of items
[docs]
class Stats(BasicDict):
"""
Structure d'une collection d'ajustements statistiques.
Attributes
----------
datatype : str
Type de la collection
name : str
Nom de la collection. Par défaut: 'stats'
See Also
--------
pyspc.core.statistics.Stat
pyspc.core.common.BasicDict
"""
[docs]
def __init__(self, datatype=None, name='stats'):
"""
Initialise l'instance de la classe Stats.
Parameters
----------
datatype : str
Type de la collection
name : str
Nom de la collection. Par défaut: 'stats'
See Also
--------
pyspc.core.statistics.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 Stats."""
strseries = ''
if len(self.keys()) > 0:
counter = 0
for key in self.keys():
counter += 1
strseries += '\n * ----------------------------------'
strseries += f"\n * STAT #{counter} : {key}"
text = """
*************************************
************ STATS ******************
*************************************
* 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.statistics.Stats
Collection d'ajustements statistiques
overwrite : bool
Écraser la donnée existante ? défaut: False
"""
_exception.raise_valueerror(not isinstance(other, Stats))
for k, v in other:
self.add(v, k, overwrite=overwrite)
[docs]
def add(self, stat=None, key=None, overwrite=False):
"""
Ajouter un ajustement dans la collection.
Parameters
----------
stat : pyspc.core.statistics.Stat
Ajustement statistique.
key : str, None
Clé d'identification de l'ajustement. Si non défini, la clé sera
stat.name.
overwrite : bool
Écraser la donnée existante ? défaut: False
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
_exception.raise_valueerror(
not isinstance(stat, Stat), 'Stat incorrecte'
)
if key is None:
key = stat.name
_exception.raise_valueerror(not isinstance(key, str), 'Clé incorrecte')
# ---------------------------------------------------------------------
# 1- Ajout
# ---------------------------------------------------------------------
if overwrite:
self[key] = stat
else:
self.setdefault(key, stat)
[docs]
def extend(self, stats=None, overwrite=False):
"""
Alimenter la collection à partir d'une autre collection.
Parameters
----------
stats : pyspc.core.statistics.Stats
Collection d'échantillons statistiques
overwrite : bool
Écraser la donnée existante ? défaut: False
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
_exception.raise_valueerror(
not isinstance(stats, type(self)),
'Collection de Sample incorrecte'
)
# ---------------------------------------------------------------------
# 1- Ajout
# ---------------------------------------------------------------------
for key, stat in stats.items():
self.add(key=key, stat=stat, overwrite=overwrite)
[docs]
def plot_gumbel_paper(self, config=None, filename=None, dirname=None,
ignore_sample=True, ignore_errorbar=True):
"""
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
- linestyle
ignore_sample : bool
Ne pas tracer les échantillons, même si ceux-ci sont disponibles.
Par défaut: True
ignore_errorbar : bool
Ne pas tracer les intervalles d'incertitudes, même si ceux-ci
sont disponibles. Par défaut: True
Returns
-------
filename : str
Nom du fichier image
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
dpi = 300
if config is None:
config = {}
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-A Création des échantillons = Marker
# ---------------------------------------------------------------------
handles = []
if not ignore_sample:
for key, stat in self.items():
sample = stat.sample
if sample is None:
continue
if sample.infer_params is None:
sample.infer()
config.setdefault(key, {})
config[key].setdefault('color', next(iter_colors))
config[key].setdefault('marker', next(iter_markers))
config[key].setdefault('markersize', 5)
ax.scatter(sample.df['ugumbel'], sample.df['value'],
c=config[key]['color'],
marker=config[key]['marker'],
s=config[key]['markersize'],
zorder=1.5, alpha=0.5)
# ---------------------------------------------------------------------
# 3-B Création des ajustements = Line2D+Marker
# ---------------------------------------------------------------------
for key, stat in self.items():
config.setdefault(key, {})
config[key].setdefault('color', next(iter_colors))
config[key].setdefault('marker', next(iter_markers))
config[key].setdefault('markersize', 5)
config[key].setdefault('linestyle', '-')
config[key].setdefault('linewidth', 1)
u = [to_ugumbel((to_freq(t))) for t in stat.df['return_period']]
ax.plot(u, stat.df['value'],
color=config[key]['color'],
linestyle=config[key]['linestyle'],
linewidth=config[key]['linewidth'],
marker=config[key]['marker'],
markersize=config[key]['markersize'])
if (not ignore_errorbar
and not stat.df['value_low'].isnull().values.all()
and not stat.df['value_high'].isnull().values.all()):
ax.errorbar(
u, stat.df['value'],
yerr=[stat.df['value'] - stat.df['value_low'],
stat.df['value_high'] - stat.df['value']],
color=config[key]['color'],
)
handles.append(
mlines.Line2D(
[], [],
color=config[key]['color'],
linestyle=config[key]['linestyle'],
linewidth=config[key]['linewidth'],
marker=config[key]['marker'],
markersize=5,
label=stat.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 statistique", fontsize=16)
# ---------------------------------------------------------------------
# 7- Enregistrement de la figure
# ---------------------------------------------------------------------
fig.savefig(filename, dpi=dpi)
mplt.close(fig)
fig = None
return filename
[docs]
@classmethod
def from_records(cls, records=None, name='stats', datatype=None):
"""
Créer un ajustement à partir de valeurs, périodes de retour.
Parameters
----------
records : dict
Données à insérer en tant qu'ajustements statistiques.
datatype : str
Type de la collection
name : str
Nom de la collection. Par défaut: 'stats'
Returns
-------
samples : pyspc.core.statistics.Stats
Échantillons
Notes
-----
``records`` est un dictionnaire où la clé sera la clé de l'ajustement
dans la collection créée et où la valeur est elle-même un dictionnaire:
- 'values' : liste des valeurs
- 'return_periods' : liste des périodes de retour
- 'values_low' : liste des valeurs basses
- 'values_high' : liste des valeurs hautes
- 'name' : nom de l'échantillon
- 'code' : identifiant du lieu
- 'varname' : grandeur physique
- 'provider' : fournisseur de la donnée
- 'method' : méthode
- 'length' : taille de l'échantillon d'origine
- 'date_range' : période temporelle de l'échantillon d'origine
- 'gradex' : GradexItem
- 'coverage' : Taux de couverture de l'intervalle d'incertitude
- 'sample' : échantillon d'origine
See Also
--------
pyspc.core.samples.Sample.from_records
"""
stats = Stats(name=name, datatype=datatype)
for key, content in records.items():
stat = Stat.from_records(**content)
stats.add(stat=stat, key=key)
return stats