#!/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 - LAMEDO - write
"""
from datetime import timedelta as td
import numpy as np
import os.path
import pandas as pnd
from pyspc.core.series import Series
import pyspc.core.exception as _exception
from pyspc.convention.lamedo import (
BDAPBP_DATE_FORMAT, BDAPBP_SCENS, BDAPBP_RAWSCENS)
from pyspc.data.lamedo import BdApbp
write_BdImage = None
[docs]
def write_BdApbp(series=None, datatype=None, dirname='.'):
"""
Créer un fichier de données MF à partir d'une instance Series
Parameters
----------
series : pyspc.core.series.Series
Collection de séries de données
datatype : str
Type de fichier json
dirname : str
Répertoire où écrire le fichier
Returns
-------
filenames : list
Liste des noms de fichiers créés
"""
# -------------------------------------------------------------------------
# 0- Contrôles
# -------------------------------------------------------------------------
_exception.check_str(dirname)
_exception.raise_valueerror(
not isinstance(series, Series),
"'series' doit être une collection 'Series'"
)
# -------------------------------------------------------------------------
# 1- Détermination des sous-collections
# -------------------------------------------------------------------------
filenames = []
groups = {}
for k in series:
r = k[2][0]
m = k[2][1]
s = k[2][2]
if s not in BDAPBP_SCENS:
continue
groups.setdefault((r, m), [])
groups[(r, m)].append(k)
if not groups:
return None
# -------------------------------------------------------------------------
# 2- Manipulation du DataFrame
# -------------------------------------------------------------------------
for g in groups:
# Fichier JSON
f = os.path.join(
dirname, f'{g[1]}_{g[0].strftime(BDAPBP_DATE_FORMAT)}.json')
# Création de la sous-collection
df = series.concat(keys=groups[g])
# Nommage des colonnes
df.columns = pnd.MultiIndex.from_tuples([(c[0], c[2][2])
for c in df.columns])
# Conversion de la journée
df.index = [i + td(days=1) for i in df.index]
# Ajout du CODE dans l'index
df = df.stack(0) # 0: POSTE
df.index.rename('DATE', level=0, inplace=True)
df.index.rename('CODE', level=1, inplace=True)
df = df.reindex(columns=BDAPBP_SCENS)
# Définir les valeurs moyennes et loc
col = df.apply(list_values, axis=1)
df = df.assign(VALUE=col.values) # VALUE : colonne avec les 4 valeurs
df[BDAPBP_RAWSCENS] = pnd.DataFrame(df.VALUE.to_list(), index=df.index)
df = df.reindex(columns=BDAPBP_RAWSCENS)
# Retirer l'index -> colonne
df = df.reset_index()
# Ajout des colonnes NAME et DTPROD
df['NAME'] = df['CODE']
df['NAME'].astype(str)
df['DTPROD'] = [g[0]] * len(df.index)
# -------------------------------------------------------------------------
# 3- Export BdApbp
# -------------------------------------------------------------------------
writer = BdApbp(filename=f)
writer.write(data=df, datatype=datatype)
filenames.append(f)
return filenames
def list_values(row):
"""
Calculer les valeurs 'MOY', 'DELTA', 'LOC', 'LOCDELTA'
selon les colonnes 'MoyInf', 'MoySup', 'LocInf', 'LocSup'
Parameters
----------
row : pandas.Series
Ligne d'un tableau de données
Returns
-------
list
Valeurs 'MoyInf', 'MoySup', 'LocInf', 'LocSup'
"""
ratio = 10 # mm => 1/10 mm
mi = row['MoyInf']
ms = row['MoySup']
if np.isnan(mi) and np.isnan(ms):
m = np.nan
d = np.nan
elif np.isnan(mi):
m = int(ms * ratio)
d = np.nan
elif np.isnan(ms):
m = int(mi * ratio)
d = np.nan
else:
m = int((mi + ms) / 2 * ratio)
d = int(max(mi, ms) * ratio - m)
li = row['LocInf']
ls = row['LocSup']
if np.isnan(li) and np.isnan(ls):
lo = np.nan
ld = np.nan
elif np.isnan(li):
lo = int(ls * ratio)
ld = np.nan
elif np.isnan(ls):
lo = int(li * ratio)
ld = np.nan
else:
lo = int((li + ls) / 2 * ratio)
ld = int(max(li, ls) * ratio - lo)
return [m, d, lo, ld]