#!/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 - Collections de Série de données
"""
import pyspc.core.exception as _exception
from pyspc.core.common import BasicDict
from pyspc.core.keyseries import str2tuple, tuple2str
from pyspc.core._series import (
Computation, # méthodes de calculs
Exporting, # méthodes d'export vers des fichiers txt, dbase...
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
Statistics, # méthodes de statistiques
TimeSerie, # méthodes de traitement temporel
)
[docs]
class Series(BasicDict, Computation, Exporting, Modeling,
Plotting, Reindexing, Scaling, Statistics, TimeSerie):
"""
Structure des Collections de Série de données.
Attributes
----------
codes : list
Liste des codes des séries
datatype : str
Type de la collection
meta : list
Liste des méta-données des séries
- Si .datatype est 'obs': meta : [('obs',), ...]
- Si .datatype est 'sim': meta : [('model',), ...]
- Si .datatype est 'fcst': meta : [('rtime', 'scen', 'prob'), ...]
name : str
Nom de la collection
varnames : list
Liste des noms de variable
See Also
--------
pyspc.core.serie.Serie
"""
[docs]
def __init__(self, datatype=None, name='series'):
"""
Initialiser l'instance de <Series>.
Attributes
----------
datatype : str
Type de la collection
name : str
Nom de la collection. Par défaut: 'series'
"""
self.check_datatype(datatype)
super().__init__(name=name, datatype=datatype)
self._codes = []
self._varnames = []
self._meta = []
def __str__(self):
"""Afficher les méta-données de l'instance Series."""
strseries = ''
if len(self.keys()) > 0:
counter = 0
for c, v, m in zip(self.codes, self.varnames, self.meta):
if isinstance(m, tuple):
m = ', '.join([str(x) for x in m])
counter += 1
strseries += '\n * ----------------------------------'
strseries += f'\n * SERIE #{counter}'
strseries += f'\n * - CODE = {c}'
strseries += f'\n * - VARNAME = {v}'
strseries += f'\n * - META = {m}'
text = """
*************************************
********** SERIES *******************
*************************************
* 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)
@property
def codes(self):
"""Liste des codes des séries."""
return self._codes
@property
def meta(self):
"""Liste des méta-données des séries."""
return self._meta
@property
def varnames(self):
"""Liste des noms de variable."""
return self._varnames
@property
def _constructor_expanddim(self):
"""Importer Series pour éviter des imports circulaires."""
from pyspc.core.serie import Serie
return Serie
[docs]
def add(self, serie=None, code=None, meta=None, overwrite=False,
strict=True, refresh=True):
"""
Ajouter une série hydrologique dans la collection.
Parameters
----------
code : str
Code de la série. Si non défini, il correspond à serie.code
meta : None, str, tuple
Autres méta-données de la série
- série d'observations : meta = None
- série de simulation : meta = model
- série de prévision : meta = (runtime, scenario, complement)
overwrite : bool
Écraser la donnée existante ? défaut: False
serie : Serie
Instance de la série
strict : bool
Contrôle strict de la cohérence de l'arguement meta
et du type de collection de séries
refresh : bool
Rafraîchir les informations de la collection. Par défaut: True
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
# Serie = self._constructor_expanddim # ()
# _exception.raise_valueerror(
# not isinstance(serie, Serie), 'Serie incorrecte'
# )
if code is None:
code = serie.code
_exception.raise_valueerror(
not isinstance(code, str), 'Code incorrect')
spc_varname = serie.spc_varname
_exception.raise_valueerror(
strict and self.datatype == 'obs'
and meta is not None and not isinstance(meta, str),
'Incohérence entre le type de collection (obs) '
f'et la série à ajouter {meta}')
_exception.raise_valueerror(
strict and self.datatype == 'fcst'
and not isinstance(meta, tuple),
'Incohérence entre le type de collection (fcst) '
'et la série à ajouter')
# ---------------------------------------------------------------------
# 1- Ajout
# ---------------------------------------------------------------------
if isinstance(meta, tuple):
while len(meta) < 4:
meta = list(meta)
meta.append(None)
meta = tuple(meta)
key = (code, spc_varname, meta)
if overwrite:
self[key] = serie
else:
self.setdefault(key, serie)
# ---------------------------------------------------------------------
# 2- Mise à jour des informations
# ---------------------------------------------------------------------
if refresh:
self.refresh()
[docs]
def asobs(self):
"""
Convertir la collection en collection d'observations.
Examples
--------
>>> series
*************************************
********** SERIES *******************
*************************************
* NOM DE LA COLLECTION = series
* TYPE DE COLLECTION = fcst
* NOMBRE DE SERIES = 2
* ----------------------------------
* SERIE #1
* - CODE = K0000000
* - VARNAME = QH
* - META = 2014-11-04 00:00:00, 1
* ----------------------------------
* SERIE #2
* - CODE = K9999999
* - VARNAME = QH
* - META = 2014-11-04 00:00:00, 1
*************************************
>>> series.asobs()
*************************************
********** SERIES *******************
*************************************
* NOM DE LA COLLECTION = series
* TYPE DE COLLECTION = obs
* NOMBRE DE SERIES = 2
* ----------------------------------
* SERIE #1
* - CODE = K0000000_2014110400_1
* - VARNAME = QH
* - META = None
* ----------------------------------
* SERIE #2
* - CODE = K9999999_2014110400_1
* - VARNAME = QH
* - META = None
*************************************
"""
obs = Series(datatype='obs', name=self.name)
for k in self:
s = self[k]
s.code = str2tuple(tuple2str(k), forceobs=True)[0]
obs.add(serie=s)
return obs
[docs]
def extend(self, series=None, overwrite=False, strict=True):
"""
Alimenter la collection à partir d'une autre collection.
Parameters
----------
series : Series
Collection d'origine
overwrite : bool
Écraser la donnée existante ? défaut: False
strict : bool
Contrôle strict de la cohérence de la collection à ajouter
et de la collection de destination
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
_exception.raise_valueerror(
not isinstance(series, type(self)),
'Collection de séries incorrecte'
)
_exception.raise_valueerror(
strict and self.datatype != series.datatype,
'Type de série incohérent'
)
# ---------------------------------------------------------------------
# 1- Ajout
# ---------------------------------------------------------------------
for key, serie in series.items():
self.add(
code=key[0],
serie=serie,
meta=key[2],
overwrite=overwrite,
strict=strict,
refresh=False
)
# ---------------------------------------------------------------------
# 2- Mise à jour des informations
# ---------------------------------------------------------------------
self.refresh()
[docs]
def check_datatype(self, datatype=None):
"""Contrôler le type de collection."""
_exception.raise_valueerror(
datatype not in self.get_types(),
'Type de collection incorrect'
)
[docs]
def check_notempty(self):
"""Contrôler si <Series> n'est pas vide."""
self.refresh()
return bool(len(set(self.codes)) != 0)
[docs]
def check_series(self):
"""Contrôler si la série est une instance <Serie>."""
Serie = self._constructor_expanddim
return all([isinstance(v, Serie) for k, v in self.items()])
[docs]
def check_unique_code(self):
"""Contrôler si <Series> correspond à un unique code."""
self.refresh_codes()
return bool(len(set(self.codes)) == 1)
[docs]
def check_unique_runtime(self):
"""Contrôler si <Series> correspond à une unique date de run."""
if self.datatype == 'obs':
return True
if self.datatype == 'fcst':
return bool(len({k[2][0] for k in self.keys()}) == 1)
return False
[docs]
def check_unique_varname(self):
"""Contrôler si <Series> correspond à un unique nom de variable."""
self.refresh_varnames()
return bool(len(set(self.varnames)) == 1)
[docs]
def refresh(self):
"""Rafraîchir les informations de la collection."""
self.refresh_codes()
self.refresh_varnames()
self.refresh_meta()
[docs]
def refresh_codes(self):
"""Rafraîchir la liste des codes."""
self._codes = [v[0] for v in self.keys()]
[docs]
def refresh_varnames(self):
"""Rafraîchir la liste des noms de variable."""
self._varnames = [v[1] for v in self.keys()]
[docs]
def replace_keys(self, assoc):
"""
Remplacer des clés de la collection.
Parameters
----------
assoc : dict
Dictionnaire de correspondance entre les anciennes (clé)
et nouvelles clés de la collection (valeur)
"""
for ok, nk in assoc.items():
ok2 = str2tuple(tuple2str(ok))
try:
s = self.pop(ok2)
s.code = str2tuple(tuple2str(nk), forceobs=True)
self.add(code=nk[0], serie=s, meta=nk[2])
except KeyError:
pass
[docs]
@classmethod
def get_types(cls):
"""Types de collections de Serie."""
return [
'obs', # observations et simulations
'fcst' # prévisions
]