#!/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 - Vigicrues - read
"""
from pyspc.convention.vigicrues import DATATYPES, RATIO_UNITS, TRENDS, VARNAMES
import pyspc.core.exception as _exception
from pyspc.data.vigicrues import Vigicrues_Data, Vigicrues_Fcst
from pyspc.metadata.vigicrues import Vigicrues_Reach, Vigicrues_Location
from pyspc.core.keyseries import str2tuple, tuple2str
from pyspc.core.location import Location, Locations
from pyspc.core.parameter import Parameter
from pyspc.core.reach import Reach, Reaches
from pyspc.core.serie import Serie
from pyspc.core.series import Series
[docs]
def read_Vigicrues(filename=None, datatype=None, codes=None):
"""
Créer une instance Reaches à partir d'un fichier
au format geojson de Vigicrues
Parameters
----------
filename : str
Nom du fichier json de Vigicrues
datatype : str
Type du fichier Vigicrues
Voir pyspc.convention.vigicrues.DATATYPES
codes : list
Liste des codes des entités
Returns
-------
- pyspc.core.reach.Reaches
Collection de tronçon de Vigicrues si datatype = 'vigicrues_reach'
- pyspc.core.location.Locations
Collection de lieux de Vigicrues si datatype = 'vigicrues_loc'
- pyspc.core.series.Series
Collection d'observation de Vigicrues si datatype = 'vigicrues_obs'
- pyspc.core.series.Series
Collection de prévisions de Vigicrues si datatype = 'vigicrues_fcst'
Examples
--------
Cas de tronçons de vigilance
>>> from pyspc.io.vigicrues import read_Vigicrues
>>> f = 'data/metadata/vigicrues/troncons.json'
>>> d = 'vigicrues_reach'
>>> content = read_Vigicrues(filename=f, datatype=d)
>>> content
*************************************
********* REACHES *******************
*************************************
* NOM DE LA COLLECTION = Vigicrues
* NOMBRE DE LIEUX = 2
* ----------------------------------
* TRONCON #1
* - CODE = LC110
* ----------------------------------
* TRONCON #2
* - CODE = LC120
*************************************
>>> content['LC110']
*************************************
************ REACH ******************
*************************************
* CODE TRONCON = LC110
* NOM TRONCON = Haut bassin de la Loire
* STATUT TRONCON = 1
* DATE STATUT = 2018-09-22 07:50:00
* NOMBRE DE LIEUX = 0
*************************************
Cas de station de mesure
>>> f = 'data/metadata/vigicrues/K118001010_station.json'
>>> d = 'vigicrues_loc'
>>> content = read_Vigicrues(filename=f, datatype=d)
>>> content
*************************************
*********** LOCATIONS ***************
*************************************
* NOM DE LA COLLECTION = Vigicrues
* NOMBRE DE LIEUX = 1
* ----------------------------------
* LIEU #1
* - CODE = K118001010
*************************************
>>> content['K118001010']
*************************************
*********** LOCATION ****************
*************************************
* CODE LIEU = K118001010
* NOM LIEU = Digoin [Pont canal]
* NOM COMPLET LIEU = Digoin [Pont canal]
* COURS D'EAU = Loire
* TYPE LIEU = point
* COORDONNEES X = 775417.00 m
* COORDONNEES Y = 6598045.00 m
* ALTITUDE LIEU = -1.00 m NGF
* SURFACE LIEU = -1.00 km2
* COMMUNES = 71176
* TRONCONS = None
*************************************
Cas de séries de données observées
>>> f = 'data/data/vigicrues/K118001010_observation.json'
>>> d = 'vigicrues_obs'
>>> content = read_Vigicrues(filename=f, datatype=d)
>>> content
*************************************
********** SERIES *******************
*************************************
* NOM DE LA COLLECTION = Vigicrues
* TYPE DE COLLECTION = obs
* NOMBRE DE SERIES = 1
* ----------------------------------
* SERIE #1
* - CODE = K118001010
* - VARNAME = QI
* - META = None
*************************************
>>> content[('K118001010', 'QI', None)]
*************************************
*********** SERIE *******************
*************************************
* NOM VARIABLE SPC = QI
* INTITULE VARIABLE = Débit instantané
* IDENTIFIANT = K118001010
* FOURNISSEUR = Provider(name='Vigicrues')
* NOM VARIABLE = QI
* UNITE = m3/s
* VALEUR MANQUANTE = []
* SERIE CONTINUE = True
* PAS DE TEMPS = None
* UNITE DE TEMPS = None
* FUSEAU HORAIRE = UTC
* PROFONDEUR SERIE = 50
* PREMIER PAS DE TEMPS = 2022-03-29 10:00:00
* DERNIER PAS DE TEMPS = 2022-03-29 14:30:00
*************************************
Cas de séries de données prévues
>>> f = 'data/data/vigicrues/K118001010_prevision.json'
>>> d = 'vigicrues_fcst'
>>> content = read_Vigicrues(filename=f, datatype=d)
>>> content
*************************************
********** SERIES *******************
*************************************
* NOM DE LA COLLECTION = Vigicrues
* TYPE DE COLLECTION = fcst
* NOMBRE DE SERIES = 3
* ----------------------------------
* SERIE #1
* - CODE = K118001010
* - VARNAME = QI
* - META = 2022-03-29 12:06:52, Vigicrues, Released, 10
* ----------------------------------
* SERIE #2
* - CODE = K118001010
* - VARNAME = QI
* - META = 2022-03-29 12:06:52, Vigicrues, Released, 50
* ----------------------------------
* SERIE #3
* - CODE = K118001010
* - VARNAME = QI
* - META = 2022-03-29 12:06:52, Vigicrues, Released, 90
*************************************
>>> k = ('K118001010', 'QI',
... (dt(2022, 3, 29, 12, 6, 52), 'Vigicrues', 'Released', '50'))
>>> content[k]
*************************************
*********** SERIE *******************
*************************************
* NOM VARIABLE SPC = QI
* INTITULE VARIABLE = Débit instantané
* IDENTIFIANT = K118001010_2022032912_Vigicrues_Released_50
* FOURNISSEUR = Provider(name='Vigicrues')
* NOM VARIABLE = QI
* UNITE = m3/s
* VALEUR MANQUANTE = []
* SERIE CONTINUE = True
* PAS DE TEMPS = None
* UNITE DE TEMPS = None
* FUSEAU HORAIRE = UTC
* PROFONDEUR SERIE = 9
* PREMIER PAS DE TEMPS = 2022-03-29 15:00:00
* DERNIER PAS DE TEMPS = 2022-03-29 23:00:00
*************************************
"""
# -------------------------------------------------------------------------
# 0- Contrôles
# -------------------------------------------------------------------------
_exception.check_str(filename)
_exception.check_str(datatype)
if codes is not None:
_exception.check_listlike(codes)
_exception.raise_valueerror(
datatype not in DATATYPES, 'Type de données incorrect')
# -------------------------------------------------------------------------
# 1- Méta-données
# -------------------------------------------------------------------------
# ---------------------------------------------------------------------
# 1.1- Locations
# ---------------------------------------------------------------------
if datatype == 'vigicrues_loc':
return _processing_loc(filename=filename, codes=codes)
# ---------------------------------------------------------------------
# 1.2- Reaches
# ---------------------------------------------------------------------
if datatype == 'vigicrues_reach':
return _processing_reach(filename=filename, codes=codes)
# -------------------------------------------------------------------------
# 2- Séries de données
# -------------------------------------------------------------------------
# ---------------------------------------------------------------------
# 2.1- Series - data_obs_hydro
# ---------------------------------------------------------------------
if datatype == 'vigicrues_obs':
return _processing_obs(filename=filename, codes=codes)
# ---------------------------------------------------------------------
# 2.2- Series - data_fcst_hydro
# ---------------------------------------------------------------------
if datatype == 'vigicrues_fcst':
return _processing_fcst(filename=filename, codes=codes)
raise NotImplementedError(f"Type '{datatype}' non implémenté")
def _processing_fcst(filename=None, codes=None):
"""
Convertir un fichier json Vigicrues en collection de prévisions
"""
# -------------------------------------------------------------------------
# 1- Lecture brute
# -------------------------------------------------------------------------
reader = Vigicrues_Fcst(filename=filename)
content = reader.read()
# -------------------------------------------------------------------------
# 2- Création de la collection
# -------------------------------------------------------------------------
series = Series(datatype='fcst', name='Vigicrues')
for c in content.columns:
loc = c[0]
varname = c[1]
runtime = c[2]
trend = c[3]
if varname not in VARNAMES:
continue
if trend not in TRENDS:
continue
if isinstance(codes, list) and loc not in codes:
continue
df = content[c].to_frame()
df = df * RATIO_UNITS[varname]
param = Parameter.find(prefix=varname, timedelta=None)
key = (loc, param.spc_varname,
(runtime, 'Vigicrues', 'Released', TRENDS[trend]))
keystr = str2tuple(tuple2str(key), forceobs=True)[0]
serie = Serie(df[c].to_frame(), code=keystr,
varname=param.spc_varname, provider='Vigicrues')
series.add(serie=serie, code=loc, meta=key[2])
return series
def _processing_obs(filename=None, codes=None):
"""
Convertir un fichier json Vigicrues en collection d'observations
"""
# -------------------------------------------------------------------------
# 1- Lecture brute
# -------------------------------------------------------------------------
reader = Vigicrues_Data(filename=filename)
content = reader.read()
# -------------------------------------------------------------------------
# 2- Création de la collection
# -------------------------------------------------------------------------
series = Series(datatype='obs', name='Vigicrues')
for c in content.columns:
loc = c[0]
varname = c[1]
if varname not in VARNAMES:
continue
if isinstance(codes, list) and loc not in codes:
continue
df = content[c].to_frame()
df = df * RATIO_UNITS[varname]
param = Parameter.find(prefix=varname, timedelta=None)
df.columns = [(loc, param.spc_varname)]
serie = Serie(df, code=loc, varname=param.spc_varname,
provider='Vigicrues')
series.add(serie)
return series
def _processing_loc(filename=None, codes=None):
"""
Convertir un fichier json Vigicrues en collection de lieux
"""
# -------------------------------------------------------------------------
# 1- Lecture brute
# -------------------------------------------------------------------------
reader = Vigicrues_Location(filename=filename)
content = reader.read()
# -------------------------------------------------------------------------
# 2- Création de la collection
# -------------------------------------------------------------------------
locations = Locations(name='Vigicrues')
c = content.get('CdStationHydro', '')
if isinstance(codes, (list, tuple)) and c not in codes:
return locations
loc = Location(
code=c,
name=content.get('LbStationHydro', ''),
longname=content.get('LbStationHydro', ''),
loctype='point',
river=content.get('LbCoursEau', ''),
locality=content.get('CdCommune', ''),
x=content.get('CoordStationHydro', {}).get('CoordXStationHydro', -1),
y=content.get('CoordStationHydro', {}).get('CoordYStationHydro', -1),
z=content.get('z', -1),
area=content.get('taille_bv', -1)
)
locations.add(code=loc.code, loc=loc)
return locations
def _processing_reach(filename=None, codes=None):
"""
Convertir un fichier json Vigicrues en collection de tronçons
"""
# -------------------------------------------------------------------------
# 1- Lecture brute
# -------------------------------------------------------------------------
reader = Vigicrues_Reach(filename=filename)
content = reader.read()
# -------------------------------------------------------------------------
# 2- Création de la collection
# -------------------------------------------------------------------------
reaches = Reaches(name='Vigicrues')
for f in content['features']:
p = f['properties']
if isinstance(codes, (list, tuple)) and p['CdEntVigiCru'] not in codes:
continue
code = p['CdEntVigiCru']
name = p['NomEntVigiCru']
status = p['NivSituVigiCruEnt']
reach = Reach(
code=code,
name=name,
status=status,
status_dt=content['DtHrInfoVigiCru']
)
reaches.add(reach=reach)
return reaches