#!/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 - RefSPC - read
"""
from pyspc.convention.refspc import ATT_L_h3, ATT_L_m, ATT_R
from pyspc.metadata.refspc import RefSPC
import pyspc.core.exception as _exception
from pyspc.core.config import Config
from pyspc.core.location import Location, Locations
from pyspc.core.reach import Reach, Reaches
[docs]
def read_RefSPC(filename=None, datatype=None, codes=None, hydro3=True):
"""
Créer une instance Reaches à partir du référentiel du SPC LCI
Parameters
----------
filename : str
Nom du fichier de la base
datatype : str
Type d'entités du référentiel
- reach : Tronçon de vigilance
- loc_hydro : Lieu de mesure hydro
- loc_meteo : Lieu de mesure météo
Voir pyspc.metadata.refspc.RefSPC.get_datatypes
codes : list
Liste des codes de tronçons / lieux
hydro3 : bool
Codes des lieux selon convention Hydro3
défaut: True
Returns
-------
reaches : Reaches
Collection de tronçon de Vigicrues
locations : Locations
Collection de lieux hydrologiques
Examples
--------
>>> from pyspc.io.refspc import read_RefSPC
>>> f = 'data/metadata/refspc/BDD_light.sqlite'
Cas des tronçons de vigilance
>>> codes = ['LC105', 'LC134', 'LC265']
>>> datatype = 'reach'
>>> reaches = read_RefSPC(
... filename=filename, codes=codes, datatype=datatype)
>>> reaches
*************************************
********* REACHES *******************
*************************************
* NOM DE LA COLLECTION = RefSPC
* NOMBRE DE LIEUX = 3
* ----------------------------------
* TRONCON #1
* - CODE = LC105
* ----------------------------------
* TRONCON #2
* - CODE = LC134
* ----------------------------------
* TRONCON #3
* - CODE = LC265
*************************************
>>> reaches['LC105']
*************************************
************ REACH ******************
*************************************
* CODE TRONCON = LC105
* NOM TRONCON = Loire vellave
* STATUT TRONCON = None
* DATE STATUT = None
* NOMBRE DE LIEUX = 16
* ----------------------------------
* LIEU #1
* - CODE = K001002010
* - NOM = None
* ----------------------------------
* LIEU #2
* - CODE = K001872010
* - NOM = None
* ----------------------------------
* LIEU #3
* - CODE = K003002010
* - NOM = None
* ----------------------------------
* LIEU #4
* - CODE = K004551001
* - NOM = None
* ----------------------------------
* LIEU #5
* - CODE = K010002010
* - NOM = None
* ----------------------------------
* LIEU #6
* - CODE = K011402001
* - NOM = None
* ----------------------------------
* LIEU #7
* - CODE = K013401001
* - NOM = None
* ----------------------------------
* LIEU #8
* - CODE = K021401001
* - NOM = None
* ----------------------------------
* LIEU #9
* - CODE = K026001002
* - NOM = None
* ----------------------------------
* LIEU #10
* - CODE = K027401002
* - NOM = None
* ----------------------------------
* LIEU #11
* - CODE = K033301001
* - NOM = None
* ----------------------------------
* LIEU #12
* - CODE = K035631001
* - NOM = None
* ----------------------------------
* LIEU #13
* - CODE = K051301001
* - NOM = None
* ----------------------------------
* LIEU #14
* - CODE = K052301001
* - NOM = None
* ----------------------------------
* LIEU #15
* - CODE = K054301001
* - NOM = None
* ----------------------------------
* LIEU #16
* - CODE = K055001010
* - NOM = None
*************************************
Cas des lieux météorologiques
>>> codes = ['07235005', '43042002', '43091005']
>>> datatype = 'loc_meteo'
>>> locs = read_RefSPC(
... filename=filename, codes=codes, datatype=datatype)
>>> locs
*************************************
*********** LOCATIONS ***************
*************************************
* NOM DE LA COLLECTION = RefSPC
* NOMBRE DE LIEUX = 3
* ----------------------------------
* LIEU #1
* - CODE = 07235005
* ----------------------------------
* LIEU #2
* - CODE = 43042002
* ----------------------------------
* LIEU #3
* - CODE = 43091005
*************************************
>>> locs['07235005']
*************************************
*********** LOCATION ****************
*************************************
* CODE LIEU = 07235005
* NOM LIEU = None
* NOM COMPLET LIEU = Ste-Eulalie
* COURS D'EAU = None
* TYPE LIEU = point
* COORDONNEES X = 793976.00 m
* COORDONNEES Y = 6413101.00 m
* ALTITUDE LIEU = 1240.00 m NGF
* SURFACE LIEU = -1.00 km2
* COMMUNES = None
* TRONCONS = None
*************************************
Cas des lieux hydrologiques
>>> codes = ['K0010020', 'K0114020', 'K0550010']
>>> datatype = 'loc_hydro'
>>> locs = read_RefSPC(
... filename=filename, codes=codes, datatype=datatype)
>>> locs
*************************************
*********** LOCATIONS ***************
*************************************
* NOM DE LA COLLECTION = RefSPC
* NOMBRE DE LIEUX = 3
* ----------------------------------
* LIEU #1
* - CODE = K0010020
* ----------------------------------
* LIEU #2
* - CODE = K0114020
* ----------------------------------
* LIEU #3
* - CODE = K0550010
*************************************
>>> locs['K0550010']
*************************************
*********** LOCATION ****************
*************************************
* CODE LIEU = K0550010
* NOM LIEU = Bas-en-Basset
* NOM COMPLET LIEU = Bas-en-Basset
* COURS D'EAU = Loire
* TYPE LIEU = basin
* COORDONNEES X = 787596.00 m
* COORDONNEES Y = 6466797.00 m
* ALTITUDE LIEU = -1.00 m NGF
* SURFACE LIEU = 3234.00 km2
* COMMUNES = 43
* TRONCONS = []
*************************************
"""
# -------------------------------------------------------------------------
# 0- Contrôles
# -------------------------------------------------------------------------
_exception.check_str(filename)
_exception.check_listlike(codes)
_exception.check_bool(hydro3)
reader = RefSPC(filename=filename)
reader._check_datatype(datatype)
# -------------------------------------------------------------------------
# 1- Cas des tronçons
# -------------------------------------------------------------------------
if datatype == 'reach':
return _refspc_reach(reader, codes, hydro3)
# -------------------------------------------------------------------------
# 2- Cas des lieux HYDRO
# -------------------------------------------------------------------------
if datatype == 'loc_hydro':
return _refspc_loc_hydro(reader, codes, hydro3)
# -------------------------------------------------------------------------
# 3- Cas des stats HYDRO
# -------------------------------------------------------------------------
if datatype == 'stat_hydro':
return _refspc_stat_hydro(reader, codes)
# -------------------------------------------------------------------------
# 4- Cas des lieux METEO
# -------------------------------------------------------------------------
if datatype == 'loc_meteo':
return _refspc_loc_meteo(reader, codes)
# -------------------------------------------------------------------------
# 5- Cas des stats METEO
# -------------------------------------------------------------------------
if datatype == 'stat_meteo':
return _refspc_stat_meteo(reader, codes)
raise ValueError('Type de données non autorisé')
def _refspc_loc_hydro(reader, codes, hydro3):
""""""
key_code = None
key_name = None
key_longname = None
key_x = None
key_y = None
loctype = None
locations = Locations(name='RefSPC')
header, content = reader.get_loc_hydro(codes=codes, hydro3=hydro3)
if 'code_site' in header:
key_code = ATT_L_h3[8][1]
key_name = ATT_L_h3[8][2]
key_x = ATT_L_h3[8][8]
key_y = ATT_L_h3[8][9]
# key_code = 'code_site'
# key_name = 'nom_site_phyc'
key_longname = key_name
loctype = 'basin'
elif 'code_station' in header:
key_code = ATT_L_h3[10][1]
key_name = ATT_L_h3[10][2]
key_x = ATT_L_h3[10][5]
key_y = ATT_L_h3[10][6]
# key_code = 'code_station'
# key_name = 'nom_station_phyc'
key_longname = key_name
loctype = 'point'
elif 'code_capteur' in header and hydro3:
key_code = ATT_L_h3[12][2]
key_name = ATT_L_h3[12][3]
key_x = ATT_L_h3[12][6]
key_y = ATT_L_h3[12][7]
# key_code = 'code_capteur'
# key_name = 'nom_capteur_phyc'
key_longname = key_name
loctype = 'point'
elif 'code_hydro2' in header and not hydro3:
key_code = ATT_L_h3[12][1]
key_name = ATT_L_h3[12][3]
key_x = ATT_L_h3[12][6]
key_y = ATT_L_h3[12][7]
# key_code = 'code_hydro2'
# key_name = 'nom_capteur_phyc'
key_longname = key_name
loctype = 'point'
for c in content:
loc = Location(
code=content[c].get(key_code, ''),
name=content[c].get(key_name, ''),
longname=content[c].get(key_longname, ''),
loctype=loctype,
river=content[c].get(ATT_L_h3[8][3], ''),
locality=content[c].get(ATT_L_h3[8][5], ''),
reach=content[c].get('reaches', []),
x=content[c].get(key_x, -1),
y=content[c].get(key_y, -1),
z=content[c].get('z', -1),
area=content[c].get(ATT_L_h3[8][7], -1)
)
locations.add(code=loc.code, loc=loc)
return locations
def _refspc_loc_meteo(reader, codes):
""""""
locations = Locations(name='RefSPC')
_, content = reader.get_loc_meteo(codes=codes)
for c in content:
code = content[c].get(ATT_L_m[1], None)
name = content[c].get(ATT_L_m[2], None)
longname = content[c].get(ATT_L_m[3], None)
x = content[c].get(ATT_L_m[4], -1)
y = content[c].get(ATT_L_m[5], -1)
z = content[c].get(ATT_L_m[6], -1)
loc = Location(code=code, name=name, longname=longname,
x=x, y=y, z=z, loctype='point')
locations.add(loc=loc)
return locations
def _refspc_reach(reader, codes, hydro3):
""""""
reaches = Reaches(name='RefSPC')
_, content = reader.get_reach(codes=codes, hydro3=hydro3)
for c in content:
locations = Locations(name='RefSPC')
code = content[c].get(ATT_R[1], None)
name = content[c].get(ATT_R[2], None)
locs = content[c].get('locations', [])
for loc in locs:
loc = Location(code=loc, loctype='point')
locations.add(loc=loc)
reach = Reach(code=code, name=name, locations=locations)
reaches.add(reach=reach)
return reaches
def _refspc_stat_hydro(reader, codes):
""""""
config = Config()
_, content = reader.get_stat_hydro(codes=codes)
config.update(content)
return config
def _refspc_stat_meteo(reader, codes):
""""""
config = Config()
_, content = reader.get_stat_meteo(codes=codes)
config.update(content)
return config