#!/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/>.
#
########################################################################
"""
Bibliothèque pyspc du projet pyspc - IO - Bareme - read
"""
from pyspc.core.ratingcurve import RatingCurve, RatingCurves
import pyspc.core.exception as _exception
from pyspc.data.bareme import Bareme
from pyspc.convention.bareme import RATIOS
from pyspc.core.serie import Serie
from pyspc.core.series import Series
[docs]
def read_Bareme(filename=None, datatype=None, codes=None, code_rtc=None,
first_dt=None, last_dt=None, hydro3=True, warning=False):
"""
Créer une instance Series ou RatingCurves à partir d'une base Bareme.
Parameters
----------
filename : str
Chemin de la base de données
datatype : str
Type de données à extraire.
Voir aussi pyspc.data.bareme.Bareme.get_datatypes
codes : list
Codes des lieux
first_dt : datetime
Premier instant où l'on cherche le jaugeage
last_dt : datetime
Dernier instant où l'on cherche le jaugeage
Other Parameters
----------------
code_rtc : str
Identifiant de la courbe de tarage, si nécessaire
hydro3 : bool
Convention hydro3 (True) ou Hydro2 (False).
Par défaut: True
warning : bool
Afficher les avertissements ? défaut: False
Returns
-------
series : pyspc.core.series.Series
Collection de séries de données
curves : pyspc.core.ratingcurve.RatingCurves
Collection de courbes de tarage
Examples
--------
>>> from datetime import datetime as dt
>>> from pyspc.io.sacha import read_Bareme
>>> f = 'data/io/dbase/bareme.mdb'
Cas des courbes de correction
>>> series = read_Bareme(filename=f, datatype='levelcor',
... codes=['K055001010'],
... first_dt=dt(2008, 9, 1), last_dt = dt(2009, 2, 1),
... hydro3=True
... )
>>> series
*************************************
********** SERIES *******************
*************************************
* NOM DE LA COLLECTION = Bareme
* TYPE DE COLLECTION = obs
* NOMBRE DE SERIES = 1
* ----------------------------------
* SERIE #1
* - CODE = K055001010
* - VARNAME = HI
* - META = Bareme-levelcor
*************************************
Cas des jaugeages
>>> series = read_Bareme(filename=f, datatype='flowmes',
... codes=['K055001010'],
... first_dt=dt(2008, 9, 1), last_dt = dt(2009, 2, 1),
... hydro3=True
... )
>>> series
*************************************
********** SERIES *******************
*************************************
* NOM DE LA COLLECTION = Bareme
* TYPE DE COLLECTION = obs
* NOMBRE DE SERIES = 2
* ----------------------------------
* SERIE #1
* - CODE = K055001010
* - VARNAME = HI
* - META = Bareme-flowmes
* ----------------------------------
* SERIE #2
* - CODE = K055001010
* - VARNAME = QI
* - META = Bareme-flowmes
*************************************
Cas des courbes de tarage
>>> curves = read_Bareme(filename=f, datatype='ratingcurve',
... codes=['K055001010'],
... first_dt=dt(2013, 9, 1), last_dt = dt(2015, 9, 1),
... hydro3=True
... )
>>> curves
*************************************
********* RATINGCURVES **************
*************************************
* NOM DE LA COLLECTION = Bareme
* NOMBRE DE COURBES = 4
* ----------------------------------
* COURBE #1
* - CODE = K055001010
* - NUM = H201314
* - FOURNISSEUR = Bareme
* ----------------------------------
* COURBE #2
* - CODE = K055001010
* - NUM = H201420ex
* - FOURNISSEUR = Bareme
* ----------------------------------
* COURBE #3
* - CODE = K055001010
* - NUM = H201414
* - FOURNISSEUR = Bareme
* ----------------------------------
* COURBE #4
* - CODE = K055001010
* - NUM = H201420
* - FOURNISSEUR = Bareme
*************************************
"""
# -------------------------------------------------------------------------
# 0- Contrôles
# -------------------------------------------------------------------------
_exception.check_str(filename)
_exception.check_str(datatype)
_exception.check_listlike(codes)
if first_dt is not None:
_exception.check_dt(first_dt)
if last_dt is not None:
_exception.check_dt(last_dt)
_exception.check_bool(hydro3)
_exception.check_bool(warning)
# -------------------------------------------------------------------------
# 1- Lecture
# -------------------------------------------------------------------------
# Création du lecteur
reader = Bareme(filename=filename)
# Contrôle du type de données
reader.check_datatypes(datatype)
# -------------------------------------------------------------------------
# 2- Jaugeages
# -------------------------------------------------------------------------
if datatype == 'flowmes':
return _bareme_flowmes(
reader=reader, codes=codes, first_dt=first_dt, last_dt=last_dt,
hydro3=hydro3, warning=warning)
# -------------------------------------------------------------------------
# 3- Courbes de correction
# -------------------------------------------------------------------------
if datatype == 'levelcor':
return _bareme_levelcor(
reader=reader, codes=codes, first_dt=first_dt, last_dt=last_dt,
hydro3=hydro3, warning=warning)
# -------------------------------------------------------------------------
# 3- Courbes de tarage
# -------------------------------------------------------------------------
if datatype == 'ratingcurve':
return _bareme_ratingcurve(
reader=reader, filename=filename, codes=codes,
code_rtc=code_rtc, first_dt=first_dt, last_dt=last_dt,
hydro3=hydro3, warning=warning)
return None
def _bareme_flowmes(reader=None, codes=None, first_dt=None, last_dt=None,
hydro3=True, warning=False):
"""Convertir un contenu Bareme en une collection de jaugeages."""
series = Series(datatype='obs', name='Bareme')
for code in codes:
df = reader.read_flowmes(
first_dt=first_dt, last_dt=last_dt, code=code, hydro3=hydro3,
warning=warning)
if df is None:
continue
for c in df.columns:
r = RATIOS.get(c, 1)
d = df[c].to_frame() * r
serie = Serie(d, code=code, provider='Bareme', varname=c,
warning=warning)
series.add(serie=serie, meta='Bareme-flowmes')
return series
def _bareme_levelcor(reader=None, codes=None, first_dt=None, last_dt=None,
hydro3=True, warning=False):
"""Convertir en une collection de courbes de correction."""
series = Series(datatype='obs', name='Bareme')
for code in codes:
df = reader.read_levelcor(
first_dt=first_dt, last_dt=last_dt, code=code, hydro3=hydro3,
warning=warning)
if df is None:
continue
for c in df.columns:
r = RATIOS.get(c, 1)
d = df[c].to_frame() * r
serie = Serie(d, code=code, provider='Bareme', varname=c,
warning=warning)
series.add(serie=serie, meta='Bareme-levelcor')
return series
def _bareme_ratingcurve(reader=None, filename=None, codes=None, code_rtc=None,
first_dt=None, last_dt=None, hydro3=True,
warning=False):
"""Convertir en une collection de courbes de tarage."""
curves = RatingCurves(name='Bareme')
for code in codes:
levelcor = read_Bareme(
filename=filename, datatype='levelcor', codes=[code],
hydro3=hydro3, warning=warning)
if len(levelcor) > 0:
levelcor = levelcor.concat()
levelcor.columns = [c[1][0].lower() for c in levelcor.columns]
else:
levelcor = None
flowmes = read_Bareme(
filename=filename, datatype='flowmes', codes=[code],
first_dt=first_dt, last_dt=last_dt,
hydro3=hydro3, warning=warning)
if len(flowmes) > 0:
flowmes = flowmes.concat()
flowmes.columns = [c[1][0].lower() for c in flowmes.columns]
else:
flowmes = None
d = reader.read_ratingcurve(
code=code, first_dt=first_dt, last_dt=last_dt,
code_rtc=code_rtc, hydro3=hydro3, warning=warning)
if d is None:
continue
for c in d.values():
# Conversion H : mm -> m
c['valid_interval'] = (c['valid_interval'][0] / 1000.,
c['valid_interval'][1] / 1000.)
c['hq']['h'] = c['hq']['h'] / 1000.
# Imposer la plage temporelle si sélection par code
if code_rtc is not None:
c['valid_dt'] = (first_dt, last_dt)
curve = RatingCurve(
code=c['code'], num=c['num'], provider='Bareme',
valid_dt=c['valid_dt'], valid_interval=c['valid_interval'],
update_dt=c['update_dt'], hq=c['hq'],
flowmes=flowmes, levelcor=levelcor
)
curves.add(curve)
return curves