#!/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 - Csv natif PyspcFile - write
"""
import os.path
from pyspc.core.keyseries import tuple2str
from pyspc.core.pyspcfile import PyspcFile
from pyspc.core.timeutil import dtheader
import pyspc.core.exception as _exception
[docs]
def write_PyspcFile(series=None, dirname='.', code=None, how='replace',
onefile=False, explicitcolname=False):
"""
Ecrire des fichiers pyspcFile
Parameters
----------
series : pyspc.core.series.Series
Collection de séries de données
dirname : str
Répertoire où écrire les fichiers
Other Parameters
----------------
code : str
Identifiant de la collection, utilisé si onefile = True
et si la collection contient lplusieurs stations
Si non défini, le 'name' de la collection sera utilisé
onefile : bool
Enregister la collection dans un unique fichier. Défaut: False
Si la collection ne contient qu'une série,
alors cette option est ignorée
explicitcolname : bool
Forcer le nommage explicite des colonnes. Défaut: False
how : str
Option d'écriture, uniquement si onefile = False
- replace : écraser si un fichier existe déjà (défaut)
- fillna : mettre à jour le fichier existant
(seulement les NaN et les valeurs aux instants non existants)
- overwrite : mettre à jour le fichier existant
(toutes les valeurs, y compris les non-NaN)
Si onefile = True, le fichier existant est écrasé.
Returns
-------
filenames : list
Liste des noms de fichiers créés
"""
# -------------------------------------------------------------------------
# 0- Contrôles
# -------------------------------------------------------------------------
ts = {s.timestep for s in series.values()}
_exception.raise_valueerror(
onefile and len(ts) > 1,
'La collection contient des séries avec des pas de temps '
'différents. Aucun export réalisé.'
)
# -------------------------------------------------------------------------
# 1- Regroupement de séries dans un fichier
# -------------------------------------------------------------------------
if onefile and len(series) > 1:
return _pyspcfile_onefile(series=series, dirname=dirname, code=code,
explicitcolname=explicitcolname, ts=ts)
# -------------------------------------------------------------------------
# 2- Par défaut, création d'un fichier par série de données
# -------------------------------------------------------------------------
return _pyspcfile_multi(series=series, dirname=dirname, how=how,
explicitcolname=explicitcolname)
def _pyspcfile_findcommonpart(keys_as_str):
"""
Recherche de la partie commune depuis la gauche
"""
return os.path.commonpath([k.replace('_', '/')
for k in keys_as_str])\
.replace('/', '_').replace('\\', '_')
def _pyspcfile_table_key_colname(keys_as_str, common_part, main_varname,
explicitcolname, datatype):
"""
Définir les noms des colonnes selon leur label (str) et leur partie commune
"""
assoc = {}
for k in keys_as_str:
if explicitcolname:
assoc[k] = k
else:
s = k[len(common_part):].replace(main_varname, '').strip('_')
if not s and datatype == 'obs':
s = 'obs'
# elif not s:
# s = '-1'
assoc[k] = s
return assoc
def _pyspcfile_multi(series=None, dirname='.', how=None,
explicitcolname=False):
"""
Export pyspcFile - Cas d'un fichier par série de données
"""
from pyspc.io.pyspcfile.reader import read_PyspcFile
filenames = []
for key, serie in series.items():
basename = tuple2str(key)
if explicitcolname:
colname = basename
else:
colname = f'{serie.spc_varname}({serie.units})'
filename = os.path.join(dirname, f'{basename}.txt')
# Charger les données existantes si la série les complète
if os.path.exists(filename) and how in ['fillna', 'overwrite']:
try:
other = read_PyspcFile(filename=filename)[key]
other.data_frame.columns = serie.data_frame.columns
if how == 'fillna':
other.update(serie, overwrite=False)
else:
other.update(serie, overwrite=True)
except (KeyError, ValueError):
pass
else:
serie = other
# df = serie.data_frame
df = serie.data_frame.copy(deep=True)
df.columns = [colname]
df.index.name = dtheader(tdelta=serie.timestep)
writer = PyspcFile(filename=filename)
writer.write(data=df)
filenames.append(writer.filename)
return filenames
def _pyspcfile_onefile(series=None, dirname=None, code=None,
explicitcolname=False, ts=None):
"""
Export pyspcFile - Cas d'un fichier pour la collection complète
"""
# Liste des clés sous forme de str
kss = [tuple2str(k) for k in series]
# Recherche de la partie commune depuis la gauche
cp = _pyspcfile_findcommonpart(kss)
if not cp or len(set(series.varnames)) > 1:
explicitcolname = True
# Table association
assoc = _pyspcfile_table_key_colname(
kss, cp, series.varnames[1], explicitcolname, series.datatype)
# if not explicitcolname:
if cp and not explicitcolname:
basename = PyspcFile.join_basename(
station=cp.replace('/', '_'),
varname=series.varnames[1])
# elif cp:
# basename = PyspcFile.join_basename(
# station=series.name.replace('_', ''),
# varname=series.varnames[1])
elif isinstance(code, str):
basename = PyspcFile.join_basename(
station=code,
varname=series.varnames[1])
else:
basename = PyspcFile.join_basename(
station=series.name.replace('_', ''),
varname=series.varnames[1])
filename = os.path.join(dirname, basename)
df = series.concat()
df.columns = [assoc[tuple2str(c)] for c in df.columns]
df.index.name = dtheader(tdelta=list(ts)[0])
writer = PyspcFile(filename=filename)
writer.write(data=df)
return [writer.filename]