Code source de pyspc.core.reach

#!/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/>.
#
########################################################################
"""
Objets natifs et convention de pyspc - Tronçon(s) de vigilance
"""
import collections
from datetime import datetime as dt
import os.path

from pyspc.core.config import Config
import pyspc.core.exception as _exception


[docs] class Reach(): """ Tronçon de vigilance Attributes ---------- code : str Code du tronçon name : str Nom du tronçon status : str Statut de vigilance status_dt : datetime Date de mise-à-jour du statut locations : Locations, list Collection des lieux du tronçon """
[docs] def __init__(self, code=None, name=None, status=None, status_dt=None, locations=None): """ Initialiser le tronçons de vigilance Parameters ---------- code : str Code du tronçon name : str Nom du tronçon status : str Statut de vigilance status_dt : datetime Date de mise-à-jour du statut locations : Locations, list Collection des lieux du tronçon """ import pyspc.core.location as _loc self._code = code self._name = name self._status = status if isinstance(status_dt, dt) or status_dt is None: self._status_dt = status_dt else: raise ValueError("La date du statut n'est pas un datetime") if isinstance(locations, (_loc.Locations, list)) or locations is None: self._locations = locations else: raise ValueError('Collection de lieux incorrecte')
def __str__(self): """ Afficher des méta-données de l'instance Reach """ strlocs = '' length = 0 if isinstance(self._locations, list): counter = 0 length = len(self._locations) for c in self._locations: counter += 1 strlocs += '\n * ----------------------------------' strlocs += f'\n * LIEU #{counter}' strlocs += f'\n * - CODE = {c}' elif self._locations is not None: counter = 0 length = len(self._locations) for c in self._locations.values(): counter += 1 strlocs += '\n * ----------------------------------' strlocs += f'\n * LIEU #{counter}' strlocs += f'\n * - CODE = {c.code}' strlocs += f'\n * - NOM = {c.name}' text = """ ************************************* ************ REACH ****************** ************************************* * CODE TRONCON = {_code} * NOM TRONCON = {_name} * STATUT TRONCON = {_status} * DATE STATUT = {_status_dt} * NOMBRE DE LIEUX = {length} {strlocs} ************************************* """ return text.format(**vars(self), length=length, strlocs=strlocs) @property def code(self): """Code du tronçon""" return self._code @property def locations(self): """Collection des lieux du tronçon""" return self._locations @property def name(self): """Nom du tronçon""" return self._name @property def status(self): """Statut de vigilance""" return self._status @property def status_dt(self): """Date de mise-à-jour du statut""" return self._status_dt
[docs] class Reaches(collections.OrderedDict): """ Classe des collections de tronçons Reach Attributes ---------- name : str Nom de la collection codes : list Liste des codes des tronçons """
[docs] def __init__(self, name='reaches'): """ Initialiser l'instance de Reaches Parameters ---------- name : str Nom de la collection """ super().__init__() self._name = name self._codes = []
def __str__(self): """ Afficher les méta-données de l'instance Reaches """ strreaches = '' if len(self.keys()) > 0: counter = 0 for c in self.codes: counter += 1 strreaches += '\n * ----------------------------------' strreaches += f'\n * TRONCON #{counter}' strreaches += f'\n * - CODE = {c}' text = """ ************************************* ********* REACHES ******************* ************************************* * NOM DE LA COLLECTION = {_name} * NOMBRE DE LIEUX = {length} {strreaches} ************************************* """ return text.format(**vars(self), length=len(self.keys()), strreaches=strreaches) @property def codes(self): """Liste des codes des tronçons""" return self._codes @property def name(self): """Nom de la collection""" return self._name
[docs] def add(self, reach=None, code=None, overwrite=False): """ Ajouter un tronçon hydrologique dans la collection Parameters ---------- code : str Code du tronçon reach : Reach Instance du tronçon overwrite : bool Écraser la donnée existante ? défaut: False """ _exception.raise_valueerror( not isinstance(reach, Reach), 'Reach incorrecte') if code is None: code = reach.code _exception.check_str(code) if overwrite: self[code] = reach else: self.setdefault(code, reach) self.refresh_codes()
[docs] def check_reaches(self): """ Contrôler si chaque tronçon est une instance Reach """ return all([isinstance(v, Reach) for k, v in self.items()])
[docs] def refresh_codes(self): """ Rafraîchir la liste des codes """ self._codes = list(self.keys())
[docs] @classmethod def from_Config(cls, config=None): """ Créer une instance Reaches à partir d'un fichier de configuration ou d'une instance Config Parameters ---------- config : str, Config - Nom du fichier de configuration - Instance pyspc.core.config.Config """ name = os.path.basename(os.path.splitext(config)[0]) reaches = Reaches(name=name) if not isinstance(config, Config): config = Config(filename=config) config.read() for section in config: reach = Reach( code=section, name=config[section].get('name', None), status=config[section].get('status', None), status_dt=config[section].get('status_dt', None) ) reaches.add(code=section, reach=reach) return reaches
[docs] def to_Config(self, filename=None): """ Créer une instance Config à partir d'une instance Locations Parameters ---------- filename : str Nom du fichier associé à la collection. Si non défini, le nom du fichier est défini à partir du nom de la collection Returns ------- Config Instance pyspc.core.config.Config """ if filename is None: filename = f'{self.name}.txt' _exception.check_str(filename) config = Config(filename=filename) for k, reach in self.items(): config.setdefault(k, collections.OrderedDict()) config[k]['code'] = k config[k]['name'] = reach.name config[k]['status'] = reach.status config[k]['status_dt'] = reach.status_dt try: config[k]['locations'] = ','.join( sorted(reach.locations.keys())) except AttributeError: config[k]['locations'] = '' return config