#!/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/>.
#
########################################################################
"""
Modélisations hydrologiques - Projet PLATHYNES - Configuration
"""
import collections
from datetime import datetime as dt
import os.path
from pyspc.core.config import Config as _Config
from pyspc.convention.plathynes import CONFIG_DTYPES, NUMBER_OF
DATE_FORMAT = '%Y-%m-%d %H:%M:00'
"""Format des dates dans les résultats PLATHYNES"""
def date_parser(txt):
""""Convertisseur de date."""
return dt.strptime(txt, DATE_FORMAT)
[docs]
class Config(collections.OrderedDict):
"""
Classe destinée à traiter la configuration de PLATHYNES.
Attributes
----------
filename : str
Nom du fichier de configuration
datatype : str
Type de fichier de configuration
"""
# Ré-utilisations des méthodes de la classe native Config
update_config = _Config.update_config
list_sections_options = _Config.list_sections_options
[docs]
def __init__(self, filename=None, datatype=None):
"""
Initialisation de l'instance Config de PLATHYNES.
Parameters
----------
filename : str
Nom du fichier de configuration
datatype : str
Type de fichier de configuration
"""
super().__init__()
self.filename = filename
self._check_datatype(
datatype=datatype
)
self._check_ext(
datatype=datatype,
ext=os.path.splitext(self.filename)[-1]
)
self.datatype = datatype
def __str__(self):
"""Afficher les méta-données de l'instance Config de PLATHYNES."""
if self:
strdata = ''
for s in self.keys():
for o, v in self[s].items():
if isinstance(v, str):
strdata += f'\n * + {s} | {o} = {v}'
elif isinstance(v, dict):
strdata += f'\n * + {s} | {o}'
for o2, v2 in v.items():
strdata += \
f'\n * - {o2[0]} | {o2[1]} = {v2}'
else:
strdata = ''
text = """
*************************************
******** PLATHYNES - Config *********
*************************************
* NOM FICHIER = {filename}
* TYPE FICHIER = {datatype}
* INFORMATIONS
* ------------------ {strdata}
*************************************
"""
return text.format(
filename=self.filename,
datatype=self.datatype,
strdata=strdata
)
def _check_datatype(self, datatype=None):
"""
Contrôler le type de fichier de configuration.
Parameters
----------
datatype : str
Type de fichier de configuration
Raises
------
ValueError
Si le type de fichier n'est pas reconnu
"""
if datatype not in self.get_types():
raise ValueError("Type de fichier de configuration incorrect")
def _check_ext(self, datatype=None, ext=None):
"""
Contrôler la cohérence entre le type et l'extension du fichier.
Parameters
----------
datatype : str
Type de fichier de configuration
ext : str
Extension du fichier de configuration
Raises
------
ValueError
Si une incohérence est constatée
"""
if ext != self.get_ext(datatype=datatype):
raise ValueError("Incohérence entre l'extension du fichier "
f"'{ext}' et le type de configuration "
f"'{datatype}'")
[docs]
def read(self):
"""Lecture du fichier de configuration."""
self.clear()
section = None
option = None
multiple = False
with open(self.filename, 'r', encoding='utf-8') as f:
for x in f.readlines():
# Cadre du commentaire
if x.startswith('#='):
multiple = False
continue
# Ligne vide
if not bool(x.strip()):
section = None
option = None
multiple = False
continue
# Libellé du commentaire
if x.startswith('# '):
section = x.replace('#', '').strip()
option = None
multiple = False
self.setdefault(section, collections.OrderedDict())
continue
# Ce n'est pas un multi-ligne
# contrairement à ce que le contenu laisse penser
if section == 'Optimisation settings':
value = ':'.join(x.split(':')[1:]).strip()
option = x.split(':')[0].strip()
self.setdefault(section, collections.OrderedDict())
self[section].setdefault(option, value)
multiple = False
# Début multi-valeurs (multi-lignes)
elif x.startswith('Nombre de') or x.startswith('Number of'):
# Cas spécifiques
if section in ['Unites modele', 'Regression functions',
'Model functions']:
option = ' '.join(
x.split(':')[0].strip().replace("'", " ")
.split(" ")[-2:])
# Cas général
else:
option = x.split(':')[0]\
.strip().replace("'", " ").split(" ")[-1]
self[section].setdefault(option, collections.OrderedDict())
multiple = True
continue
# Cas multi-lignes
if multiple:
if section is None or option is None:
continue
sublbl = x.split(':')[0].strip()
subval = ':'.join(x.split(':')[1:]).strip()
# subkey = subval.split(' ', maxsplit=1)[0]
# subkey = subval.rsplit(' ', maxsplit=1)[-1]
if self.datatype == 'event' \
or section in ['Regression functions']:
subkey = subval.rsplit(' ', maxsplit=1)[-1]
else:
subkey = subval.split(' ', maxsplit=1)[0]
self[section][option].setdefault(
(subkey, sublbl),
subval
)
# Autres lignes
else:
if section is None:
continue
value = ':'.join(x.split(':')[1:]).strip()
if section in ['Model functions']:
sublbl = x.split(':')[0].strip()
subkey = value.split(' ', maxsplit=1)[0]
option = (subkey, sublbl)
else:
option = x.split(':')[0].strip()
self.setdefault(section, collections.OrderedDict())
self[section].setdefault(option, value)
[docs]
def write(self):
"""
Écriture du fichier de configuration.
"""
self._check_ext(
datatype=self.datatype,
ext=os.path.splitext(self.filename)[-1]
)
with open(self.filename, 'w', encoding='utf-8') as f:
for section in self:
f.write('#==================================================='
'============================\n')
f.write(f'# {section}\n')
f.write('#==================================================='
'============================\n')
for option, value in self[section].items():
if isinstance(value, str) and isinstance(option, tuple):
f.write(f'{option[1]}: {value}\n')
elif isinstance(value, str):
f.write(f'{option}: {value}\n')
elif isinstance(value, dict):
txt = NUMBER_OF[self.datatype][(section, option)]
counter = len({k[0] for k in value.keys()})
f.write(f'{txt}{option}: {counter}\n')
for k, v in value.items():
f.write(f'{k[1]}: {v}\n')
f.write('\n')
[docs]
@staticmethod
def get_ext(datatype=None):
"""
Liste des extensions des fichiers de configuration PLATHYNES.
Parameters
----------
datatype : str
Type de configuration PLATHYNES
Returns
-------
list
Extensions des fichiers de configuration PLATHYNES
.. seealso:: pyspc.model.plathynes.convention.CONFIG_DTYPES
"""
try:
ext = CONFIG_DTYPES[datatype]
except KeyError as ke:
raise ValueError('Extension inconnue') from ke
return ext
[docs]
@staticmethod
def get_types():
"""
Liste des types de configuration PLATHYNES.
Returns
-------
list
Types de configuration PLATHYNES
.. seealso:: pyspc.model.plathynes.convention.CONFIG_DTYPES
"""
return sorted(list(CONFIG_DTYPES.keys()))