#!/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 - Online Reports - Online Reports
"""
from datetime import datetime as dt
try:
from datetime import UTC
except ImportError:
from datetime import timezone
UTC = timezone.utc
import os.path
import pyspc.core.exception as _exception
from pyspc.convention.vigicrues import (
VARNAMES as VIGICRUES_VARNAMES, CODETYPES as VIGICRUES1_CODETYPES)
from pyspc.webservice._basic import _Basic_webservice
from pyspc.webservice.report.convention import (
HOSTNAMES, REMOTENAMES, LOCALNAMES) # USERNAMES, PASSWORDS
[docs]
class OnlineReport(_Basic_webservice):
"""
Structure de données Online_Report
Attributes
----------
hostname : str
Hôte du webservice
proxies : None, dict
Dictionnaire des proxys {'protocol': 'proxy'}
timeout : None, int
Durée maximale de la requête
url : None, str
Adresse de la requête
filename : None, str
Fichier enregistré en local
reporttype : str
Type de rapport
"""
[docs]
def __init__(self, reporttype=None, proxies=None, timeout=None):
"""
Instanciation du wbeservice
Parameters
----------
reporttype : str
Type de rapport
proxies : None, dict
Dictionnaire des proxys {'protocol': 'proxy'}
timeout : None, int
Durée maximale de la requête
"""
try:
hostname = HOSTNAMES[reporttype]
except KeyError as ke:
raise ValueError(
f"Type de rapport '{reporttype}' incorrect") from ke
super().__init__(hostname=hostname, proxies=proxies, timeout=timeout)
self.reporttype = reporttype
def __str__(self):
"""
Afficher les méta-données de l'instance Report
"""
text = """
*************************************
********* WEBSERVICE - Report *******
*************************************
* TYPE DE RAPPORT = {reporttype}
* HÔTE = {hostname}
* PROXYS = {proxies}
* TIMEOUT = {timeout}
* DERNIERE URL = {url}
* DERNIER FICHIER = {filename}
*************************************
"""
return text.format(**vars(self))
[docs]
def set_filename(self, code=None, varname=None, date=None, codetype=None,
dirname='.'):
"""
Définir le fichier des documents en ligne
Parameters
----------
code : str
Identifiant (Hydro2, BNBV, station, région, département...)
varname : str
Grandeur
date : datetime
Date du bulletin
codetype : str
Type d'entité Vigicrues (1.1)
dirname : str
Répertoire local de stockage
"""
if self.reporttype in ['inrae_explore2070', 'inrae_hydroclim',
'inrae_premhyce',
'inrae_shyreg_bnbv', 'inrae_shyreg_hydro',
'mf_station', 'mf_clim', 'mf_climdata',
'vigicrues_loc']:
return self.set_filename_bycode(code=code, dirname=dirname)
if self.reporttype == 'mf_warning':
return self.set_filename_bydate(
date=date, dirname=dirname, dtfmt='%Y_%m_%d')
if self.reporttype == 'mf_dailyreport':
return self.set_filename_bydate(
date=date, dirname=dirname, dtfmt="%Y%m%d")
if self.reporttype == 'mf_monthlyreport':
return self.set_filename_mf_monthlyreport(code=code, date=date,
dirname=dirname)
if self.reporttype in ['vigicrues_fcst', 'vigicrues_obs',
'vigicrues_sandre']:
return self.set_filename_bycodevarname(code=code, varname=varname,
dirname=dirname)
if self.reporttype in ['vigicrues-1_info', 'vigicrues-1_domain',
'vigicrues-1_reach', 'vigicrues-1_loc']:
return self.set_filename_bycodedatetype(
code=code, codetype=codetype, date=date, dirname=dirname)
if self.reporttype in ['vigicrues_reach', 'vigicrues-1_geoinfo']:
return self.set_filename_bydate(
date=dt.now(UTC), dirname=dirname, dtfmt='%Y%m%d%H%M')
if self.reporttype in ['mf_station_geojson']:
return self.set_filename_noarg(dirname=dirname)
raise ValueError("Type de rapport incorrect")
[docs]
def set_filename_bycode(self, code=None, dirname='.'):
"""
Définir le fichier local
Parameters
----------
code : str
Identifiant Hydro2
dirname : str
Répertoire local de stockage
"""
_exception.check_str(code)
self.filename = os.path.join(dirname,
LOCALNAMES[self.reporttype].format(code))
[docs]
def set_filename_bycodevarname(self, code=None, varname=None, dirname='.'):
"""
Définir le fichier local
Parameters
----------
code : str
Identifiant
varname : str
Grandeur
dirname : str
Répertoire local de stockage
"""
_exception.check_str(code)
_exception.check_str(varname)
self.filename = os.path.join(
dirname,
LOCALNAMES[self.reporttype].format(code, varname))
[docs]
def set_filename_bycodedatetype(self, code=None, date=None, codetype=None,
dirname='.', dtfmt="%Y%m%d%H%M"):
"""
Définir le fichier local
Parameters
----------
code : str
Identifiant
date : datetime
Date du bulletin
codetype : str
Type d'entité Vigicrues (1.1)
dirname : str
Répertoire local de stockage
dtfmt : str
Format de la date
"""
_exception.check_str(code)
_exception.check_dt(date)
_exception.check_str(codetype)
self.filename = os.path.join(
dirname,
LOCALNAMES[self.reporttype].format(
code, codetype, date.strftime(dtfmt)))
[docs]
def set_filename_bydate(self, date=None, dirname='.', dtfmt="%Y%m%d%H%M"):
"""
Définir le fichier local
Parameters
----------
date : datetime
Date du fichier
dirname : str
Répertoire local de stockage
dtfmt : str
Format de la date
"""
_exception.check_dt(date)
_exception.check_str(dtfmt)
self.filename = os.path.join(
dirname,
LOCALNAMES[self.reporttype].format(date.strftime(dtfmt)))
[docs]
def set_filename_mf_monthlyreport(self, code=None, date=None, dirname='.'):
"""
Définir le fichier des bulletins mensuels de Météo-France
Parameters
----------
code : str
Identifiant du département/région
date : datetime
Date du bulletin
dirname : str
Répertoire local de stockage
"""
self.set_url_mf_monthlyreport(code=code, date=date)
self.filename = os.path.join(dirname, os.path.basename(self.url))
[docs]
def set_filename_noarg(self, dirname='.'):
"""
Définir le fichier sans option
Parameters
----------
dirname : str
Répertoire local de stockage
"""
self.filename = os.path.join(dirname, REMOTENAMES[self.reporttype])
[docs]
def set_verify(self):
"""
Définir le chemin local du certificat SSL
See Also
--------
https://www.ssllabs.com/ssltest/analyze.html?d=webgr.inrae.fr
"""
# voir le lien ci-dessus
# cliquer sur Certification Paths
# cliquer sur Mozilla
# cliquer sur Download Chain (icone à droite)
pemfile = os.path.join(os.path.dirname(__file__),
f'trusted_{self.reporttype}.pem')
if os.path.exists(pemfile):
self.verify = pemfile
else:
self.verify = None
[docs]
def set_url(self, code=None, date=None, varname=None, codetype=None):
"""
Définir l'url des documents en ligne
Parameters
----------
code : str
Identifiant (Hydro2, BNBV, station, région, département...)
date : datetime
Date du bulletin
varname : str
Grandeur (Vigicrues)
codetype : str
Type d'entité Vigicrues (1.1)
See Also
--------
pyspc.convention.vigicrues.CODETYPES
pyspc.convention.vigicrues.VARNAMES
"""
self.set_verify()
if self.reporttype in ['inrae_hydroclim']:
self.verify = False
if self.reporttype in ['inrae_explore2070', 'inrae_hydroclim',
'inrae_premhyce',
'inrae_shyreg_bnbv', 'inrae_shyreg_hydro',
'mf_station', 'mf_clim', 'mf_climdata',
'vigicrues_loc']:
return self.set_url_bycode(code=code)
if self.reporttype == 'mf_warning':
return self.set_url_mf_warning(date=date)
if self.reporttype == 'mf_dailyreport':
return self.set_url_mf_dailyreport(date=date)
if self.reporttype == 'mf_monthlyreport':
return self.set_url_mf_monthlyreport(code=code, date=date)
if self.reporttype in ['vigicrues_fcst', 'vigicrues_obs',
'vigicrues_sandre']:
return self.set_url_vigicrues(code=code, varname=varname)
if self.reporttype in ['vigicrues-1_info', 'vigicrues-1_domain',
'vigicrues-1_reach', 'vigicrues-1_loc']:
return self.set_url_vigicrues1(code=code, codetype=codetype)
if self.reporttype in ['vigicrues_reach', 'vigicrues-1_geoinfo',
'mf_station_geojson']:
return self.set_url_noarg()
raise ValueError("Type de rapport incorrect")
[docs]
def set_url_bycode(self, code=None):
"""
Définir une url avec l'argument code
Parameters
----------
code : str
Identifiant
"""
_exception.check_str(code)
self.url = self.hostname + REMOTENAMES[self.reporttype].format(code)
[docs]
def set_url_noarg(self):
"""
Définir une url sans option
"""
self.url = self.hostname + REMOTENAMES[self.reporttype]
[docs]
def set_url_mf_dailyreport(self, date=None):
"""
Définir l'url des bulletins journaliers de Météo-France
Parameters
----------
date : datetime
Date du bulletin
"""
_exception.check_dt(date)
_exception.raise_valueerror(
date < dt(1973, 1, 1, tzinfo=UTC),
'La date doit être postérieure au 1973-01-01'
)
self.url = f"{self.hostname}BQA/{date.strftime('%Y%m%d')}.pdf"
[docs]
def set_url_mf_monthlyreport(self, code=None, date=None):
"""
Définir l'url des bulletins mensuels de Météo-France
Parameters
----------
code : str
Identifiant du département/région
date : datetime
Date du bulletin
"""
# Contrôles
_exception.check_str(code)
_exception.check_dt(date)
b = ''
d = ''
# Bulletin mensuel FRANCE
if code == '00':
_exception.raise_valueerror(
date < dt(1999, 1, 1, tzinfo=UTC),
'La date doit être postérieure au 1999-01-01')
d = 'BCM'
b = f'{date.strftime("%Y%m")}.pdf'
# Bulletin mensuel départemental
elif dt(2001, 1, 1, tzinfo=UTC) <= date <= dt(2011, 12, 31, tzinfo=UTC):
d = 'BCMD'
b = f'{d}_{code}_{date.strftime("%Y%m")}.pdf'
# Bulletin mensuel régionnal
elif dt(2012, 1, 1, tzinfo=UTC) <= date <= dt.now(UTC):
d = 'BCMR'
b = f'{d}_{code}_{date.strftime("%Y%m")}.pdf'
else:
_exception.raise_valueerror(
date < dt(2001, 1, 1, tzinfo=UTC),
'La date doit être postérieure au 2001-01-01')
# Définition URL
self.url = f'{self.hostname}{d}/{b}'
[docs]
def set_url_mf_warning(self, date=None):
"""
Définir l'url des vigilances de Météo-France
Parameters
----------
date : datetime
Date de la vigilance
"""
# Contrôles
_exception.check_dt(date)
# Type de bulletin selon la date de production
if dt(2008, 11, 1, tzinfo=UTC) <= date <= dt(2011, 9, 30, tzinfo=UTC):
base = "vigilance1a3"
elif dt(2011, 10, 1, tzinfo=UTC) <= date <= dt.now(UTC):
base = "vigilance4"
else:
raise ValueError('La date doit être postérieure au 2008-11-01')
self.url = f"{self.hostname}?dateVigi={date.strftime('%Y-%m-%d')}"\
f"&base={base}"
[docs]
def set_url_vigicrues(self, code=None, varname=None):
"""
Définir l'url des données de Vigicrues
Parameters
----------
code : str
Identifiant Hydro3 de la station
varname : str
Grandeur parmi ('H', 'Q')
"""
_exception.check_str(code)
_exception.check_str(varname)
_exception.raise_valueerror(
varname not in VIGICRUES_VARNAMES,
f"La grandeur '{varname}' est incorrecte")
self.url = self.hostname + REMOTENAMES[self.reporttype].format(
code, varname)
[docs]
def set_url_vigicrues1(self, code=None, codetype=None):
"""
Définir l'url des données de Vigicrues - service version 1.1
Parameters
----------
code : str
Identifiant Hydro3 de la station
codetype : str
Type d'entité Vigicrues (1.1)
See Also
--------
pyspc.convention.vigicrues.CODETYPES
"""
_exception.check_str(code)
_exception.check_str(codetype)
_exception.raise_valueerror(
codetype not in VIGICRUES1_CODETYPES['vigicrues-1'].values(),
f"Le type d'entité '{codetype}' est incorrect")
self.url = self.hostname + REMOTENAMES[self.reporttype].format(
code, codetype)
[docs]
@classmethod
def get_reporttypes(cls):
"""
Liste des types de rapport
Returns
-------
list
Types de rapport
See Also
--------
pyspc.webservice.report.HOSTNAMES
"""
return sorted(list(HOSTNAMES.keys()))