#!/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/>.
#
########################################################################
"""
Méta-données (statistiques, synthèse) - Bareme - Données
"""
from datetime import datetime as dt, timedelta as td
import numpy as np
import pandas as pnd
import pyspc.core.exception as _exception
from pyspc.io.dbase.mdb import Mdb
from pyspc.convention.bareme import ATTS, DATATYPES, SQL, TABLES
[docs]
class Bareme(Mdb):
"""
Classe destinée à traiter les bases Bareme.
Attributes
----------
filename : str
Chemin de la base de données
sql : str
Requête courante au format SQL
"""
[docs]
def __init__(self, filename=None):
"""
Initialiser l'instance Bareme.
Parameters
----------
filename : str
Chemin de la base de données
"""
super().__init__(filename=filename)
[docs]
def check_datatypes(self, datatype):
"""
Contrôle du type de données Bareme.
Parameters
----------
dataype : str
Type de données Bareme
Raises
------
ValueError
Si le type de données Bareme est incorrect
See Also
--------
Bareme.get_datatypes
"""
_exception.raise_valueerror(
datatype not in self.get_datatypes()
)
[docs]
def read_flowmes(self, code=None, first_dt=None, last_dt=None,
hydro3=True, warning=False):
"""
Lecture des jaugeages Bareme.
Parameters
----------
code : str
Code du lieu
first_dt : datetime
Premier instant où l'on cherche le jaugeage
last_dt : datetime
Dernier instant où l'on cherche le jaugeage
hydro3 : bool
Convention hydro3 (True) ou Hydro2 (False).
Par défaut: True
warning : bool
Afficher les avertissements ? défaut: False
Returns
-------
df : pandas.DataFrame
Tableau des jaugeages (hauteur en mm et débit en m3/s)
Examples
--------
>>> from datetime import datetime as dt
>>> from pyspc.data.bareme import Bareme
>>> f = 'data/io/dbase/bareme.mdb'
>>> reader = Bareme(filename=f)
>>> df = reader.read_flowmes(code='K055001010',
... first_dt=dt(2008, 9, 1), last_dt=dt(2009, 2, 1),
... hydro3=True
... )
>>> df
jcote jdebit
jdatedeb
2008-09-02 07:00:00 -1190 9.270000
2008-09-25 13:15:00 -1250 6.660000
2008-10-02 07:00:00 -1270 6.260000
2008-11-02 00:00:00 5050 252.000000
2008-11-02 11:07:00 5070 2790.000000
2008-11-02 11:08:00 5070 2850.000000
2008-11-02 11:09:00 4660 2790.000000
2008-11-02 13:15:00 4360 2500.000000
2009-01-29 14:45:00 -785 42.599998
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
_exception.check_str(code)
_exception.check_bool(hydro3)
_exception.check_bool(warning)
# ---------------------------------------------------------------------
# 1- Définir la requête SQL
# ---------------------------------------------------------------------
atts = ATTS['flowmes']
t = TABLES['flowmes']
sql_atts = ",".join([f'"{t}"."{a}"' for a in atts])
self.sql = SQL['flowmes'].format(
TABLES['station'], sql_atts, t, ATTS[('station', hydro3)], code)
if isinstance(first_dt, dt):
x = self.from_datetime(first_dt, "%Y%m%d%H", -60)
self.sql += f" AND ({t}.{atts[0]}>={x})"
if isinstance(last_dt, dt):
x = self.from_datetime(last_dt, "%Y%m%d%H", +60)
self.sql += f" AND ({t}.{atts[0]}<={x})"
# ---------------------------------------------------------------------
# 2- Appliquer la requête SQL
# ---------------------------------------------------------------------
self.connect()
content = self.execute(warning=warning)
self.close()
# ---------------------------------------------------------------------
# 3- Traitement du résultat de la requête SQL
# ---------------------------------------------------------------------
if self.check_sql_return(content=content, warning=warning) is None:
return None
# Créer le pnd.DataFrame
df = {c: [x[k] for x in content]
for k, c in enumerate(atts)}
df = pnd.DataFrame(df)
# Définir l'index par la colonnes DateVal
df = df.set_index(keys=atts[0], drop=True)
df = df.sort_index()
# ---------------------------------------------------------------------
# 7- Retour
# ---------------------------------------------------------------------
return df
[docs]
def read_levelcor(self, code=None, first_dt=None, last_dt=None,
hydro3=True, warning=False):
"""
Lecture des courbes de correction Bareme.
Parameters
----------
code : str
Code du lieu
first_dt : datetime
Premier instant où l'on cherche le jaugeage
last_dt : datetime
Dernier instant où l'on cherche le jaugeage
hydro3 : bool
Convention hydro3 (True) ou Hydro2 (False).
Par défaut: True
warning : bool
Afficher les avertissements ? défaut: False
Returns
-------
df : pandas.DataFrame
Tableau des courbes de correction (hauteur en mm)
Examples
--------
>>> from datetime import datetime as dt
>>> from pyspc.data.bareme import Bareme
>>> f = 'data/io/dbase/bareme.mdb'
>>> reader = Bareme(filename=f)
>>> df = reader.read_levelcor(code='K055001010',
... first_dt=dt(2014, 1, 1), last_dt=dt(2015, 1, 1),
... hydro3=True
... )
>>> df
valeur
ladate
2014-04-23 13:40:00 0
2014-05-22 08:20:00 -30
2014-06-05 08:35:00 -30
2014-06-18 06:20:00 -30
2014-07-01 09:50:00 -30
2014-07-06 20:00:00 -30
2014-07-09 00:00:00 0
2014-08-04 12:00:00 0
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
_exception.check_str(code)
_exception.check_bool(hydro3)
_exception.check_bool(warning)
# ---------------------------------------------------------------------
# 1- Définir la requête SQL
# ---------------------------------------------------------------------
atts = ATTS['levelcor']
t = TABLES['levelcor']
sql_atts = ",".join([f'"{t}"."{a}"' for a in atts])
self.sql = SQL['levelcor'].format(
TABLES['station'], sql_atts, t, ATTS[('station', hydro3)], code)
if isinstance(first_dt, dt):
x = self.from_datetime(first_dt, "%Y%m%d%H", -60)
self.sql += f" AND ({t}.{atts[0]}>={x})"
if isinstance(last_dt, dt):
x = self.from_datetime(last_dt, "%Y%m%d%H", +60)
self.sql += f" AND ({t}.{atts[0]}<={x})"
# ---------------------------------------------------------------------
# 2- Appliquer la requête SQL
# ---------------------------------------------------------------------
self.connect()
content = self.execute(warning=warning)
self.close()
# ---------------------------------------------------------------------
# 3- Traitement du résultat de la requête SQL
# ---------------------------------------------------------------------
if self.check_sql_return(content=content, warning=warning) is None:
return None
# Créer le pnd.DataFrame
df = {c: [x[k] for x in content]
for k, c in enumerate(atts)}
df = pnd.DataFrame(df)
# Définir l'index par la colonnes DateVal
df = df.set_index(keys=atts[0], drop=True)
df = df.sort_index()
# ---------------------------------------------------------------------
# 7- Retour
# ---------------------------------------------------------------------
return df
[docs]
def read_ratingcurve(self, code=None, first_dt=None, last_dt=None,
code_rtc=None, hydro3=True, warning=False):
"""
Lire les jaugeages Bareme.
Parameters
----------
code : str
Code du lieu
first_dt : datetime
Premier instant où l'on cherche le jaugeage
last_dt : datetime
Dernier instant où l'on cherche le jaugeage
code_rtc : str
Identifiant de la courbe de tarage
hydro3 : bool
Convention hydro3 (True) ou Hydro2 (False).
Par défaut: True
warning : bool
Afficher les avertissements ? défaut: False
Returns
-------
data : dict
Jaugeages
Notes
-----
Si code_ct est défini, alors seulement cette courbe de tarage
sera extraite de la base Bareme.
Sinon, les contraintes temporelles sont appliquées
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
_exception.check_str(code)
_exception.check_bool(hydro3)
_exception.check_bool(warning)
# ---------------------------------------------------------------------
# 1- Définir la requête SQL
# ---------------------------------------------------------------------
atts = ATTS['ratingcurve']
t = TABLES['ratingcurve']
sql_atts = ",".join([f'"{t}"."{a}"' for a in atts])
self.sql = SQL['ratingcurve'].format(
TABLES['station'], sql_atts, t, ATTS[('station', hydro3)], code)
if isinstance(code_rtc, str):
self.sql += f" AND ({t}.{atts[5]}='{code_rtc}')"
else:
if isinstance(first_dt, dt):
x = self.from_datetime(first_dt, "%Y%m%d%H", -60)
self.sql += f" AND ({t}.{atts[1]}>={x})"
if isinstance(last_dt, dt):
x = self.from_datetime(last_dt, "%Y%m%d%H", +60)
self.sql += f" AND ({t}.{atts[0]}<={x})"
# ---------------------------------------------------------------------
# 2- Appliquer la requête SQL
# ---------------------------------------------------------------------
self.connect()
content = self.execute(warning=warning)
self.close()
# ---------------------------------------------------------------------
# 3- Traitement du résultat de la requête SQL
# ---------------------------------------------------------------------
if self.check_sql_return(content=content, warning=warning) is None:
return None
data = {}
for c in content:
noct = c[3]
ctype = c[4]
values = None
if ctype == 'C':
values = self.read_rtc_points(noct=noct, warning=warning)
elif ctype == 'T':
values = self.read_rtc_power(noct=noct, warning=warning)
data[noct] = {
'code': code,
'num': c[5],
'valid_dt': (c[0], c[1] - td(seconds=1)), # éviter overlapping
'valid_interval': (c[6], c[7]),
'update_dt': c[2],
'levelcor': None,
'hq': values}
# ---------------------------------------------------------------------
# 4- Retour
# ---------------------------------------------------------------------
return data
[docs]
def read_rtc_points(self, noct=None, warning=False):
"""
Récupérer les "points" de la courbe de tarage.
Parameters
----------
noct : int
Identifiant de la courbe de tarage
warning : bool
Afficher les avertissements ? défaut: False
Returns
-------
df : pandas.DataFrame
Courbe de tarage (hauteur en mm et débit en m3/s)
Examples
--------
>>> from pyspc.data.bareme import Bareme
>>> f = 'data/io/dbase/bareme.mdb'
>>> reader = Bareme(filename=f)
>>> df = reader.read_rtc_points(noct=200758)
>>> df
h q
nopt
1 1040 0.414000
2 1070 0.634000
3 1110 0.955000
4 1160 1.390000
5 1170 1.490000
6 1170 1.510000
7 1210 2.120000
8 1250 2.710000
9 1360 4.970000
10 1420 6.890000
11 1500 10.000000
12 1500 9.740000
13 1530 12.300000
14 1550 14.000000
15 1580 16.600000
16 1630 21.299999
17 1690 27.500000
18 1780 37.799999
19 1900 53.400002
20 2020 70.800003
21 2180 96.699997
22 2220 104.000000
23 2240 107.000000
24 2270 113.000000
25 2290 116.000000
26 2330 124.000000
27 2350 127.000000
28 2580 173.000000
29 2830 229.000000
30 3140 306.000000
31 3350 362.000000
32 3530 417.000000
33 4050 592.000000
34 4740 849.000000
35 5120 1000.000000
36 5140 1010.000000
37 5190 1030.000000
38 5210 1040.000000
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
_exception.check_int(noct)
_exception.check_bool(warning)
# ---------------------------------------------------------------------
# 1- Définir la requête SQL
# ---------------------------------------------------------------------
atts = ATTS['rtc_point']
t = TABLES['rtc_point']
sql_atts = ",".join([f'"{t}"."{a}"' for a in atts])
self.sql = SQL['rtc_point'].format(t, sql_atts, noct)
# ---------------------------------------------------------------------
# 2- Appliquer la requête SQL MODELE
# ---------------------------------------------------------------------
self.connect()
content = self.execute(warning=warning)
self.close()
# ---------------------------------------------------------------------
# 3- Traitement du résultat de la requête SQL
# ---------------------------------------------------------------------
if self.check_sql_return(content=content, warning=warning) is None:
return None
# Créer le pnd.DataFrame
df = {c: [x[k] for x in content]
for k, c in enumerate(atts)}
df = pnd.DataFrame(df)
# Définir l'index par la colonnes DateVal
df = df.set_index(keys=atts[0], drop=True)
# ---------------------------------------------------------------------
# 7- Retour
# ---------------------------------------------------------------------
return df
[docs]
def read_rtc_power(self, noct=None, delta=10, warning=False):
"""
Récupérer les "tronçons" de la courbe de tarage.
Ces tronçons sont ensuite converties en points de coordonnées
(hauteur, débit)
Parameters
----------
noct : int
Identifiant de la courbe de tarage
delta : int
Incrément en hauteur (mm) entre deux points convertis
Par défaut: 10 mm
warning : bool
Afficher les avertissements ? défaut: False
Returns
-------
df : pandas.DataFrame
Courbe de tarage (hauteur en mm et débit en m3/s)
Examples
--------
>>> from pyspc.data.bareme import Bareme
>>> f = 'data/io/dbase/bareme.mdb'
>>> reader = Bareme(filename=f)
>>> df = reader.read_rtc_power(noct=200664, delta=100)
>>> df
h q
nopt
1 -118.0 0.000000
2 -18.0 0.173092
3 82.0 0.488561
4 128.0 0.665843
5 228.0 1.156614
6 255.0 1.309881
7 355.0 2.966919
8 455.0 5.086275
9 555.0 7.608920
10 655.0 10.496204
11 755.0 13.720089
12 855.0 17.258944
13 955.0 21.095380
14 1055.0 25.214986
15 1155.0 29.605556
16 1255.0 34.256559
17 1355.0 39.158782
18 1455.0 44.304071
19 1555.0 49.685135
20 1595.0 51.904956
21 1695.0 58.675252
22 1795.0 65.752225
23 1895.0 73.127264
24 1995.0 80.792551
25 2095.0 88.740943
26 2195.0 96.965875
27 2295.0 105.461290
28 2395.0 114.221569
29 2495.0 123.241481
30 2595.0 132.516143
... ... ...
45 4095.0 299.616612
46 4195.0 312.484225
47 4295.0 325.553750
48 4395.0 338.823129
49 4495.0 352.290379
50 4595.0 365.953582
51 4695.0 379.810884
52 4795.0 393.860492
53 4895.0 408.100668
54 4995.0 422.529730
55 5095.0 437.146045
56 5195.0 451.948029
57 5295.0 466.934145
58 5395.0 482.102897
59 5495.0 497.452834
60 5595.0 512.982542
61 5695.0 528.690646
62 5795.0 544.575807
63 5895.0 560.636721
64 5995.0 576.872116
65 6095.0 593.280753
66 6195.0 609.861422
67 6295.0 626.612943
68 6395.0 643.534163
69 6495.0 660.623956
70 6595.0 677.881224
71 6695.0 695.304889
72 6795.0 712.893901
73 6895.0 730.647232
74 6995.0 748.563875
[74 rows x 2 columns]
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
_exception.check_int(noct)
_exception.check_bool(warning)
# ---------------------------------------------------------------------
# 1- Définir la requête SQL
# ---------------------------------------------------------------------
atts = ATTS['rtc_power']
t = TABLES['rtc_power']
sql_atts = ",".join([f'"{t}"."{a}"' for a in atts])
self.sql = SQL['rtc_power'].format(t, sql_atts, noct)
# ---------------------------------------------------------------------
# 2- Appliquer la requête SQL MODELE
# ---------------------------------------------------------------------
self.connect()
content = self.execute(warning=warning)
self.close()
# ---------------------------------------------------------------------
# 3- Traitement du résultat de la requête SQL
# ---------------------------------------------------------------------
if self.check_sql_return(content=content, warning=warning) is None:
return None
# Créer le pnd.DataFrame
df = {c: [x[k] for x in content]
for k, c in enumerate(atts)}
df = pnd.DataFrame(df)
# Définir l'index par la colonnes DateVal
df = df.set_index(keys=atts[0], drop=True)
# ---------------------------------------------------------------------
# 4- Conversion des tronçons en points
# ---------------------------------------------------------------------
hinf = list(df['hsup'].values)
hinf = hinf[:-1]
hinf.insert(0, np.nan)
df['hinf'] = hinf
# Suppression de la 1e ligne inutile
df = df.drop([1])
df = _pow2pts(df, delta)
# ---------------------------------------------------------------------
# 7- Retour
# ---------------------------------------------------------------------
return df
[docs]
@classmethod
def get_datatypes(cls):
"""
Types de données Bareme.
Returns
-------
list
Types de données Bareme
"""
return sorted(DATATYPES)
def _pow2pts(df, delta):
"""
Convertir une courbe tronçon en courbe points.
Parameters
----------
df : pandas.DataFrame
Courbe de tarage par tronçon
delta : int
Incrément en hauteur (mm) entre deux points convertis
Returns
-------
df : pandas.DataFrame
Courbe de tarage par points
"""
data = []
for a, b, h0, hsup, hinf in zip(df['a'], df['b'], df['h0'],
df['hsup'], df['hinf']):
h = hinf
while h < hsup:
try:
q = a * (h - h0) ** b
except ValueError:
pass
else:
data.append((h, q))
finally:
h += delta
df = pnd.DataFrame(data)
df.index.name = 'nopt'
df.index = df.index + 1
df.columns = ['h', 'q']
return df