outils_OSM/rdoo.py

#!/usr/bin/env python3


# Copyright 2021 Olav63, SebF
#
# 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.  If not, see <https://www.gnu.org/licenses/>.


"""
Module principal : 
- récupération de données par appel à Overpass
- géocodage inverse
- export des données en JSON pour utilisation avec umap
- sauvegarde des données en ods
"""


import time
import os
import argparse
import logging
import sys
from osm_vc63 import errors
from osm_vc63 import requetes
from osm_vc63.utils import Utils

OVERPASS_URL = "http://overpass-api.de/api/interpreter"
GEO_API_URL = "https://api-adresse.data.gouv.fr"

# nombre maxi de retries quand echec API
MAX_RETRY = 4

# delai en secondes entre les tentatives
RETRY_DELAY = 120


# traductions des tags bicycle_parking
TRAD_BICYCLE_PARKING = {
    "stands": "Arceaux",
    "wall_loops": "Pince roues",
    "rack": "Râteliers",
    "anchors": "Ancrage",
    "shed": "Abri collectif",
    "bollard": "Potelet",
    "lockers": "Abris individuels",
    "wide_stands": "Arceaux espacés",
    "ground_slots": "Fente dans le sol",
    "building": "Bâtiment",
    "informal": "Informel",
    "wave": "Râteliers",
    "streetpod": "Arceaux",
    "tree": "Arbre à bicyclettes",
    "crossbar": "Barre",
    "rope": "Câble",
    "two-tier": "Deux étages",
    "floor": "Sol",
    "handlebar_holder": "Accroche-guidons",
}


def init_argparse() -> argparse.ArgumentParser:
    """Définition des arguments possibles."""

    parser = argparse.ArgumentParser(
        usage="%(prog)s [OPTIONS] ...",
        description="Exporte les données de la Cyclosphère d'une zone géographique.",
    )

    parser.add_argument(
        "-z",
        "--zone",
        type=int,
        help="Choisir la zone géographique à inspecter.",
        default=7406,
    )

    parser.add_argument(
        "-gi",
        "--geocoding-inverse",
        dest="geocoding_inverse",
        action="store_true",
        help="Activer le géocodage inverse",
    )
    parser.add_argument(
        "-ngi",
        "--no-geocoding-inverse",
        dest="geocoding_inverse",
        action="store_false",
        help="Désactiver le géocodage inverse",
    )
    parser.set_defaults(geocoding_inverse=True)

    parser.add_argument(
        "-l", "--log-level", type=str, help="Définir le niveau de log.", default="INFO"
    )

    parser.add_argument(
        "-r",
        "--dossier-resultats",
        type=str,
        help="Définir le dossier d'archive",
        default="resultats/",
    )

    parser.add_argument(
        "-a",
        "--dossier-archive",
        type=str,
        help="Définir le dossier d'archive",
        default="resultats/archives/",
    )

    parser.add_argument(
        "-na",
        "--no-archivage",
        dest="archivage",
        action="store_false",
        help="Désactiver l'archivage'",
    )
    parser.set_defaults(archivage=True)

    return parser


def main():
    """Routine principale"""

    parser = init_argparse()
    args = parser.parse_args()

    logging.basicConfig(
        format="%(asctime)s [%(levelname)s] %(message)s",
        level=getattr(logging, args.log_level.upper()),
        handlers=[logging.FileHandler("rdoo.log"), logging.StreamHandler(sys.stdout)],
    )

    utils = Utils(OVERPASS_URL, GEO_API_URL, args.dossier_resultats)

    if args.archivage:
        utils.archivage(args.dossier_archive)

    logging.info("##### Nouvelle récupération ######")

    # l'id de l'area se calcule en ajoutant 3600000000 au numéro de l'objet OSM
    aire_de_recherche = str(3_600_000_000 + args.zone)

    for req in requetes.REQS:
        for nb_essai in range(MAX_RETRY):  # on tente max_retry fois
            try:

                logging.info(f"# Requête en cours : {req.nom}")

                # appel overpass
                data = utils.run_overpass_query(req.critere, aire_de_recherche)
                nb_resultats = len(data["elements"])
                logging.info(f"{nb_resultats} résultats")

                if nb_resultats > 0:
                    if args.geocoding_inverse:
                        # géocodage inverse
                        data = utils.geocodage(data)

                    # traduction
                    data = utils.traduction(
                        "bicycle_parking", TRAD_BICYCLE_PARKING, data
                    )

                    # Sauvegarde
                    os.makedirs(args.dossier_resultats, exist_ok=True)
                    export_json = utils.nettoyage_json_pour_umap(data, req.champs)

                    utils.save_as_json(export_json, req.nom)
                    utils.save_as_ods(req.champs, data, req.nom)

                # doucement sur overpass
                time.sleep(30)
                break
            except errors.ApiError:

                if nb_essai == MAX_RETRY:
                    logging.error("Trop d'erreurs d'API - abandon")
                exit()

                logging.error(f"Erreur API - on retente dans {RETRY_DELAY}s")

            time.sleep(RETRY_DELAY)

    logging.info("#####   Terminé   #####")


if __name__ == "__main__":
    main()