162 lines
4.5 KiB
Python
162 lines
4.5 KiB
Python
#!/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
|
||
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"
|
||
DOSSIER_SAUVEGARDE = "resultats/"
|
||
|
||
# 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)
|
||
|
||
return parser
|
||
|
||
|
||
def main():
|
||
"""Routine principale"""
|
||
|
||
parser = init_argparse()
|
||
args = parser.parse_args()
|
||
|
||
# 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:
|
||
utils = Utils(OVERPASS_URL, GEO_API_URL, DOSSIER_SAUVEGARDE)
|
||
|
||
print(f"{75*'#'}\r\nRequête en cours : {req.nom}")
|
||
|
||
# appel overpass
|
||
data = utils.run_overpass_query(req.critere, aire_de_recherche)
|
||
nb_resultats = len(data["elements"])
|
||
print(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(DOSSIER_SAUVEGARDE, 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:
|
||
print("trop d'erreurs d'API - abandon")
|
||
exit()
|
||
|
||
print("erreur API - on retente dans " + str(RETRY_DELAY) + "s")
|
||
|
||
time.sleep(RETRY_DELAY)
|
||
|
||
print("\r\n ### Terminé ###")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|