#!/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 <phyc2plathynes.py>."""
import os
import os.path
from pyspc.binutils.get_stations_list import get_stations_list
from pyspc.binutils.plathynes import (
events_in_project, process_locations, process_project)
from pyspc.convention.sandre import ASSOC
from pyspc.core.config import Config
import pyspc.core.exception as _exception
from pyspc.webservice.phyc import PHyC
from pyspc import read_Sandre, Series
import warnings
warnings.filterwarnings("ignore")
# -------------------------------------------------------------------
# OPTIONS FUNCTIONS
# -------------------------------------------------------------------
[docs]
def phyc2plathynes(options):
"""
Exécution des opérations du binaire <phyc2plathynes.py>.
Parameters
----------
options
Retour de pyspc.binutils.args.phyc2plathynes.phyc2plathynes
Returns
-------
filenames : list
Fichiers enregistrés
{clé=(code, datatype), valeur = liste des fichiers associés}
"""
# ==================================================================
# 1-- VERIFICATION DES OPTIONS/ARGUMENTS
# ==================================================================
# --------------------------------------------------------------
# 1.0 :
# --------------------------------------------------------------
assoc = {v: k for k, v in ASSOC.items()}
filenames = []
# --------------------------------------------------------------
# 1.1 : LISTE DES STATIONS D'INJECTION
# --------------------------------------------------------------
injections_list = get_stations_list(
station_name=options.station_name,
stations_list_file=options.stations_list_file)
# ==================================================================
# 2-- PROJET PLATHYNES
# ==================================================================
# --------------------------------------------------------------
# 2.1 : LECTURE DU FICHIER DE CONFIGURATION 'PROJET'
# --------------------------------------------------------------
prj_cfg, locations = process_project(
options.output_dir, options.projet_filename)
_exception.Information(
options.verbose,
" + Lecture du projet PLATHYNES : {}", prj_cfg.filename)
# --------------------------------------------------------------
# 2.2 : LECTURE DES INFORMATIONS SUR LES SITES / STATIONS
# --------------------------------------------------------------
stations, injections, suffix = process_locations(
locations, injections_list)
_exception.Information(
options.verbose,
" - Station pluviométrique : {}", stations['P'])
_exception.Information(
options.verbose,
" - Station hydrométrique : {}", stations['Q'])
_exception.Information(
options.verbose,
" - Station injectée : {} {}", list(injections.items()))
# --------------------------------------------------------------
# 2.3 : EVENEMENTS
# --------------------------------------------------------------
home_dirname = os.path.splitext(prj_cfg.filename)[0]
_exception.Information(
options.verbose,
" - Evenements définis dans le projet PLATHYNES")
events = events_in_project(prj_cfg, home_dirname)
_exception.Information(
options.verbose,
" + Evénement : {}", list(events.keys()))
# Gestion des événements définis par l'utilisateur
if options.events is None:
options.events = list(events.keys())
else:
options.events = list(set(events).intersection(set(options.events)))
_exception.raise_valueerror(
not options.events, "aucune événement à traiter")
# ==================================================================
# 3-- TELECHARGEMENT DES DONNEES
# ==================================================================
if options.processing_method[0] in ['dl', 'download']:
filenames = download(options, events, stations, assoc)
# ==================================================================
# 4-- CONVERSION DES DONNEES
# ==================================================================
elif options.processing_method[0] in ['cv', 'convert']:
filenames = convert(options, events, home_dirname,
stations, locations, injections, suffix)
# ==================================================================
# 5-- AUTRE
# ==================================================================
else:
_exception.Warning(
None,
f"méthode de traitement '{options.processing_method[0]}' inconnue "
"et/ou Nombre de paramètre(s) incorrect "
f"'{len(options.processing_method)}'")
# ===============================================================
# 4-- FIN DU PROGRAMME
# ===============================================================
_exception.Information(
options.verbose, " -- Fin du script phyc2plathynes")
return filenames
[docs]
def download(options, events, stations, assoc):
"""Téléchargement des données."""
dl_filenames = []
_exception.Information(
options.verbose, " + Téléchargement des données")
# --------------------------------------------------------------
# 3.1 : 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 clients 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.2 : STATIONS PLUVIOMETRIQUES ET HYDROMETRIQUES
# --------------------------------------------------------------
for evt in options.events:
fdt = events[evt]['first_dtime']
ldt = events[evt]['last_dtime']
ts = events[evt]['timestep']
_exception.Information(
options.verbose, " - Evenement sélectionné : {}", evt)
_exception.Information(
options.verbose, " + Premier instant : {}", str(fdt))
_exception.Information(
options.verbose, " + Dernier instant : {}", str(ldt))
_exception.Information(
options.verbose, " + Pas de temps : {}", str(ts))
varnames = ['P', 'Q']
datatypes = ['data_obs_meteo', 'data_obs_hydro']
for varname, datatype in zip(varnames, datatypes):
# Recherche de la grandeur selon pyspc
filenames = phyc_session.retrieve(
dirname=options.input_dir,
datatype=datatype,
codes=stations[varname],
varname=assoc[varname],
timestep=ts,
first_dtime=fdt,
last_dtime=ldt
)
# Renommage des fichiers xml
for f in filenames:
f2 = os.path.join(
os.path.dirname(f),
f"{evt}_{varname}_"
f"{os.path.basename(f).split('_')[0]}_obs.xml")
try:
os.replace(f, f2)
dl_filenames.append(f2)
except OSError:
_exception.Information(
options.verbose,
" + Ecriture du fichier sans renommage : "
"{}", f)
else:
_exception.Information(
options.verbose,
" + Ecriture du fichier avec renommage : "
"{}", f2)
# --------------------------------------------------------------
# 3.3 : FERMETURE SESSION PHyC
# --------------------------------------------------------------
_exception.Information(options.verbose, " + Déconnexion PHyC")
phyc_session.logout()
return dl_filenames
[docs]
def convert(options, events, home_dirname,
stations, locations, injections, suffix):
"""Conversion des données."""
_exception.Information(options.verbose, " + Conversion des données")
cv_filenames = []
for evt in options.events:
_exception.Information(
options.verbose, f" - Evenement sélectionné : {evt}")
files_todo = events[evt]['filenames']
files_done = []
event_dirname = os.path.join(home_dirname, f'Ev_{evt}')
# --------------------------------------------------------------
# 4.1 : STATIONS PLUVIOMETRIQUES
# --------------------------------------------------------------
varname = 'P'
datatype = 'data_obs_meteo'
if varname in options.varname:
sta_todo = stations[varname]
else:
sta_todo = []
sta_done = []
# Initialisation de la collection finale des données
series = Series(datatype='obs', name='PLATHYNES_Data')
# Boucle sur les stations à exporter
for sta in sta_todo:
# Lecture du fichier xml
xml_filename = os.path.join(
options.input_dir, f"{evt}_{varname}_{sta}_obs.xml")
if not os.path.exists(xml_filename):
continue
_exception.Information(
options.verbose,
" - Fichier XML : {}", xml_filename)
tmp_series = read_Sandre(
filename=xml_filename,
datatype=datatype,
warning=options.warning
)
# Si la collection est vide, je passe à la base suivante
if not tmp_series.codes:
continue
for key, serie in tmp_series.items():
loc = locations[key[0]]
loc.code = loc.name
serie.code = loc
series.add(
code=loc.name,
serie=serie,
meta=key[2]
)
sta_done.append(key[0])
sta_todo = list(set(sta_todo).difference(set(sta_done)))
# Export du fichier contenant l'ensemble des séries de pluie
filenames = series.to_PLATHYNES_Data(
dirname=event_dirname,
event=evt
)
_exception.Information(
options.verbose,
" + Export du fichier : {}", filenames)
files_done.extend(filenames)
# --------------------------------------------------------------
# 4.2 : STATIONS HYDROMETRIQUES
# --------------------------------------------------------------
varname = 'Q'
datatype = 'data_obs_hydro'
if varname in options.varname:
sta_todo = stations[varname]
else:
sta_todo = []
sta_done = []
# Boucle sur les stations à exporter
for sta in sta_todo:
# Lecture du fichier xml
xml_filename = os.path.join(
options.input_dir, f"{evt}_{varname}_{sta}_obs.xml")
if not os.path.exists(xml_filename):
continue
_exception.Information(
options.verbose,
" - Fichier XML : {}", xml_filename)
series = read_Sandre(
filename=xml_filename,
datatype=datatype,
warning=options.warning
)
# Si la collection est vide, je passe à la station suivante
if not series.codes:
continue
key = (series.codes[0], series.varnames[0], series.meta[0])
loc = locations[key[0]]
loc.code = loc.name # kxxx -> LaLoireChadrac
series[key].code = loc
# Ecriture des fichiers PLATHYNES
filenames = series.to_PLATHYNES_Data(
dirname=event_dirname,
injections={loc.name: injections[key[0]]},
event=evt,
suffix=suffix # series[key].location.code:
)
_exception.Information(
options.verbose,
" + Export du fichier : {}", filenames)
sta_done.append(key[0])
files_done.extend(filenames)
sta_todo = list(set(sta_todo).difference(set(sta_done)))
# --------------------------------------------------------------
# 4.3 : Contrôles des fichiers et des séries
# --------------------------------------------------------------
_exception.Information(
sta_todo,
"Station inconnue des données PHyC : {}", sta_todo)
files_todo = [f for f in files_todo
if os.path.join(home_dirname, f) not in files_done]
_exception.Information(
files_todo,
"Fichier attendu mais non créé : {}", files_todo)
cv_filenames.extend(files_done)
return cv_filenames