#!/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/>.
#
########################################################################
"""
Bibliothèque pyspc du projet pyspc - IO - PLATHYNES - read
"""
import collections
import os.path
from pyspc.convention.plathynes import DATATYPES, PREFIX_VARNAMES, RATIOS
import pyspc.core.exception as _exception
from pyspc.model.plathynes import (
Config as PLA_Config, Data as PLA_Data, Export as PLA_Export,
Results as PLA_Results, str2td)
from pyspc.model.plathynes.config import date_parser as PLA_date_parser
from pyspc.core.config import Config
from pyspc.core.location import Location, Locations
from pyspc.core.parameter import Parameter
from pyspc.core.serie import Serie
from pyspc.core.series import Series
[docs]
def read_Plathynes(filename=None, datatype=None, asloc=False, warning=True):
"""
Créer une instance Series à partir d'un fichier PLATHYNES
Parameters
----------
filename : str
Nom du fichier PLATHYNES
datatype : str
Type du fichier PLATHYNES
Other Parameters
----------------
asloc : bool
Convertir en Locations au lieu de Config.
Valable uniquement si datatype = 'project'
warning : bool
Affiche les avertissements si True
Returns
-------
series : pyspc.core.series.Series
Collection de séries de données si datatype parmi
- data
- export
- results
config : pyspc.core.config.Config
Configuration si datatype parmi
- event
- project
locations : pyspc.core.location.Locations
Lieux de mesure si datatype = project et si asloc = True
"""
# -------------------------------------------------------------------------
# 0- Contrôles
# -------------------------------------------------------------------------
_exception.check_str(filename)
_exception.raise_valueerror(
datatype not in DATATYPES,
f"Type de données prv '{datatype}' incorrect")
_exception.check_bool(warning)
provider = 'PLATHYNES'
# -------------------------------------------------------------------------
# 1.1- Conversion - DONNEES - EXPORT
# -------------------------------------------------------------------------
if datatype == 'export':
return _plathynes_export(filename, datatype, provider, warning)
# -------------------------------------------------------------------------
# 1.2- Conversion - DONNEES - RESULTS
# -------------------------------------------------------------------------
if datatype == 'results':
return _plathynes_results(filename, datatype, provider, warning)
# -------------------------------------------------------------------------
# 1.3- Conversion - DONNEES - OBSERVATION
# -------------------------------------------------------------------------
if datatype == 'data':
return _plathynes_data(filename, provider, warning)
# -------------------------------------------------------------------------
# 2.1- Conversion - CONFIGURATION - EVENT
# -------------------------------------------------------------------------
if datatype == 'event':
return _plathynes_event(filename, datatype, warning)
# -------------------------------------------------------------------------
# 2.2- Conversion - CONFIGURATION - PROJECT -> Locations
# -------------------------------------------------------------------------
if datatype == 'project' and asloc:
return _plathynes_project_asloc(filename, datatype, warning)
# -------------------------------------------------------------------------
# 2.3- Conversion - CONFIGURATION - PROJECT -> Config
# -------------------------------------------------------------------------
if datatype == 'project':
return _plathynes_project(filename, datatype, warning)
# -------------------------------------------------------------------------
# X- Conversion - CAS inconnu
# -------------------------------------------------------------------------
raise NotImplementedError
def _plathynes_data(filename, provider, warning):
"""
Conversion - DONNEES - OBSERVATION
"""
# Initialisation
series = Series(datatype='obs', name='PLATHYNES-data')
# Lecture
reader = PLA_Data(filename=filename)
df, meta = reader.read()
# Lieux
if reader.varname == 'P':
locs = {n: Location(
code=m['code'].replace('_', '-'), loctype='point',
x=m['x'], y=m['y'])
for n, m in meta['locs'].items()}
target = meta['Pas de temps']
else:
locs = {meta['loc']: Location(
code=meta['loc'].replace('_', '-'), loctype='basin',
x=meta['x'], y=meta['y'])}
target = Parameter.infer_timestep(
index=df.index, prefix=reader.varname)
# série de données
param = Parameter.find(prefix=reader.varname, timedelta=target)
df = df * RATIOS.get(reader.varname, 1)
for c in df.columns:
serie = Serie(
df[c].to_frame(), code=locs[c], varname=param.spc_varname,
provider=provider, warning=warning)
series.add(serie)
return series
def _plathynes_export(filename, datatype, provider, warning):
"""
Conversion - DONNEES - EXPORT
"""
series = Series(datatype='obs', name='PLATHYNES-export')
reader = PLA_Export(filename=filename)
dfs = reader.read()
for key, df in dfs.items():
meta = key[0].replace('_', '-')
code = key[1].replace('_', '-')
df = df.set_index(keys=['Date'], drop=True)
for c in df.columns:
prefix = PREFIX_VARNAMES.get((datatype, c), None)
if prefix is None:
continue
target = Parameter.infer_timestep(
index=df.index, prefix=prefix)
param = Parameter.find(prefix=prefix, timedelta=target)
serie = Serie(
df[c].to_frame(), code=code, varname=param.spc_varname,
provider=provider, warning=warning)
if c.endswith('sim'):
series.add(serie, meta=f"{meta}-sim")
else:
series.add(serie, meta=meta)
return series
def _plathynes_results(filename, datatype, provider, warning):
"""
Conversion - DONNEES - RESULTS
"""
series = Series(datatype='obs', name='PLATHYNES-results')
reader = PLA_Results(filename=filename)
df, meta = reader.read()
loc = Location(code=meta['loc'], name=meta['loc'], loctype='basin',
x=meta['x'], y=meta['y'])
sim = meta['evt']
df = df.set_index(keys=['Date Time'], drop=True)
df.index.name = 'Date'
for c in df.columns:
prefix = PREFIX_VARNAMES.get((datatype, c), None)
if prefix is None:
continue
target = Parameter.infer_timestep(
index=df.index, prefix=prefix)
param = Parameter.find(prefix=prefix, timedelta=target)
serie = Serie(
df[c].to_frame(), code=loc, varname=param.spc_varname,
provider=provider, warning=warning)
series.add(serie, meta=sim)
return series
def _plathynes_event(filename, datatype, warning):
"""
Conversion - CONFIGURATION - EVENT
"""
# Initialisation
config = Config()
# Lecture du fichier
reader = PLA_Config(filename=filename, datatype=datatype)
reader.read()
# Nom, Type et Description
section, info = _plathynes_event_name(reader, warning)
config.setdefault(section, collections.OrderedDict())
config[section]['datatype'] = datatype
config[section].update(info)
# Informations temporelles
config[section].update(_plathynes_event_time(reader, warning))
# Fichiers de données
config[section].update(_plathynes_event_input(reader, warning))
# Parameters
config[section].update(_plathynes_event_param(reader, warning))
return config
def _plathynes_event_input(reader, warning):
"""
Conversion - CONFIGURATION - EVENT - Fichiers de données
"""
mainkey = 'Forcing settings'
config = collections.OrderedDict()
keys = [('pluies', 'rrobs_filenames'), ('debits', 'qinj_filenames')]
for k in keys:
try:
info = reader[mainkey][k[0]]
except KeyError:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {mainkey} | { k[0]}")
info = None
else:
info = [v.replace('/', os.sep)
for v in reader[mainkey][k[0]].values()]
config[k[1]] = info
mainkey = 'Observations'
keys = [
('observations', 'qobs_filenames'),
]
for k in keys:
try:
info = reader[mainkey][k[0]]
except KeyError:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {mainkey} | { k[0]}")
info = None
else:
info = [v.replace('/', os.sep)
for v in reader[mainkey][k[0]].values()]
config[k[1]] = info
return config
def _plathynes_event_name(reader, warning):
"""
Conversion - CONFIGURATION - EVENT - Nom, Type et Description
"""
mainkey = 'event settings'
try:
name = reader[mainkey]["Nom de l'evenement"]
except KeyError:
if warning:
_exception.Warning(
__name__,
f"Elément manquant: {mainkey} | Nom de l'evenement")
name = None
section = os.path.basename(os.path.splitext(reader.filename)[0])
else:
section = name
try:
desc = reader[mainkey]["Description"]
except KeyError:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {mainkey} | Description")
desc = None
return section, collections.OrderedDict([('name', name), ('desc', desc)])
def _plathynes_event_param(reader, warning):
"""
Conversion - CONFIGURATION - EVENT - Parameters
"""
config = collections.OrderedDict()
mainkey = 'Parameters'
if mainkey not in reader:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {mainkey}")
return config
for k in reader[mainkey]:
k2 = f'{mainkey}_{k}'.lower()
config.setdefault(k2, collections.OrderedDict())
for x, y in reader[mainkey][k].items():
config[k2].setdefault(x[0], y.split(' '))
return config
def _plathynes_event_time(reader, warning):
"""
Conversion - CONFIGURATION - EVENT - Informations temporelles
"""
config = collections.OrderedDict()
mainkey = 'Temporal settings'
keys = [('Date de debut', 'first_dt'), ('Date de fin', 'last_dt')]
for k in keys:
try:
info = reader[mainkey][k[0]]
info = PLA_date_parser(info)
except KeyError:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {mainkey} | {k[0]}")
info = None
config[k[1]] = info
keys = [
('Pas de temps de forcage', 'input_ts'),
('Pas de temps de calcul', 'model_ts'),
('Pas de temps des sorties', 'output_ts'),
('Pas de temps des bilans', 'balance_ts')
]
for k in keys:
try:
info = reader[mainkey][k[0]]
if ' ' not in info:
info = '0 ' + info
if info.count(':') == 1:
info += ':00'
info = str2td(info)
except KeyError:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {mainkey} | {k[0]}")
info = None
config[k[1]] = info
return config
def _plathynes_project(filename, datatype, warning):
"""
Conversion - CONFIGURATION - PROJECT
"""
# Initialisation
config = Config(filename=filename)
section = 'project'
config.setdefault(section, collections.OrderedDict())
config[section]['datatype'] = datatype
# Lecture du fichier
reader = PLA_Config(filename=filename, datatype=datatype)
reader.read()
# Nom, Type et Description
config[section].update(_plathynes_project_name(reader, warning))
# Tables
config[section].update(_plathynes_project_tables(reader, warning))
# Charts
config[section].update(_plathynes_project_charts(reader, warning))
# Structures
config[section].update(_plathynes_project_structures(reader, warning))
# Stations RR
config[section].update(_plathynes_project_stations_RR(reader))
# Stations Q
config[section].update(_plathynes_project_stations_Q(reader))
# Configurations
config[section].update(_plathynes_project_configurations(reader))
# Events
config[section].update(_plathynes_project_events(reader))
# Simulations
config[section].update(_plathynes_project_simulations(reader, warning))
return config
def _plathynes_project_charts(reader, warning):
""""""
config = collections.OrderedDict()
mainkey = 'Charts set'
keys = [('graphs', 'figures')]
for k in keys:
try:
info = reader[mainkey][k[0]]
except KeyError:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {mainkey} | {k[0]}")
info = None
else:
info = list(reader[mainkey][k[0]].values())
config[k[1]] = info
return config
def _plathynes_project_configurations(reader):
""""""
config = collections.OrderedDict()
mainkey = 'Configurations'
for k in reader[mainkey]:
k = k.lower()
config.setdefault(k, collections.OrderedDict())
for x, y in reader[mainkey][k].items():
config[k].setdefault(x[0], y.split(' '))
return config
def _plathynes_project_events(reader):
""""""
config = collections.OrderedDict()
mainkey = 'Events'
for k in reader[mainkey]:
k = k.lower()
config.setdefault(k, collections.OrderedDict())
for x, y in reader[mainkey][k].items():
config[k].setdefault(x[0], y.split(' '))
return config
def _plathynes_project_name(reader, warning):
""""""
config = collections.OrderedDict()
mainkey = 'Settings'
keys = [('Version', 'version'), ('Nom', 'name'),
('Repertoire racine', 'root'), ('Bassin', 'basin')]
for k in keys:
try:
info = reader[mainkey][k[0]]
except KeyError:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {mainkey} | {k[0]}")
info = None
config[k[1]] = info
return config
def _plathynes_project_simulations(reader, warning):
""""""
config = collections.OrderedDict()
mainkey = 'Simulations'
keys = [
('simulations', 'simulations'),
]
for k in keys:
try:
info = reader[mainkey][k[0]]
except KeyError:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {mainkey} | {k[0]}")
info = None
else:
info = list(reader[mainkey][k[0]].values())
config[k[1]] = info
return config
def _plathynes_project_stations_Q(reader):
""""""
config = collections.OrderedDict()
mainkey = 'Stations'
subkey = 'hydro'
options = {
'Station hydro': 'stations_q',
'Debit de base': 'baseflows',
}
for k, v in reader[mainkey][subkey].items():
o = options[k[1]]
config.setdefault(o, collections.OrderedDict())
config[o].setdefault(k[0], v.split(' '))
return config
def _plathynes_project_stations_RR(reader):
""""""
config = collections.OrderedDict()
mainkey = 'Stations'
subkey = 'pluvio'
option = 'stations_rr'
config.setdefault(option, collections.OrderedDict())
for k, v in reader[mainkey][subkey].items():
config[option].setdefault(k[0], v.split(' '))
return config
def _plathynes_project_structures(reader, warning):
""""""
config = collections.OrderedDict()
mainkey = 'Structures set'
keys = [('structures', 'structures')]
for k in keys:
try:
info = reader[mainkey][k[0]]
except KeyError:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {mainkey} | {k[0]}")
info = None
else:
info = list(reader[mainkey][k[0]].values())
config[k[1]] = info
return config
def _plathynes_project_tables(reader, warning):
""""""
config = collections.OrderedDict()
mainkey = 'Tables set'
keys = [('tables', 'tables')]
for k in keys:
try:
info = reader[mainkey][k[0]]
except KeyError:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {mainkey} | {k[0]}")
info = None
else:
info = list(reader[mainkey][k[0]].values())
config[k[1]] = info
return config
def _plathynes_project_asloc(filename, datatype, warning):
"""
Conversion - CONFIGURATION - PROJECT -> Locations
"""
# Initialisation
locations = Locations(name=os.path.splitext(os.path.basename(filename))[0])
# Lecture du fichier de configuration
project = read_Plathynes(
filename=filename, datatype=datatype, warning=warning)
# Définition de la collection des lieux de mesures
keys = {'stations_rr': 'point', 'stations_q': 'basin'}
for k, k2 in keys.items():
try:
stations = project[datatype][k]
except KeyError:
if warning:
_exception.Warning(
__name__, f"Elément manquant: {datatype} | {k}")
continue
for s, info in stations.items():
loc = Location(code=info[1], name=s, loctype=k2,
x=float(info[2]), y=float(info[3]))
locations.add(loc=loc)
return locations