#!/usr/bin/python3
# -*- coding: utf-8 -*-
########################################################################
#
# This file is part of python module <pySPC>.
# Copyright (C) 2013-2020 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/>.
#
########################################################################
"""Exécution des opérations du binaire <phyc2xml.py>."""
from pyspc.binutils.get_stations_list import get_stations_list
from pyspc.convention.sandre import ASSOC
import pyspc.core.exception as _exception
from pyspc.core.config import Config
from pyspc.core.parameter import Parameter
from pyspc.core.timeutil import str2dt
from pyspc.webservice.phyc import PHyC
import warnings
warnings.filterwarnings("ignore")
# -------------------------------------------------------------------
# OPTIONS FUNCTIONS
# -------------------------------------------------------------------
[docs]
def phyc2xml(options):
"""
Exécution des opérations du binaire <phyc2xml.py>.
Parameters
----------
options
Retour de pyspc.binutils.args.phyc2xml.phyc2xml
Returns
-------
filenames : list
Fichiers enregistrés
"""
# ===============================================================
# 1-- VERIFICATION DES OPTIONS/ARGUMENTS
# ===============================================================
stations_list = get_stations_list(
station_name=options.station_name,
stations_list_file=options.stations_list_file)
_exception.raise_valueerror(
not stations_list, "aucune station à traiter"
)
options.first_dtime = str2dt(options.first_dtime)
options.last_dtime = str2dt(options.last_dtime)
print(options.user_option)
# ===============================================================
# 2-- LECTURE CONFIGURATION
# INITIALISATION DE LA SESSION PHYC
# ===============================================================
_exception.Information(
options.verbose,
" + Lecture de la configuration PHyC : {}", options.cfg_filename)
phyc_config = Config(filename=options.cfg_filename)
phyc_config.read()
if 'password' not in phyc_config['session'] or \
phyc_config['session']['password'] == '':
phyc_config['session']['password'] = input('Mot de passe absent '
'de la configuration. '
'Veuillez le renseigner :')
_exception.Information(options.verbose, " + Connexion PHyC")
_exception.Information(options.verbose, " - Création du client SOAP")
phyc_session = PHyC(
hostname=phyc_config['session'].get('hostname', None),
username=phyc_config['session']['username'],
password=phyc_config['session']['password'])
_exception.Information(
options.verbose,
f" - Connexion au serveur PHyC : {phyc_session.hostname}")
_exception.Information(
options.verbose,
f" - Authentification (identifiant) : {phyc_session.username}")
phyc_session.login()
_exception.Information(
options.verbose,
f" - Session PHyC : {phyc_session.session}")
# ===============================================================
# 3-- RECUPERATION DES FLUX XML
# ===============================================================
_exception.Information(
options.verbose,
" + Appel(s) PHyC de type {}", options.datatype)
# ---------------------------------------------------------------
# 3.1-- CONTROLES DES LIEUX
# ---------------------------------------------------------------
stations_list = check_stations_list(
stations_list=stations_list,
varname=options.varname,
datatype=options.datatype,
verbose=options.verbose
)
# ---------------------------------------------------------------
# 3.2-- DEFINITION DE LA GRANDEUR SANDRE
# ---------------------------------------------------------------
if options.varname is not None:
param = Parameter(options.varname)
timestep = param.timestep
assoc = {v: k for k, v in ASSOC.items()}
varname = assoc[options.varname[0]]
else:
varname = None
timestep = None
# ---------------------------------------------------------------
# 3.3-- TELECHARGEMENT
# ---------------------------------------------------------------
elab = options.user_option.get('elab', False)
plusvalide = options.user_option.get('plusvalide', True)
true_values = ['oui', 'OUI', 'Oui', 1, '1', 'true', 'True', 'TRUE', True]
filenames = phyc_session.retrieve(
dirname=options.output_dir,
datatype=options.datatype,
codes=stations_list,
varname=varname,
timestep=timestep,
first_dtime=options.first_dtime,
last_dtime=options.last_dtime,
elab=bool(elab in true_values),
plusvalide=bool(plusvalide in true_values),
)
_exception.Information(
options.verbose, " - Ecriture du fichier : {}", filenames)
# ===============================================================
# 4-- FIN DU PROGRAMME
# ===============================================================
_exception.Information(options.verbose, " + Déconnexion PHyC")
phyc_session.logout()
_exception.Information(
options.verbose, " -- Fin du script phyc2xml")
return filenames
[docs]
def check_stations_list(stations_list, varname, datatype, verbose):
"""Vérifier la liste des stations (cohérence avec grandeur)."""
if not datatype.startswith('data'):
return stations_list
stations_list_ok = \
[s
for s in stations_list
if (varname.startswith('Q') and len(s) in [8, 10] and # = 8
datatype.endswith('hydro')) or
(varname.startswith('H') and len(s) == 10 and
datatype.endswith('hydro')) or
(varname.startswith('P') and s.isdigit() and
datatype.endswith('meteo')) or
(varname.startswith('T') and s.isdigit() and
datatype.endswith('meteo'))]
stations_list_ko = list(set(stations_list).difference(
set(stations_list_ok)))
_exception.Information(
verbose,
"L'identifiant {} est incompatible", stations_list_ko)
return stations_list_ok