#!/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 - Observations
"""
import collections
from datetime import datetime as dt
import numpy as np
import os.path
import pandas as pnd
import pyspc.core.exception as _exception
from pyspc.convention.plathynes import DATA_EXT
from .tdelta import td2str, str2td
DATE_FORMAT = '%Y-%m-%d %H:%M:00'
"""Format des dates dans les observations pour PLATHYNES"""
def date_parser(txt):
"""Conversion de date"""
return dt.strptime(txt, DATE_FORMAT)
[docs]
class Data():
"""
Structure de données des observations/injections PLATHYNES
Attributes
----------
filename : str
Nom du fichier de données
varname : str
Nom de la variable.
injection : bool
- Débit injecté si True
- Débit exutoire si False
warning : bool
Si True, alors les avertissements des méthodes sont imprimées.
"""
[docs]
def __init__(self, filename=None, warning=True):
"""
Initialisation de l'instance des données PLATHYNES
Parameters
----------
filename : str
Nom du fichier de données
warning : bool
Si True, alors les avertissements des méthodes sont imprimées.
"""
self.filename = filename
meta = self.define_file_metadata(self.filename)
self.varname = meta[0]
self.injection = meta[1]
self.warning = warning
def __str__(self):
"""
Afficher les méta-données de l'instance Data de PLATHYNES
"""
text = """
*************************************
******* PLATHYNES - Data ************
*************************************
* NOM FICHIER = {filename}
* VARIABLE PLATHYNES = {varname}
* INJECTION = {injection}
*************************************
"""
return text.format(**vars(self))
[docs]
def read(self):
"""
Lecture du fichier de données Data de PLATHYNES
Returns
-------
data : pnd.DataFrame
Dataframe des données
meta : dict
Méta-données des lieux
.. seealso::
pyspc.model.plathynes.Data.read_mgr
pyspc.model.plathynes.Data.read_mqoi
"""
if self.varname.startswith('Q'):
return self.read_mqoi()
if self.varname.startswith('P'):
return self.read_mgr()
_exception.raise_valueerror(
True,
f"La lecture de fichier de variable '{self.varname}' est "
"indisponible", warning=self.warning)
return None
[docs]
def read_mgr(self):
"""
Lecture du fichier de données Data de PLATHYNES - PLUIE
Returns
-------
data : pnd.DataFrame
Dataframe des données
meta : dict
Méta-données des lieux
Examples
--------
>>> import pyspc.model.plathynes as _model
>>> f = 'data/model/plathynes/8001_RRobs.mgr'
>>> d = _model.Data(filename=f)
>>> df, meta = d.read()
>>> df
LE_PUY_CHADRAC MAZAN_ABBAYE_RAD CAYRES
2019-11-23 00:00:00 0.0 50.0 18.0
2019-11-23 01:00:00 14.0 68.0 36.0
2019-11-23 02:00:00 20.0 74.0 42.0
2019-11-23 03:00:00 28.0 82.0 68.0
2019-11-23 04:00:00 38.0 88.0 62.0
2019-11-23 05:00:00 42.0 132.0 48.0
2019-11-23 06:00:00 52.0 107.0 74.0
>>> meta
{'Type de donnees': 'PLUVIO',
'Station': 'PROJECT_SET',
'Pas de temps': datetime.timedelta(seconds=3600),
'Facteur multiplicatif': 1.0,
'locs': {
'LE_PUY_CHADRAC': {'x': 722740.0,
'y': 2007270.0,
'code': 'LE_PUY_CHADRAC'},
'MAZAN_ABBAYE_RAD': {'x': 738420.0,
'y': 1971872.0,
'code': 'MAZAN_ABBAYE_RAD'},
'CAYRES': {'x': 716010.0,
'y': 1992118.0,
'code': 'CAYRES'},
}
}
"""
meta = {}
# Lecture des méta-données
with open(self.filename, 'r', encoding='utf-8') as f:
# Entete
f.readline()
f.readline()
f.readline()
# Type de donnees
info = f.readline().strip().split(':')
meta.setdefault(info[0], info[-1].strip())
# Station ?
info = f.readline().strip().split(':')
meta.setdefault(info[0], info[-1].strip())
# Pas de temps
info = f.readline().strip().split(': ') # Laisser espace apres :
tdelta = str2td(tstep=info[-1].strip())
meta.setdefault(info[0], tdelta)
# Facteur multiplicatif
info = f.readline().strip().split(':')
ratio = float(info[-1].strip())
meta.setdefault(info[0], ratio)
# Stations de mesure
nbsta = int(float(f.readline().strip()))
meta['locs'] = collections.OrderedDict()
for x in range(nbsta):
info = f.readline().strip().split(' ')
meta['locs'].setdefault(
info[-1],
{
'x': float(info[0]),
'y': float(info[1]),
'code': info[-1],
}
)
# Contrôle du pas de temps
# Lecture des données
with open(self.filename, 'r', encoding='utf-8') as f:
colspecs = [(0, 20)]
for x in range(nbsta):
colspecs.append((20+x*8, 20+(x+1)*8))
data = pnd.read_fwf(
f,
colspecs=colspecs,
index_col=0,
names=list(meta['locs'].keys()),
skiprows=8+nbsta,
na_values=-1,
keep_default_na=True,
parse_dates=True,
date_format=DATE_FORMAT
)
if isinstance(ratio, (float, int)):
data *= ratio
return data, meta
[docs]
def read_mqoi(self):
"""
Lecture du fichier de données Data de PLATHYNES - DEBIT
Returns
-------
data : pnd.DataFrame
Dataframe des données
meta : dict
Méta-données du lieu
Examples
--------
>>> import pyspc.model.plathynes as _model
>>> f = 'data/model/plathynes/8001_1.mqo'
>>> d = _model.Data(filename=f)
>>> df, meta = d.read()
>>> df
LaLoireChadrac
2019-11-23 00:00:00 71.2
2019-11-23 01:00:00 81.3
2019-11-23 02:00:00 90.7
2019-11-23 03:00:00 104.0
2019-11-23 04:00:00 122.0
2019-11-23 05:00:00 155.0
2019-11-23 06:00:00 207.0
2019-11-23 07:00:00 287.0
2019-11-23 08:00:00 429.0
2019-11-23 09:00:00 582.0
2019-11-23 10:00:00 706.0
2019-11-23 11:00:00 808.0
2019-11-23 12:00:00 842.0
2019-11-23 13:00:00 850.0
2019-11-23 14:00:00 888.0
2019-11-23 15:00:00 879.0
2019-11-23 16:00:00 810.0
2019-11-23 17:00:00 751.0
2019-11-23 18:00:00 675.0
>>> meta
{'loc0': 'LaLoireChadrac',
'loc': 'LaLoireChadrac',
'x': '239.0',
'y': '2008719.0',
'obj': 'Qobs',
'value': '0.000'
}
"""
meta = {}
# Lecture des méta-données
with open(self.filename, 'r', encoding='utf-8') as f:
loc0 = f.readline().split(',')[0].split(' ')[-1]
f.readline() # ligne vide
f.readline().strip() # 1
f.readline() # StationID ....
info = f.readline().strip().split(' ')
meta['loc0'] = loc0
meta['loc'] = info[0]
meta['x'] = info[1]
meta['y'] = info[2]
meta['obj'] = info[3]
meta['value'] = info[4]
# Lecture du tableau de données
with open(self.filename, 'r', encoding='utf-8') as f:
data = pnd.read_fwf(
f,
colspecs=[(0, 20), (20, 33)],
index_col=0,
names=[loc0],
skiprows=6,
na_values=-1,
keep_default_na=True,
parse_dates=True,
date_format=DATE_FORMAT
)
return data, meta
[docs]
def write(self, data=None, meta=None):
"""
Écriture du fichier de données Data de PLATHYNES
Parameters
----------
data : pnd.DataFrame
Dataframe des données
meta : dict
Méta-données des lieu
Returns
-------
str
Nom du fichier de données
.. seealso::
pyspc.model.plathynes.Data.write_mgr
pyspc.model.plathynes.Data.write_mqoi
"""
if self.varname.startswith('Q'):
return self.write_mqoi(data=data, meta=meta)
if self.varname.startswith('P'):
return self.write_mgr(data=data, meta=meta)
_exception.raise_valueerror(
True,
f"L'écriture de fichier de variable '{self.varname}' est "
"indisponible", warning=self.warning)
return None
[docs]
def write_mgr(self, data=None, meta=None):
"""
Écriture du fichier de données Data de PLATHYNES - PLUIE
Parameters
----------
data : pandas.DataFrame
Dataframe des données
meta : dict
Méta-données des lieu
Returns
-------
str
Nom du fichier de précipitations de PLATHYNES
"""
if not isinstance(data, pnd.DataFrame):
raise ValueError('Les données sont mal formatées')
if not isinstance(meta, (dict, collections.OrderedDict)):
raise ValueError('Les méta-données sont mal formatées')
if not meta:
raise ValueError('Aucune méta-donnée à écrire')
# Écriture
with open(self.filename, 'w', encoding='utf-8', newline='\r\n') as f:
f.write('########################################################'
'########################\n')
f.write('# Data settings\n')
f.write('########################################################'
'########################\n')
f.write(f"Type de donnees: {meta.get('Type de donnees', '')}\n")
f.write(f"Station: {meta.get('Station', '')}\n")
tdelta = meta.get('Pas de temps', None)
if tdelta is None:
f.write('Pas de temps: \n')
else:
f.write(f'Pas de temps: {td2str(tdelta=tdelta)}\n')
ratio = meta.get('Facteur multiplicatif', '')
if isinstance(ratio, int):
ratio = f'{ratio:2d}'
elif isinstance(ratio, float):
ratio = f'{int(ratio):2d}'
f.write(f'Facteur multiplicatif: {ratio}\n')
meta.setdefault('locs', {})
f.write(f"{len(meta['locs'])}\n")
for loc in meta['locs'].values():
f.write(f"{loc['x']:14.6f} {loc['y']:14.6f} {loc['code']}\n")
for idx in data.index:
f.write(idx.strftime('%Y-%m-%d %H:%M:00'))
for loc in meta['locs'].keys():
val = data[loc].loc[idx]
if np.isnan(val):
val = -1
f.write(f'{val:8.2f}')
f.write('\n')
return self.filename
[docs]
def write_mqoi(self, data=None, meta=None):
"""
Écriture du fichier de données Data de PLATHYNES - DEBIT
Parameters
----------
data : pandas.DataFrame
Dataframe des données
meta : dict
Méta-données du lieu
Returns
-------
str
Nom du fichier de débit de PLATHYNES
"""
if not isinstance(data, pnd.DataFrame):
raise ValueError('Les données sont mal formatées')
if not isinstance(meta, (dict, collections.OrderedDict)):
raise ValueError('Les méta-données sont mal formatées')
if not meta:
raise ValueError('Aucune méta-donnée à écrire')
# Écriture
with open(self.filename, 'w', encoding='utf-8', newline='\r\n') as f:
loc0 = meta.get('loc0', '')
f.write(f'MQO file for station {loc0}, series 1\n')
f.write('\n')
f.write(' 1\n')
f.write('StationID X Y Type\n')
f.write(f"{meta.get('loc', '')} {meta.get('x', '')} "
f"{meta.get('y', '')} {meta.get('obj', '')} "
f"{meta.get('value', '')}\n")
f.write(' Date Time Qobs [m3/s]\n')
for idx in data.index:
val = data[loc0].loc[idx]
if np.isnan(val):
val = -1
f.write(f"{idx.strftime('%Y-%m-%d %H:%M:00')} {val:12.3f}\n")
return self.filename
[docs]
def write_mhoi(self, data=None, meta=None):
"""
Écriture du fichier de données Data de PLATHYNES - HAUTEUR
Parameters
----------
data : pandas.DataFrame
Dataframe des données
meta : dict
Méta-données du lieu
Returns
-------
str
Nom du fichier de hauteur de PLATHYNES
"""
if not isinstance(data, pnd.DataFrame):
raise ValueError('Les données sont mal formatées')
if not isinstance(meta, (dict, collections.OrderedDict)):
raise ValueError('Les méta-données sont mal formatées')
if not meta:
raise ValueError('Aucune méta-donnée à écrire')
# Écriture
with open(self.filename, 'w', encoding='utf-8', newline='\r\n') as f:
loc0 = meta.get('loc0', '')
f.write(f'MHI file for station {loc0}, series 1\n')
f.write('\n')
f.write('1\n')
f.write('StationID X Y Type\n')
f.write(f"{meta.get('loc', '')} {meta.get('x', '')} "
f"{meta.get('y', '')} {meta.get('obj', '')} "
f"{meta.get('value', '')}\n")
f.write(' Date Time Hobs [m]\n')
for idx in data.index:
val = data[loc0].loc[idx]
if np.isnan(val):
val = -1
f.write(f"{idx.strftime('%Y-%m-%d %H:%M:00')} {val:12.3f}\n")
return self.filename