#!/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/>.
#
########################################################################
"""
Données d'observation et de prévision - Projet LAMEDO - BdApBp
"""
from datetime import datetime as dt
import json
import numpy as np
import pandas as pnd
import pyspc.core.exception as _exception
from pyspc.convention.lamedo import (
BDAPBP_DATATYPES, BDAPBP_COLUMNS, BDAPBP_DATE_FORMAT)
[docs]
class BdApbp():
"""
Structure de données BdApbp.
Attributes
----------
filename : str
Fichier Json BdApbp
"""
[docs]
def __init__(self, filename=None):
"""
Instanciation du wbeservice.
Parameters
----------
filename : str
Fichier Json BdApbp
"""
self.filename = filename
self._datatype = None
def __str__(self):
"""Afficher les méta-données de l'instance BdApBp."""
text = """
*************************************
********* LAMEDO - BdApBp ***********
*************************************
* FICHIER JSON = {filename}
*************************************
"""
return text.format(**vars(self))
[docs]
def read(self):
"""
Lire le résultat d'un web-service BdApbp.
Returns
-------
pandas.DataFrame
Tableau des données
None
Si le contenu est incorrect
Examples
--------
>>> from pyspc.data.lamedo import BdApbp
Exemple de prévision BdApbp au format 'short'
>>> f = 'data/webservice/lamedo/bp_short.json'
>>> reader = BdApbp(filename=f)
>>> content = reader.read()
>>> content
CODE NAME DATE MOY DELTA LOC LOCDELTA DTPROD
0 41003 NaN 2020-06-13 850 NaN 1100.0 NaN 2020-06-12 06:00:00
1 41003 NaN 2020-06-14 110 NaN NaN NaN 2020-06-12 06:00:00
2 41003 NaN 2020-06-15 0 NaN NaN NaN 2020-06-12 06:00:00
3 41005 NaN 2020-06-13 1150 NaN 2000.0 NaN 2020-06-12 06:00:00
4 41005 NaN 2020-06-14 110 NaN NaN NaN 2020-06-12 06:00:00
5 41005 NaN 2020-06-15 0 NaN NaN NaN 2020-06-12 06:00:00
Exemple de prévision BdApbp au format 'long'
>>> f = 'data/webservice/lamedo/bp_long.json'
>>> reader = BdApbp(filename=f)
>>> content = reader.read()
>>> content
CODE NAME DATE MOY DELTA LOC LOCDELTA DTPROD
0 41003 Borne - Ance du Nord 2020-06-13 850 150.0 1100.0 NaN 2020-06-12 05:26:00
1 41003 Borne - Ance du Nord 2020-06-14 110 40.0 NaN NaN 2020-06-12 05:26:00
2 41003 Borne - Ance du Nord 2020-06-15 0 NaN NaN NaN 2020-06-12 05:26:00
3 41005 Source Loire 2020-06-13 1150 150.0 2000.0 NaN 2020-06-12 05:26:00
4 41005 Source Loire 2020-06-14 110 40.0 NaN NaN 2020-06-12 05:26:00
5 41005 Source Loire 2020-06-15 0 NaN NaN NaN 2020-06-12 05:26:00
"""
# ---------------------------------------------------------------------
# 0- Chargement des données
# ---------------------------------------------------------------------
with open(self.filename, 'r', encoding='utf-8') as j:
content = json.load(j)
# ---------------------------------------------------------------------
# 1- Contrôle des données
# ---------------------------------------------------------------------
if content['statut'] != 0:
_exception.Warning(
__name__,
f"Le contenu JSON est incorrect dans {self.filename}")
return None
try:
self._datatype = content['rapport']['requete']['format']
except KeyError:
self._datatype = None
self.check_datatypes(self._datatype)
dtprod = content['rapport']['requete']['date']
# ---------------------------------------------------------------------
# 2- Conversion en dataframe
# ---------------------------------------------------------------------
df = pnd.DataFrame(content['data'])
_exception.raise_valueerror(
df.empty,
f"Le fichier {self.filename} ne contient aucune valeur BP")
df.columns = BDAPBP_COLUMNS[self._datatype]
df = df.reindex(columns=BDAPBP_COLUMNS['long']) # Colonnes par défaut
df["CODE"] = df["CODE"].astype(str)
df["DTPROD"] = df["DTPROD"].fillna(value=dtprod)
df["DATE"] = df["DATE"].map(
lambda x: dt.strptime(x, BDAPBP_DATE_FORMAT))
df["DTPROD"] = df["DTPROD"].map(
lambda x: dt.strptime(x, BDAPBP_DATE_FORMAT))
df["DELTA"] = df["DELTA"].map(
lambda x: np.nan if x is None else x)
df["LOC"] = df["LOC"].map(
lambda x: np.nan if x is None else x)
df["LOCDELTA"] = df["LOCDELTA"].map(
lambda x: np.nan if x is None else x)
# Tri par zone AP puis par date
df.sort_values(by=['CODE', 'DATE'], axis=0, inplace=True)
df.reset_index(inplace=True)
df.drop('index', axis=1, inplace=True)
# ---------------------------------------------------------------------
# 3- Retour
# ---------------------------------------------------------------------
return df
[docs]
def write(self, data=None, datatype=None):
"""
Ecrire un fichier JSON au format de BdApbp.
Parameters
----------
data : pandas.DataFrame
Tableau des données
datatype : str
Type de fichier json
See Also
--------
BdApbp.get_types
"""
# ---------------------------------------------------------------------
# 0- Contrôles
# ---------------------------------------------------------------------
self.check_datatypes(datatype)
_exception.check_dataframe(data)
# ---------------------------------------------------------------------
# 1- Zones et Date
# ---------------------------------------------------------------------
zones = sorted(set(data['CODE'].values))
zones_str = " ".join(zones)
runtime = pnd.to_datetime(sorted(set(data['DTPROD'].values))[-1])
# ---------------------------------------------------------------------
# 2- Rapport
# ---------------------------------------------------------------------
content = {}
content['rapport'] = {
"date": None,
"duree": None,
"requete": {
"zones": zones_str,
"version": "201107",
"service": "pyspc.data.lamedo",
"format": datatype,
"date": runtime.strftime(BDAPBP_DATE_FORMAT),
"request": "BdApbp.write",
},
"message": "execution correcte"
}
# ---------------------------------------------------------------------
# 3- Données
# ---------------------------------------------------------------------
df = data.reindex(columns=BDAPBP_COLUMNS[datatype])
df['CODE'] = df['CODE'].astype(int)
df["DATE"] = df["DATE"].map(lambda x: x.strftime(BDAPBP_DATE_FORMAT))
try:
df["DTPROD"] = df["DTPROD"].map(
lambda x: x.strftime(BDAPBP_DATE_FORMAT))
except KeyError:
pass
df = df.astype(object)
df = df.where(df.notnull(), None)
content['data'] = df.values.tolist()
# ---------------------------------------------------------------------
# 4- Statut
# ---------------------------------------------------------------------
content['statut'] = 0
# ---------------------------------------------------------------------
# 5- Ecriture
# ---------------------------------------------------------------------
with open(self.filename, 'w', encoding='utf-8') as outfile:
json.dump(content, outfile)
[docs]
def check_datatypes(self, datatype):
"""
Vérifier le format des données ApBp.
Parameters
----------
datatype
Format à vérifier
Raises
------
ValueError
Si le formar est incorrect
"""
_exception.raise_valueerror(
datatype not in self.get_types(),
f"Format incorrect. {datatype} n'est pas compatible avec "
f"{self.get_types()}")
[docs]
@classmethod
def get_types(cls):
"""
Lister les formats des données ApBp au format JSON.
- short : retour concis
- long : retour complet
"""
return sorted(BDAPBP_DATATYPES)