#!/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/>.
#
########################################################################
"""
Webservice - Projet Hydro2 - Export
"""
from datetime import datetime as dt, timedelta as td
from pyspc.convention.hydro2 import (
MAX_PERIOD_EXPORT,
DTFMT_EXPORT,
DATATYPES
)
import pyspc.core.exception as _exception
[docs]
class Export():
"""
Structure de données Export HYDRO-2
Attributes
----------
stations : list
Liste des entités
datatype : str
Type d'export
precision : None, int
Précision (si export TOUSMOIS)
first_dt : datetime
Premier pas de temps
last_dt : datetime
Dernier pas de temps
period : timedelta
Période extraite
dtfmt : str
Format de date
exports : None, list
Liste des exports, définis par les méthodes set_export*
"""
[docs]
def __init__(self, stations=None, datatype=None, precision=None,
first_dt=None, last_dt=None):
"""
Initialisation de l'instance de la classe Export d'Hydro-2
Parameters
----------
stations : list
Liste des entités
datatype : str
Type d'export
precision : None, int
Précision (si export TOUSMOIS)
first_dt : datetime
Premier pas de temps
last_dt : datetime
Dernier pas de temps
"""
self.datatype = datatype
self.check_dtype()
self.stations = stations
self.precision = precision
self.check_precision()
self.first_dt = first_dt
self.last_dt = last_dt
self.period = self.get_maxperiod()
self.dtfmt = self.get_dtfmt()
self.exports = None
def __str__(self):
"""
Afficher les méta-données de l'instance Export d'Hydro-2
"""
text = """
*************************************
******** HYDRO-2 - Export ***********
*************************************
* STATIONS = {stations}
* PROCEDURE = {datatype}
* DEBUT = {first_dt}
* FIN = {last_dt}
* PERIOD = {period}
* DTFMT = {dtfmt}
* PRECISION = {precision}
* EXPORTS = {exports}
*************************************
"""
return text.format(**vars(self))
[docs]
def check_dtype(self):
"""
Contrôler le type d'export
"""
try:
self.get_datatypes().index(self.datatype)
except ValueError as ve:
raise ValueError("Type d'export incorrect") from ve
[docs]
def check_precision(self):
"""
Contrôler la précision
"""
# Format des valeurs réelles (0=int, 1=.1f, ...)
test = self.precision is None or (isinstance(self.precision, int) and
0 <= self.precision <= 3)
if not test:
raise ValueError("Précision mal renseignée")
[docs]
def check_stations(self):
"""
Contrôler si les stations sont fournies sous forme de liste
"""
_exception.raise_valueerror(
not isinstance(self.stations, list),
'Les identifiants des stations ne sont pas'
'fournis sous forme de liste'
)
[docs]
def check_dtime(self):
"""
Contrôler si les dates sont fournies sous forme de datetime
"""
_exception.raise_valueerror(
not isinstance(self.first_dt, dt),
"La date de début n'est pas une instance datetime.datetime"
)
_exception.raise_valueerror(
not isinstance(self.last_dt, dt),
"La date de début n'est pas une instance datetime.datetime"
)
[docs]
def get_dtfmt(self):
"""
Contrôler et renvoyer le format des dates
"""
try:
x = DTFMT_EXPORT[self.datatype]
except KeyError as ke:
raise ValueError("Type d'export incorrect") from ke
return x
[docs]
def get_maxperiod(self):
"""
Contrôler et renvoyer la durée maximale d'un export
"""
try:
x = MAX_PERIOD_EXPORT[self.datatype]
except KeyError as ke:
raise ValueError("Type d'export incorrect") from ke
return x
[docs]
def set_export(self, onefile=None):
"""
Définir la liste des procédures d'export Hydro2
Parameters
----------
onefile : str
Nom du fichier, si l'utilisateur souhaite avoir toutes les
stations dans un seul fichier
"""
self.check_dtype()
if self.datatype in ['H-TEMPS', 'QTFIX', 'QTVAR', 'QJM']:
return self.set_export_series()
if self.datatype in ['DEBCLA']:
return self.set_export_debcla(onefile=onefile)
if self.datatype in ['CRUCAL']:
return self.set_export_crucal(onefile=onefile)
if self.datatype in ['TOUSMOIS']:
return self.set_export_tousmois(onefile=onefile)
if self.datatype in ['SYNTHESE']:
return self.set_export_synthese(onefile=onefile)
raise ValueError("Type d'export incorrect")
[docs]
def set_export_series(self):
"""
Procédure Export de séries de données
- H-TEMPS
- QTVAR
- QTFIX
- QJM
"""
# Contrôles
self.check_dtime()
self.check_stations()
# Initialisation
used_indexes = {}
self.exports = []
end = {
'QTFIX': '01', # Horaire
'QTVAR': '01', # Précision 1%
'H-TEMPS': '00', # avec valeurs mesurées et corrigées
'QJM': 'ONNO', # avec débits journaliers et validité année
}
# Exports
len_dt = int((self.last_dt - self.first_dt) / self.period) + 1
for s in self.stations:
for kt in range(len_dt):
t0 = self.first_dt + kt * self.period
tN = self.first_dt + (kt+1) * self.period
if tN > self.last_dt:
tN = self.last_dt + td(hours=1)
idx, used_indexes = self.set_index_export(
station=s,
indexes=used_indexes
)
request = [
self.datatype,
s,
self.set_shortfilename(station=s, index=idx),
t0.strftime(self.dtfmt),
tN.strftime(self.dtfmt),
end[self.datatype]
]
self.exports.append(request)
[docs]
def set_export_debcla(self, onefile=None):
"""
Procédure Export DEBCLA
Parameters
----------
onefile : str
Nom du fichier, si l'utilisateur souhaite avoir toutes les
stations dans un seul fichier
"""
# Contrôles
self.check_dtime()
self.check_stations()
# Initialisation
used_indexes = {}
self.exports = []
# Exports
if onefile is None:
for s in self.stations:
idx, used_indexes = self.set_index_export(
station=s,
indexes=used_indexes
)
request = [
self.datatype,
s,
self.set_shortfilename(station=s, index=idx),
'O', # stations antérieures
self.first_dt.strftime(self.dtfmt[0]),
self.last_dt.strftime(self.dtfmt[0]),
self.first_dt.strftime(self.dtfmt[1]),
self.last_dt.strftime(self.dtfmt[1])
]
self.exports.append(request)
else:
request = [
self.datatype,
','.join(self.stations),
'.'.join(onefile.split('.')[:-1])[:8]
+ '.' + onefile.split('.')[-1],
'O', # stations antérieures
self.first_dt.strftime(self.dtfmt[0]),
self.last_dt.strftime(self.dtfmt[0]),
self.first_dt.strftime(self.dtfmt[1]),
self.last_dt.strftime(self.dtfmt[1])
]
self.exports.append(request)
[docs]
def set_export_crucal(self, onefile=None):
"""
Procédure Export CRUCAL
Parameters
----------
onefile : str
Nom du fichier, si l'utilisateur souhaite avoir toutes les
stations dans un seul fichier
"""
# Contrôles
self.check_dtime()
self.check_stations()
# Initialisation
used_indexes = {}
self.exports = []
# Exports
if onefile is None:
for s in self.stations:
idx, used_indexes = self.set_index_export(
station=s,
indexes=used_indexes
)
request = [
self.datatype,
s,
self.set_shortfilename(station=s, index=idx),
'2',
'O', # stations antérieures
'90', # intervalle de confiance
self.first_dt.strftime(self.dtfmt[0]),
self.last_dt.strftime(self.dtfmt[0]),
self.first_dt.strftime(self.dtfmt[1]),
self.last_dt.strftime(self.dtfmt[1]),
'O',
'3'
]
self.exports.append(request)
else:
request = [
self.datatype,
','.join(self.stations),
'.'.join(onefile.split('.')[:-1])[:8]
+ '.' + onefile.split('.')[-1],
'2',
'O', # stations antérieures
'90', # intervalle de confiance
self.first_dt.strftime(self.dtfmt[0]),
self.last_dt.strftime(self.dtfmt[0]),
self.first_dt.strftime(self.dtfmt[1]),
self.last_dt.strftime(self.dtfmt[1]),
'O',
'3'
]
self.exports.append(request)
[docs]
def set_export_tousmois(self, onefile=None):
"""
Procédure Export TOUSMOIS
Parameters
----------
onefile : str
Nom du fichier, si l'utilisateur souhaite avoir toutes les
stations dans un seul fichier
"""
# Contrôles
self.check_dtime()
self.check_precision()
self.check_stations()
# Initialisation
used_indexes = {}
self.exports = []
# Exports
if isinstance(self.precision, int):
p = f'{(self.precision + 1):1d}'
else:
p = '2'
if onefile is None:
for s in self.stations:
idx, used_indexes = self.set_index_export(
station=s,
indexes=used_indexes
)
request = [
self.datatype,
s,
self.set_shortfilename(station=s, index=idx),
'2', # 1=mesuré / 2=naturel
'O', # stations antérieures
self.first_dt.strftime(self.dtfmt[0]),
self.last_dt.strftime(self.dtfmt[0]),
self.first_dt.strftime(self.dtfmt[1]),
self.last_dt.strftime(self.dtfmt[1]),
'1',
'O',
p # Format des valeurs réelles (1=int, 2=.1f, ...)
]
self.exports.append(request)
else:
request = [
self.datatype,
','.join(self.stations),
'.'.join(onefile.split('.')[:-1])[:8]
+ '.' + onefile.split('.')[-1],
'2', # 1=mesuré / 2=naturel
'O', # stations antérieures
self.first_dt.strftime(self.dtfmt[0]),
self.last_dt.strftime(self.dtfmt[0]),
self.first_dt.strftime(self.dtfmt[1]),
self.last_dt.strftime(self.dtfmt[1]),
'1',
'O',
p # Format des valeurs réelles (1=int, 2=.1f, ...)
]
self.exports.append(request)
[docs]
def set_export_synthese(self, onefile=None):
"""
Procédure Export SYNTHESE
Parameters
----------
onefile : str
Nom du fichier, si l'utilisateur souhaite avoir toutes les
stations dans un seul fichier
"""
# Contrôles
self.check_stations()
# Initialisation
used_indexes = {}
self.exports = []
# Exports
if onefile is None:
for s in self.stations:
idx, used_indexes = self.set_index_export(
station=s,
indexes=used_indexes
)
request = [
self.datatype,
s,
self.set_shortfilename(station=s, index=idx)
]
self.exports.append(request)
else:
request = [
self.datatype,
','.join(self.stations),
'.'.join(onefile.split('.')[:-1])[:8]
+ '.' + onefile.split('.')[-1]
]
self.exports.append(request)
[docs]
@staticmethod
def set_index_export(station=None, indexes=None):
"""
Définir l'indice unique servant au suffixe du fichier d'export
Parameters
----------
station : str
Code de la station
indexes : dict
Dictionnaire des indices déjà utilisés
Returns
-------
index : int
Premier indice valide pour la station courante
indexes : dict
Dictionnaire des indices déjà utilisés
"""
if indexes is None:
indexes = {}
_exception.raise_valueerror(
station is None,
'Le code de la station est inconnu'
)
_exception.raise_valueerror(
not isinstance(indexes, dict),
"<indexes> n'est pas un dictionnaire"
)
short = station[0:4]
if short not in indexes:
index = 0
indexes[short] = [index]
else:
used = sorted(indexes[short])
index = used[-1] + 1
indexes[short].append(index)
if index > 999:
raise ValueError("Indice trop élevé pour définir le "
"suffixe du fichier d'export")
return index, indexes
[docs]
def set_shortfilename(self, station=None, index=0):
"""
Définir le nom du fichier d'export Hydro2
Parameters
----------
station : str
Code de la station
index : int
Suffixe du fichier
Returns
-------
shortfilename : str
Nom court sur 8 caractères + extension
"""
_exception.raise_valueerror(
station is None,
'Le code de la station est inconnu'
)
params = {
'QJM': [4, 'j{0:03d}'],
'QTFIX': [4, 'f{0:03d}'],
'QTVAR': [4, 'v{0:03d}'],
'TOUSMOIS': [4, 'm{0:03d}'],
'H-TEMPS': [4, 'h{0:03d}'],
'DEBCLA': [5, 'DC{0:1d}'],
'CRUCAL': [5, 'QC{0:1d}'],
'SYNTHESE': [5, 'SY{0:1d}'],
}
try:
p = params[self.datatype]
except KeyError as ke:
raise ValueError('Le type de donnée est inconnu') from ke
return station[0:p[0]] + p[1].format(index) + ".txt"
[docs]
def write(self, filename=None):
"""
Ecrire le fichier contenant les procédures d'export Hydro2
Parameters
----------
filename : str
Fichier des commandes abrégées pour Hydro2, à placer dans cdesabr
Returns
-------
filename : str
Fichier des commandes abrégées pour Hydro2, à placer dans cdesabr
"""
_exception.raise_valueerror(
not self.exports,
'Exports inconnus'
)
_exception.raise_valueerror(
filename is None,
'Nom de fichier inconnu'
)
with open(filename, 'w',
encoding='utf-8', newline="\r\n") as f:
for e in self.exports:
f.write(';'.join(e))
f.write('\n')
return filename
[docs]
@classmethod
def get_datatypes(cls):
"""
Obtenir la liste des exports
Returns
-------
list
Liste des types d'export
"""
return sorted(DATATYPES['export'])