#!/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/>.
#
########################################################################
"""
Evaluation de simulations et prévisions - Projet SCORES - Fichier xml
"""
import copy
from datetime import datetime as dt
import itertools
import matplotlib.pyplot as mplt
import numpy as np
import os.path
import pandas as pnd
import pylab
import warnings
from pyspc.convention.scores import SCORES_ASSOC
import pyspc.core.exception as _exception
from pyspc.core._series._plotting import DEFAULT_FIGURE
from pyspc.io.xml.from_xml import from_file, get_attrib, remove_namespace
from pyspc.plotting.colors import TABCOLORS, COLORS
TMP_SCORES_ASSOC = copy.deepcopy(SCORES_ASSOC)
[docs]
class Results():
"""
Classe permettant la manipulation du xml Scores
Attributes
----------
filename : str
Nom du fichier xml de Scores
"""
[docs]
def __init__(self, filename=None):
"""
Initialiser l'instance de la classe Results (xml) de Scores
Parameters
----------
filename : str
Nom du fichier xml de Scores
"""
self.filename = filename
self._score_error = None
self._score_table = None
def __str__(self):
"""
Afficher des méta-données de l'instance Data (xml) de Scores
"""
text = """
*************************************
********** SCORES - Results (xml) ***
*************************************
* NOM FICHIER = {filename}
*************************************
"""
return text.format(**vars(self))
[docs]
def read(self):
"""
Lecture du fichier de résultats de Scores au format XML
Les résultats sont également stockés dans
- self._score_error : résultats de scores - Groupes 1 - 6
- self._score_table : résultats de scores - Groupe 7
Returns
-------
pandas.DataFrame
Tableau des résultats de scores - Groupes 1 - 6
pandas.DataFrame
Tableau des résultats de scores - Groupe 7
"""
# ---------------------------------------------------------------------
# 0- Fichier -> ElementTree
# ---------------------------------------------------------------------
tree = from_file(filename=self.filename)
# remove all existing namespaces
nsmap = tree.getroot().nsmap
if nsmap != {}:
for alias in nsmap:
remove_namespace(tree.getroot(), nsmap[alias])
# ---------------------------------------------------------------------
# 1- Récupération du contenu
# Groupe 1-6 > Event > Leadtime > Sample > Score
# Groupe 7 > Method > Event_7 > Threshold > Score
# ---------------------------------------------------------------------
element = tree.getroot()
df, df_7 = _process_Group(element)
try:
self._score_error = df.reset_index()
except AttributeError:
pass
try:
self._score_table = df_7.reset_index()
except AttributeError:
pass
return self._score_error, self._score_table
[docs]
def to_csv(self, dirname=None):
"""
Export des résultats de Scores au format CSV
Parameters
----------
dirname : str
Répertoire local d'enregistrement des fichiers csv
Returns
-------
file_error : str
Nom du fichier csv contenant les résultats (erreurs, groupes 1-6)
None si les résultats ne sont pas accessibles
file_table : str
Nom du fichier csv contenant les résultats (table, groupe 7)
None si les résultats ne sont pas accessibles
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
if dirname is None:
dirname = '.'
_exception.check_str(dirname)
basename = os.path.basename(self.filename).replace('.xml', '')
# ---------------------------------------------------------------------
# 1- Export du tableau des erreurs
# ---------------------------------------------------------------------
if isinstance(self._score_error, pnd.DataFrame):
file_error = os.path.join(dirname, basename + '_errors.csv')
self._score_error.to_csv(
file_error,
sep=';',
float_format='%g',
header=True,
index=False,
lineterminator='\n',
date_format='%Y-%m-%d %H:%M',
)
else:
file_error = None
# ---------------------------------------------------------------------
# 2- Export de la table de contingence
# ---------------------------------------------------------------------
if isinstance(self._score_table, pnd.DataFrame):
file_table = os.path.join(dirname, basename + '_table.csv')
self._score_table.to_csv(
file_table,
sep=';',
float_format='%g',
header=True,
index=False,
lineterminator='\n',
date_format='%Y-%m-%d %H:%M',
)
else:
file_table = None
# ---------------------------------------------------------------------
# 3- Retour des noms de fichiers créés
# ---------------------------------------------------------------------
return file_error, file_table
[docs]
def to_png(self, dirname=None, config=None):
"""
Export des résultats de Scores au format PNG
Parameters
----------
dirname : str
Répertoire local d'enregistrement des fichiers png
config : dict
Configuration de la figure
Voir pyspc.core._series._plotting.DEFAULT_FIGURE
Returns
-------
files_error : list
Noms des fichiers png : 1 fichier par score (erreurs, groupes 1-6)
None si les résultats ne sont pas accessibles
file_table : str
Nom du fichier png contenant les résultats (table, groupe 7)
None si les résultats ne sont pas accessibles
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
if dirname is None:
dirname = '.'
if config is None:
config = {}
_exception.check_str(dirname)
basename = os.path.basename(self.filename).replace('.xml', '')
# ---------------------------------------------------------------------
# 1- Export du tableau des erreurs
# ---------------------------------------------------------------------
# files_error = []
if isinstance(self._score_error, pnd.DataFrame):
files_error = []
# Boucle sur les scores principaux
for s in sorted(SCORES_ASSOC.keys()):
# Définition des scores à conserver
if s.endswith('CDF'):
cs = [c for c in self._score_error if c.startswith(s)]
else:
cs = [c for c in self._score_error if c == s]
if len(cs) == 0:
continue
file_error = os.path.join(dirname, basename + f'_{s}.png')
df = self._score_error.drop(
[c for c in self._score_error.columns
if c.upper() == c and c not in cs], axis=1)
files_error.append(_create_figure(df, file_error, s, config))
else:
files_error = None
# ---------------------------------------------------------------------
# 2- Export de la table de contingence
# ---------------------------------------------------------------------
if isinstance(self._score_table, pnd.DataFrame):
file_table = _create_figure_table(
self._score_table,
os.path.join(dirname, basename + '_table.png'), config)
else:
file_table = None
# ---------------------------------------------------------------------
# 3- Retour des noms de fichiers créés
# ---------------------------------------------------------------------
return files_error, file_table
def _create_figure(df, filename, name, config):
"""
Création de la figure des scores d'erreur
Parameters
----------
df : pandas.DataFrame
Tableau de données
filename : str
Nom du fichier png
name : str
Nom du score
config : dict
Configuration de la figure
Returns
-------
filename : str
Nom du fichier png
"""
# ---------------------------------------------------------------------
# 0- Contrôles et initialisation
# ---------------------------------------------------------------------
_exception.check_dataframe(df)
_exception.check_str(filename)
_exception.check_dict(config)
config = {
**DEFAULT_FIGURE, # Configuration par défaut
**{'title': 'Evaluation SCORES-1.3.3'},
**config # Configuration de l'utilisateur
}
# ---------------------------------------------------------------------
# 1- Transformation du tableau de données
# ---------------------------------------------------------------------
df['sample'] = df.apply(_apply_sample, axis=1)
df = df.drop(['sample_inf', 'sample_sup'], axis=1)
df['event'] = df.apply(_apply_event, axis=1)
df = df.drop(['event_start', 'event_end'], axis=1)
ns = len(df['sample'].unique())
ne = len(df['event'].unique())
if ne > 1 and ns > 1:
raise NotImplementedError('Add subplots sample/event option')
# ---------------------------------------------------------------------
# 2- Création de la figure et des zones graphiques
# ---------------------------------------------------------------------
fig, axs = mplt.subplots(
nrows=1, ncols=ns, sharex=True, sharey=True,
tight_layout={'rect': [0, 0, 1, 0.95]},
figsize=(11.69, 8.27), dpi=config['dpi'])
try:
axs = axs.flatten()
except AttributeError:
axs = [axs]
# ---------------------------------------------------------------------
# 3- Création des courbes + titre par zone graphique
# ---------------------------------------------------------------------
for tag, ax in zip(df['sample'].unique(), axs):
subdf = df.loc[df['sample'] == tag]
subdf = subdf.drop('sample', axis=1)
if len(subdf.index) <= len(TABCOLORS):
iter_colors = itertools.cycle(TABCOLORS)
else:
iter_colors = itertools.cycle(COLORS)
_plot_subdf(subdf, ax, name, config, iter_colors)
ax.set_title(
tag, fontsize=max(config['xfontsize'], config['yfontsize']))
# ---------------------------------------------------------------------
# 5- Axes Y
# ---------------------------------------------------------------------
axs[0].set_ylabel(name, fontsize=config['yfontsize'])
# ---------------------------------------------------------------------
# 6- Titre
# ---------------------------------------------------------------------
fig.suptitle(config['title'], fontsize=config['tfontsize'])
# ---------------------------------------------------------------------
# 7- Légende
# ---------------------------------------------------------------------
nl = len(df['ltime'].unique())
if nl > 1:
axs[-1].legend(loc=config['legend'], fontsize=config['lfontsize'],
bbox_to_anchor=(1, 1))
# ---------------------------------------------------------------------
# 8- Enregistrement de la figure
# ---------------------------------------------------------------------
pylab.savefig(filename, dpi=config['dpi'])
mplt.close(fig)
return filename
def _create_figure_table(df, filename, config):
"""
Création de la figure des scores de la table de contingence
Parameters
----------
df : pandas.DataFrame
Tableau de données
filename : str
Nom du fichier png
config : dict
Configuration de la figure
Returns
-------
filename : str
Nom du fichier png
"""
# ---------------------------------------------------------------------
# 0- Contrôles et initialisation
# ---------------------------------------------------------------------
_exception.check_dataframe(df)
_exception.check_str(filename)
_exception.check_dict(config)
config = {
**DEFAULT_FIGURE, # Configuration par défaut
**{'title': 'Evaluation SCORES-1.3.3'},
**config # Configuration de l'utilisateur
}
# ---------------------------------------------------------------------
# 1- Transformation du tableau de données
# ---------------------------------------------------------------------
# df = df.dropna(axis=1, how='all')
df = df.drop(['event_start', 'event_end', 'method'], axis=1)
df['threshold'] = df['threshold'].astype(float)
scores = [c for c in df.columns if c.upper() == c]
nc = len(scores)
# ---------------------------------------------------------------------
# 2- Création de la figure et des zones graphiques
# ---------------------------------------------------------------------
fig, axs = mplt.subplots(
nrows=1, ncols=nc, sharex=True, sharey=True,
tight_layout={'rect': [0, 0, 1, 0.95], 'w_pad': 1.5},
figsize=(11.69, 8.27), dpi=config['dpi'])
try:
axs = axs.flatten()
except AttributeError:
axs = [axs]
# ---------------------------------------------------------------------
# 3- Création des courbes + titre par zone graphique
# ---------------------------------------------------------------------
for s, ax in zip(scores, axs):
if len(df.index) <= len(TABCOLORS):
iter_colors = itertools.cycle(TABCOLORS)
else:
iter_colors = itertools.cycle(COLORS)
_plot_subdf_table(df, ax, s, config, iter_colors)
# ---------------------------------------------------------------------
# 6- Titre
# ---------------------------------------------------------------------
fig.suptitle(config['title'], fontsize=config['tfontsize'])
# ---------------------------------------------------------------------
# 8- Enregistrement de la figure
# ---------------------------------------------------------------------
pylab.savefig(filename, dpi=config['dpi'])
mplt.close(fig)
return filename
def _plot_subdf(df, ax, name, config, iter_colors):
"""
Tracer un subplot
Parameters
----------
df : pandas.DataFrame
Tableau de données
ax : AxesSubplot
Zonge grapgique
name : str
Nom du score
config : dict
Configuration de la figure
iter_colors : iterator
Itérateur des noms de couleurs
"""
df = df.dropna(axis=1, how='all')
if 'ltime' in df.columns:
df['ltime'] = df['ltime'].astype(int)
df = df.set_index(keys='ltime', drop=True)
df = df.pivot(columns='event')
if name.endswith('CDF'):
df.columns = [f"{c[0]} ({c[1]})" for c in df.columns]
else:
df.columns = df.columns.droplevel([0])
for c in df.columns:
color = next(iter_colors)
label = c
ax.plot(df.index, df[c], color=color, label=label)
# Label de l'axe X
ax.set_xlabel("Echéance", fontsize=config['xfontsize'])
else:
# Tracer les histogrammes
for i in df.index:
color = next(iter_colors)
values = [df.loc[i, c] for c in df.columns if c.upper() == c]
nv = len(values)
if nv == 1:
ax.bar(i, values[0], color=color, align='edge', width=0.5)
else:
for k, v in enumerate(values):
ax.bar(i+k/nv, v, color=color, align='edge', width=1/nv)
# Hide major tick labels
ax.set_xticks([])
ax.set_xticklabels('')
# Customize minor tick labels
x = list(range(0, len(df.index)))
ax.set_xticks([i + 0.5 for i in x], minor=True)
ax.set_xticklabels(list(df['event'].values),
horizontalalignment='right',
minor=True)
ax.tick_params(axis='x', labelsize=config['lfontsize'], rotation=45,
which='minor', width=0)
# Label de l'axe X
# ax.set_xlabel("Evénement", fontsize=config['xfontsize'])
# ax.xaxis.grid(True, color='darkgrey', linestyle='--')
def _plot_subdf_table(df, ax, name, config, iter_colors):
"""
Tracer un subplot
Parameters
----------
df : pandas.DataFrame
Tableau de données
ax : AxesSubplot
Zonge grapgique
name : str
Nom du score
config : dict
Configuration de la figure
iter_colors : iterator
Itérateur des noms de couleurs
"""
for i in df.index:
ax.bar(i, df.loc[i, name],
color=next(iter_colors), align='edge', width=0.5)
# Hide major tick labels
ax.set_xticks([])
ax.set_xticklabels('')
# Customize minor tick labels
x = list(range(0, len(df.index)))
ax.set_xticks([i + 0.5 for i in x], minor=True)
ax.set_xticklabels(list(df['threshold'].values),
horizontalalignment='right', minor=True)
ax.tick_params(axis='x', labelsize=config['lfontsize'],
rotation=0, which='minor', width=0)
ax.set_ylabel(name, fontsize=config['yfontsize'])
def _apply_sample(row):
"""
Fusionner les informations du début/fin d'échantillon
pour obtenir le label de l'échantillon
Parameters
----------
row : pandas.Series
Ligne d'un tableau de données
Returns
-------
list
Valeurs 'event'
"""
if np.isnan(row['sample_inf']) and np.isnan(row['sample_sup']):
return 'chronique entière'
raise NotImplementedError
def _apply_event(row):
"""
Fusionner les informations du début/fin d'un événement
pour obtenir le label de l'événement
Parameters
----------
row : pandas.Series
Ligne d'un tableau de données
Returns
-------
list
Valeurs 'event'
"""
try:
e = f"{row['event_start'].strftime('%Y-%m-%d')} - "\
f"{row['event_end'].strftime('%Y-%m-%d')}"
except ValueError:
e = 'chronique entière'
return e
def find_score(attrib):
"""
Renvoyer le nom du score selon l'attribut de la balise score
"""
name = None
nature = None
value = None
if attrib.get('nature', '') == 'QUANTILE':
nature = attrib.pop('nature')
value = attrib.pop('valeur')
for n, a in TMP_SCORES_ASSOC.items():
ko = set(a.items()) ^ set(attrib.items())
if len(ko) == 0:
name = n
if name is not None and nature is not None:
name += value
return name
def _process_Scores(element):
"""
Traiter la balise score
Returns
-------
content : pandas.DataFrame
Tableau des scores
"""
content = {}
if element is not None:
for item in element.findall('score'):
name = find_score(item.attrib)
if name is None:
continue
value = float(item.text)
content.setdefault(name, [value])
return pnd.DataFrame(content)
return None
def _process_Sample(element):
"""Traiter la balise classe"""
if element is not None:
# Cas SANS classe
if element.find('classe') is None:
content = _process_Scores(element)
content['sample_inf'] = None
content['sample_sup'] = None
return content
# Cas AVEC classe
content = []
for item in element.findall('classe'):
c = _process_Scores(item)
c['sample_inf'] = get_attrib(item, 'borneInferieure')
c['sample_sup'] = get_attrib(item, 'borneSuperieure')
content.append(c)
return pnd.concat(content, ignore_index=True)
return None
def _process_Leadtime(element):
"""Traiter la balise echeance"""
if element is not None:
# Cas SIMULATION
if element.find('echeance') is None:
content = _process_Sample(element)
content['ltime'] = None
return content
# Cas PREVISION
content = []
for item in element.findall('echeance'):
c = _process_Sample(item)
c['ltime'] = get_attrib(item, 'valeur')
content.append(c)
return pnd.concat(content, ignore_index=True)
return None
def _process_Event(element):
"""Traiter la balise evenement"""
if element is not None:
content = []
for item in element.findall('evenement'):
try:
start = dt.strptime(get_attrib(item, 'dateDebut'),
'%Y-%m-%dT%H:%M:%S')
end = dt.strptime(get_attrib(item, 'dateFin'),
'%Y-%m-%dT%H:%M:%S')
except KeyError:
start = None
end = None
c = _process_Leadtime(item)
c['event_start'] = start
c['event_end'] = end
content.append(c)
# return pnd.concat(content).reset_index(drop=True)
with warnings.catch_warnings():
warnings.simplefilter(action='ignore', category=FutureWarning)
return pnd.concat(content, ignore_index=True)
return None
def _process_Group(element):
"""Traiter la balise groupe"""
if element is not None:
content = []
content_7 = []
for item in element.findall('groupe'):
n = get_attrib(item, 'numero')
if n == '7':
df = _process_Method(item)
keys = [c for c in df.columns if c.lower() == c]
df = df.set_index(keys=keys)
content_7.append(df)
else:
df = _process_Event(item)
keys = [c for c in df.columns if c.lower() == c]
df = df.set_index(keys=keys)
content.append(df)
try:
df_7 = pnd.concat(content_7, axis=1)
except ValueError:
df_7 = None
try:
df = pnd.concat(content, axis=1)
except ValueError:
df = None
return (df, df_7)
return None
def _process_Method(element):
"""Traiter la balise methode"""
if element is not None:
content = []
for item in element.findall('methode'):
n = get_attrib(item, 'nom')
if n != 'SIMPLE':
continue
c = _process_Event_group7(item)
c['method'] = n
content.append(c)
return pnd.concat(content, ignore_index=True)
return None
def _process_Event_group7(element):
"""Traiter la balise evenement du groupe 7"""
if element is not None:
content = []
for item in element.findall('evenement'):
try:
start = get_attrib(item, 'dateDebut')
end = get_attrib(item, 'dateFin')
except KeyError:
start = None
end = None
c = _process_Threshold(item)
c['event_start'] = start
c['event_end'] = end
content.append(c)
return pnd.concat(content, ignore_index=True)
return None
def _process_Threshold(element):
"""Traiter la balise seuil"""
if element is not None:
content = []
for item in element.findall('seuil'):
c = _process_Scores(item)
c['threshold'] = get_attrib(item, 'valeur')
content.append(c)
return pnd.concat(content, ignore_index=True)
return None