forked from Olav63/outils_OSM
200 lines
6.3 KiB
Python
200 lines
6.3 KiB
Python
#!/usr/bin/env python3
|
|
|
|
# inspiration :
|
|
# https://towardsdatascience.com/loading-data-from-openstreetmap-with-python-and-the-overpass-api-513882a27fd0
|
|
# https://geo.api.gouv.fr/adresse
|
|
# https://wiki.cartocite.fr/doku.php?id=umap:10_-_je_valorise_les_donnees_openstreetmap_avec_umap
|
|
# https://sites-formations.univ-rennes2.fr/mastersigat/Cours/Intro_Overpass.pdf
|
|
|
|
# usage des tags : https://taginfo.openstreetmap.org/tags/?key=amenity&value=bicycle_parking#combinations
|
|
|
|
# exemple URL données pour umap : https://www.velocite63.fr/velocite63/OSM/stationnements_velos_publics.json
|
|
# penser à cocher "proxy" dans la rubrique "données distantes" du calque
|
|
|
|
# export ODS :
|
|
# https://pythonhosted.org/pyexcel-ods/
|
|
# pip3 install pyexcel-ods3
|
|
|
|
import requests
|
|
import json
|
|
import time
|
|
from pyexcel_ods3 import save_data
|
|
from collections import OrderedDict
|
|
import os
|
|
from osm_vc63 import errors
|
|
from osm_vc63 import requetes
|
|
from osm_vc63.save import Save
|
|
|
|
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
|
|
|
|
|
|
# id du département "Puy de Dôme" : 7406
|
|
# id Riom : 1693144
|
|
# id Clermont : 110866
|
|
# id Romagnat : 138269
|
|
|
|
# l'id de l'area se calcule en ajoutant 3600000000 au numéro de l'objet OSM
|
|
aire_de_recherche = str(3600000000+110866)
|
|
|
|
|
|
|
|
# ----------------------------------------------
|
|
|
|
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 run_overpass_query(critere, aire_de_recherche) :
|
|
"""Envoie la requête Overpass et retourne la réponse JSON."""
|
|
|
|
overpass_query = """[out:json];
|
|
(
|
|
"""+critere+"""
|
|
);
|
|
out center;
|
|
"""
|
|
overpass_query = overpass_query.replace("aire_de_recherche", aire_de_recherche)
|
|
|
|
print("Execution requete overpass : \n" + overpass_query)
|
|
response = requests.get(overpass_url, params={'data': overpass_query})
|
|
|
|
if (response.status_code != 200) :
|
|
raise errors.Overpass_error(response.status_code)
|
|
|
|
return (response.json())
|
|
|
|
|
|
def run_reverse_geocoding(lat, lon) :
|
|
"""Retourne une adresse JSON à partir d'une position GPS."""
|
|
|
|
url = geo_api_url + "/reverse/"
|
|
|
|
response = requests.get(url, params={'lon' : str(lon), 'lat' : str(lat)})
|
|
|
|
if (response.status_code != 200) :
|
|
raise errors.Geo_api_error(response.status_code)
|
|
|
|
return (response.json())
|
|
|
|
|
|
def executer_requete_et_exporter_resultats(nom_req, critere, aire_de_recherche, overpass_query_fields) :
|
|
|
|
data = run_overpass_query(critere, aire_de_recherche)
|
|
|
|
nb_elements = len(data["elements"])
|
|
|
|
print("Nombre d'elements : "+str(nb_elements))
|
|
"""
|
|
print("Géocodage inversé : ", end="", flush=True)
|
|
|
|
# @TODO : optimiser en faisant un appel au service /reverse/csv/ plutot que le service unitaire /reverse/
|
|
|
|
for element in data["elements"]:
|
|
|
|
if (element["type"] == "node") :
|
|
rev_geocode = run_reverse_geocoding(element["lat"], element["lon"])
|
|
else :
|
|
rev_geocode = run_reverse_geocoding(element["center"]["lat"], element["center"]["lon"])
|
|
|
|
api_adresse = rev_geocode["features"][0]
|
|
|
|
element["tags"]["api_adresse:geometry:coordinates:lon"] = api_adresse["geometry"]["coordinates"][0]
|
|
element["tags"]["api_adresse:geometry:coordinates:lat"] = api_adresse["geometry"]["coordinates"][1]
|
|
|
|
element["tags"]["api_adresse:properties:label"] = api_adresse["properties"]["label"]
|
|
element["tags"]["api_adresse:properties:score"] = api_adresse["properties"]["score"]
|
|
|
|
if ("housenumber" in api_adresse["properties"]) :
|
|
element["tags"]["api_adresse:properties:housenumber"] = api_adresse["properties"]["housenumber"]
|
|
|
|
element["tags"]["api_adresse:properties:type"] = api_adresse["properties"]["type"]
|
|
|
|
element["tags"]["api_adresse:properties:name"] = api_adresse["properties"]["name"]
|
|
element["tags"]["api_adresse:properties:postcode"] = api_adresse["properties"]["postcode"]
|
|
element["tags"]["api_adresse:properties:citycode"] = api_adresse["properties"]["citycode"]
|
|
element["tags"]["api_adresse:properties:city"] = api_adresse["properties"]["city"]
|
|
|
|
if ("street" in api_adresse["properties"]) :
|
|
element["tags"]["api_adresse:properties:street"] = api_adresse["properties"]["street"]
|
|
|
|
element["tags"]["api_adresse:properties:attribution"] = rev_geocode["attribution"]
|
|
element["tags"]["api_adresse:properties:licence"] = rev_geocode["licence"]
|
|
|
|
|
|
# traduction
|
|
if "bicycle_parking" in element["tags"]:
|
|
element["tags"]["bicycle_parking"] = trad_bicycle_parking[element["tags"]["bicycle_parking"]]
|
|
|
|
print("X", end="", flush=True)
|
|
|
|
#else :
|
|
# print("-", end="", flush=True)
|
|
|
|
print()
|
|
"""
|
|
|
|
export_json = Save().nettoyage_json_pour_umap(data, overpass_query_fields)
|
|
|
|
# Sauvegarde
|
|
os.makedirs(dossier_sauvegarde, exist_ok = True)
|
|
|
|
Save().as_json(export_json, dossier_sauvegarde, nom_req)
|
|
Save().as_ods(overpass_query_fields, data, dossier_sauvegarde, nom_req)
|
|
|
|
|
|
|
|
def main():
|
|
for req in requetes.reqs:
|
|
|
|
for nb_essai in range(max_retry): # on tente max_retry fois
|
|
try:
|
|
executer_requete_et_exporter_resultats(
|
|
req.nom, req.critere, aire_de_recherche, req.champs
|
|
)
|
|
break
|
|
except errors.Api_error:
|
|
|
|
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("Fini")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|
|
|