Code source de pyspc.model.plathynes.export

#!/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 - Exports
"""
import collections
from datetime import datetime as dt
import pandas as pnd


DATE_FORMAT = '%d/%m/%Y %H:%M:00'
"""Format des dates dans les exports PLATHYNES"""


def date_parser(txt):
    """Conversion de date"""
    return dt.strptime(txt, DATE_FORMAT)


[docs] class Export(): """ Structure de données QI des exports PLATHYNES Attributes ---------- filename : str Nom du fichier des exports PLATHYNES """
[docs] def __init__(self, filename=None): """ Initialisation de l'instance de la classe Export Parameters ---------- filename : str Nom du fichier des exports PLATHYNES """ self.filename = filename
def __str__(self): """ Afficher les méta-données de l'instance Export """ text = """ ************************************* ******* PLATHYNES - Export ********** ************************************* * NOM FICHIER = {filename} ************************************* """ return text.format(**vars(self))
[docs] def read(self): """ Lecture d'un fichier d'exports PLATHYNES Returns ------- dict de pandas.DataFrame Dictionnaire des séries d'export PLATHYNES (événement, station) : pandas.DataFrame Examples -------- >>> import pyspc.model.plathynes as _model >>> f = 'data/model/plathynes/plathynes_export_1.txt' >>> d = _model.Export(filename=f) >>> df = d.read() >>> df[('2008_11', 'LaLoireChadrac')] Date RR Qobs Qsim 0 2008-11-01 18:00:00 5.18 37.8 38.035 1 2008-11-01 19:00:00 5.60 50.7 44.906 2 2008-11-01 20:00:00 4.85 86.3 53.811 3 2008-11-01 21:00:00 5.72 139.0 64.923 4 2008-11-01 22:00:00 6.58 192.0 79.728 5 2008-11-01 23:00:00 3.97 292.0 99.110 6 2008-11-02 00:00:00 7.41 393.0 128.635 7 2008-11-02 01:00:00 8.28 492.0 167.130 8 2008-11-02 02:00:00 9.39 592.0 217.723 9 2008-11-02 03:00:00 9.08 742.0 276.985 10 2008-11-02 04:00:00 10.84 1040.0 347.986 11 2008-11-02 05:00:00 9.81 1130.0 526.110 12 2008-11-02 06:00:00 10.07 1130.0 758.124 13 2008-11-02 07:00:00 6.93 1080.0 957.992 14 2008-11-02 08:00:00 1.97 1030.0 1091.555 15 2008-11-02 09:00:00 0.36 976.0 1112.929 16 2008-11-02 10:00:00 0.13 923.0 1060.861 17 2008-11-02 11:00:00 0.15 862.0 1002.470 18 2008-11-02 12:00:00 0.00 783.0 932.845 19 2008-11-02 13:00:00 0.02 704.0 845.717 20 2008-11-02 14:00:00 0.16 627.0 752.635 21 2008-11-02 15:00:00 0.47 545.0 660.793 22 2008-11-02 16:00:00 0.39 475.0 575.557 23 2008-11-02 17:00:00 1.69 432.0 507.451 24 2008-11-02 18:00:00 8.80 414.0 459.594 """ # Initialisation cg = 0 cs = 0 frames = collections.OrderedDict() # Récupération des numéros de lignes des blocs de données with open(self.filename, 'r', encoding='utf-8', newline='\n') as f: s = None e = None for line in f.readlines(): if line.startswith('Station'): s = line.split(':')[-1].strip() cs = -1 elif line.startswith('Resultat'): s = None e = line.strip().split(' ')[-1] elif line == '\n' and s is not None and e is not None: frames.setdefault((e, s), (cg - cs, cs)) cg += 1 cs += 1 # Récupération des données sous la forme de pandas.DataFrame for k, v in frames.items(): df = pnd.read_csv( self.filename, sep=';', skiprows=v[0], nrows=v[1] - 1, converters={'Date': date_parser}, ) frames[k] = df return frames
[docs] def write(self): """ Ecrire le fichier d'exports PLATHYNES Raises ------ NotImplmentedError """ raise NotImplementedError