#!/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 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 # 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+7406) # fields api_adresse (issus du géocodage inversé) api_adresse_fields = { "api_adresse:geometry:coordinates:lon" : {"export_json" : "Non", "FR" : "lon_adresse_etalab"}, "api_adresse:geometry:coordinates:lat" : {"export_json" : "Non", "FR" : "lat_adresse_etalab"}, "api_adresse:properties:label" : {"export_json" : "Non", "FR" : "adresse_etalab"}, "api_adresse:properties:score" : {"export_json" : "Non", "FR" : "score_etalab"}, "api_adresse:properties:housenumber" : {"export_json" : "Non", "FR" : "numero_etalab"}, "api_adresse:properties:type" : {"export_json" : "Non", "FR" : "type_etalab"}, "api_adresse:properties:name" : {"export_json" : "Non", "FR" : "numero_et_voie_etalab"}, "api_adresse:properties:postcode" : {"export_json" : "Non", "FR" : "code_postal_etalab"}, "api_adresse:properties:citycode" : {"export_json" : "Non", "FR" : "code_INSEE_etalab"}, "api_adresse:properties:city" : {"export_json" : "Non", "FR" : "ville_etalab"}, "api_adresse:properties:street" : {"export_json" : "Non", "FR" : "rue_etalab"}, } requetes_overpass = { "stationnements_velos_publics" : { #"critere" : '["amenity"="bicycle_parking"]["access"~"(yes|public)"]', "critere" : """nwr["amenity"="bicycle_parking"](area:"""+aire_de_recherche+"""); - nwr["amenity"="bicycle_parking"]["access"~"(no|permit|private|customers)"](area:"""+aire_de_recherche+");", "fields" : { "amenity" : {"export_json" : "Non", "FR" : "aménagement"} , "capacity" : {"export_json" : "Oui", "FR" : "nombre d'emplacements"}, "access" : {"export_json" : "Oui", "FR" : "accès"}, "amenity" : {"export_json" : "Non", "FR" : "aménagement"}, "bicycle_parking" : {"export_json" : "Oui", "FR" : "type"}, "covered" : {"export_json" : "Oui", "FR" : "couvert"}, "operator" : {"export_json" : "Oui", "FR" : "opérateur"}, "operator:type" : {"export_json" : "Oui", "FR" : "type d'opérateur"}, "fee" : {"export_json" : "Oui", "FR" : "frais"}, "check_date:capacity" : {"export_json" : "Non", "FR" : "date_vérification"}, "source" : {"export_json" : "Non", "FR" : "source"} } }, "stationnements_velos_non_publics" : { # "critere" : '["amenity"="bicycle_parking"]["access"~"(no|permit|private|customers|permissive)"]', "critere" : """nwr["amenity"="bicycle_parking"]["access"~"(no|permit|private|customers)"](area:"""+aire_de_recherche+"); ", "fields" : { "amenity" : {"export_json" : "Non", "FR" : "aménagement"} , "capacity" : {"export_json" : "Oui", "FR" : "nombre d'emplacements"}, "access" : {"export_json" : "Oui", "FR" : "accès"}, "amenity" : {"export_json" : "Non", "FR" : "aménagement"}, "bicycle_parking" : {"export_json" : "Oui", "FR" : "type"}, "covered" : {"export_json" : "Oui", "FR" : "couvert"}, "operator" : {"export_json" : "Oui", "FR" : "opérateur"}, "operator:type" : {"export_json" : "Oui", "FR" : "type d'opérateur"}, "fee" : {"export_json" : "Oui", "FR" : "frais"}, "check_date:capacity" : {"export_json" : "Non", "FR" : "date_vérification"}, "source" : {"export_json" : "Non", "FR" : "source"} } }, "ateliers_autoreparation" : { # "critere" : '["service:bicycle:diy"="yes"]', "critere" : """nwr["service:bicycle:diy"="yes"](area:"""+aire_de_recherche+"); ", "fields" : { "service:bicycle:diy" : {"export_json" : "Non", "FR" : ""}, "name" : {"export_json" : "Oui", "FR" : ""}, "description" : {"export_json" : "Oui", "FR" : ""}, "website" : {"export_json" : "Oui", "FR" : ""}, "addr:housenumber" : {"export_json" : "Oui", "FR" : ""}, "addr:street" : {"export_json" : "Oui", "FR" : ""}, "addr:postcode" : {"export_json" : "Oui", "FR" : ""}, "addr:city" : {"export_json" : "Oui", "FR" : ""}, "contact:email" : {"export_json" : "Oui", "FR" : "email"}, "contact:twitter" : {"export_json" : "Oui", "FR" : "Twitter"}, "contact:facebook" : {"export_json" : "Oui", "FR" : "Facebook"}, "contact:phone" : {"export_json" : "Oui", "FR" : "Téléphone"}, "network" : {"export_json" : "Oui", "FR" : "Réseau"}, "office" : {"export_json" : "Oui", "FR" : "Bureau"}, "opening_hours" : {"export_json" : "Oui", "FR" : "Horaires"} } }, "associations_velo" : { #"critere" : '["association"="bicycle"]', "critere" : """nwr["association"="bicycle"](area:"""+aire_de_recherche+"); ", "fields" : { "association" : {"export_json" : "Non", "FR" : ""}, "name" : {"export_json" : "Oui", "FR" : ""}, "description" : {"export_json" : "Oui", "FR" : ""}, "website" : {"export_json" : "Oui", "FR" : ""}, "addr:housenumber" : {"export_json" : "Oui", "FR" : ""}, "addr:street" : {"export_json" : "Oui", "FR" : ""}, "addr:postcode" : {"export_json" : "Oui", "FR" : ""}, "addr:city" : {"export_json" : "Oui", "FR" : ""}, "contact:email" : {"export_json" : "Oui", "FR" : "email"}, "contact:twitter" : {"export_json" : "Oui", "FR" : "Twitter"}, "contact:facebook" : {"export_json" : "Oui", "FR" : "Facebook"}, "contact:phone" : {"export_json" : "Oui", "FR" : "Téléphone"}, "network" : {"export_json" : "Oui", "FR" : "Réseau"}, "office" : {"export_json" : "Oui", "FR" : "Bureau"}, "opening_hours" : {"export_json" : "Oui", "FR" : "Horaires"} } }, "fabriquants_velo" : { #"critere" : '["craft"="bicycle"]', "critere" : """nwr["craft"="bicycle"](area:"""+aire_de_recherche+"); ", "fields" : { "craft" : {"export_json" : "Non", "FR" : ""}, "name" : {"export_json" : "Oui", "FR" : ""}, "description" : {"export_json" : "Oui", "FR" : ""}, "website" : {"export_json" : "Oui", "FR" : ""}, "addr:housenumber" : {"export_json" : "Oui", "FR" : ""}, "addr:street" : {"export_json" : "Oui", "FR" : ""}, "addr:postcode" : {"export_json" : "Oui", "FR" : ""}, "addr:city" : {"export_json" : "Oui", "FR" : ""}, "contact:email" : {"export_json" : "Oui", "FR" : "email"}, "contact:twitter" : {"export_json" : "Oui", "FR" : "Twitter"}, "contact:facebook" : {"export_json" : "Oui", "FR" : "Facebook"}, "contact:phone" : {"export_json" : "Oui", "FR" : "Téléphone"}, "network" : {"export_json" : "Oui", "FR" : "Réseau"}, "office" : {"export_json" : "Oui", "FR" : "Bureau"}, "opening_hours" : {"export_json" : "Oui", "FR" : "Horaires"} } }, "vendeurs_velo" : { #"critere" : '["shop"="bicycle"]', "critere" : """nwr["shop"="bicycle"](area:"""+aire_de_recherche+"); "+ """nwr["service:bicycle:retail"="yes"](area:"""+aire_de_recherche+"); ", "fields" : { "shop" : {"export_json" : "Non", "FR" : ""}, "name" : {"export_json" : "Oui", "FR" : ""}, "description" : {"export_json" : "Oui", "FR" : ""}, "website" : {"export_json" : "Oui", "FR" : ""}, "addr:housenumber" : {"export_json" : "Oui", "FR" : ""}, "addr:street" : {"export_json" : "Oui", "FR" : ""}, "addr:postcode" : {"export_json" : "Oui", "FR" : ""}, "addr:city" : {"export_json" : "Oui", "FR" : ""}, "contact:email" : {"export_json" : "Oui", "FR" : "email"}, "contact:twitter" : {"export_json" : "Oui", "FR" : "Twitter"}, "contact:facebook" : {"export_json" : "Oui", "FR" : "Facebook"}, "contact:phone" : {"export_json" : "Oui", "FR" : "Téléphone"}, "network" : {"export_json" : "Oui", "FR" : "Réseau"}, "office" : {"export_json" : "Oui", "FR" : "Bureau"}, "opening_hours" : {"export_json" : "Oui", "FR" : "Horaires"} } }, "velos_libre_service" : { #"critere" : '["amenity"="bicycle_rental"]', "critere" : """nwr["amenity"="bicycle_rental"](area:"""+aire_de_recherche+"); ", "fields" : { "amenity" : {"export_json" : "Non", "FR" : ""}, "name" : {"export_json" : "Oui", "FR" : ""}, "description" : {"export_json" : "Oui", "FR" : ""}, "website" : {"export_json" : "Oui", "FR" : ""}, "addr:housenumber" : {"export_json" : "Oui", "FR" : ""}, "addr:street" : {"export_json" : "Oui", "FR" : ""}, "addr:postcode" : {"export_json" : "Oui", "FR" : ""}, "addr:city" : {"export_json" : "Oui", "FR" : ""}, "contact:email" : {"export_json" : "Oui", "FR" : "email"}, "contact:twitter" : {"export_json" : "Oui", "FR" : "Twitter"}, "contact:facebook" : {"export_json" : "Oui", "FR" : "Facebook"}, "contact:phone" : {"export_json" : "Oui", "FR" : "Téléphone"}, "network" : {"export_json" : "Oui", "FR" : "Réseau"}, "office" : {"export_json" : "Oui", "FR" : "Bureau"}, "opening_hours" : {"export_json" : "Oui", "FR" : "Horaires"} } }, "location_velo" : { #"critere" : '["service:bicycle:rental"="yes"]', "critere" : """nwr["service:bicycle:rental"="yes"](area:"""+aire_de_recherche+"); ", "fields" : { "service:bicycle:rental" : {"export_json" : "Non", "FR" : ""}, "name" : {"export_json" : "Oui", "FR" : ""}, "description" : {"export_json" : "Oui", "FR" : ""}, "website" : {"export_json" : "Oui", "FR" : ""}, "addr:housenumber" : {"export_json" : "Oui", "FR" : ""}, "addr:street" : {"export_json" : "Oui", "FR" : ""}, "addr:postcode" : {"export_json" : "Oui", "FR" : ""}, "addr:city" : {"export_json" : "Oui", "FR" : ""}, "contact:email" : {"export_json" : "Oui", "FR" : "email"}, "contact:twitter" : {"export_json" : "Oui", "FR" : "Twitter"}, "contact:facebook" : {"export_json" : "Oui", "FR" : "Facebook"}, "contact:phone" : {"export_json" : "Oui", "FR" : "Téléphone"}, "network" : {"export_json" : "Oui", "FR" : "Réseau"}, "office" : {"export_json" : "Oui", "FR" : "Bureau"}, "opening_hours" : {"export_json" : "Oui", "FR" : "Horaires"} } } } # ---------------------------------------------- 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"} # ---------------------------------------------- class Api_error(Exception) : def __init__(self, http_code, message="erreur appel API"): self.http_code = http_code self.message = message super().__init__(self.message) def __str__(self): return f'{self.http_code} -> {self.message}' class Overpass_error(Api_error) : pass class Geo_api_error(Api_error) : pass # ---------------------------------------------- def run_overpass_query(query) : response = requests.get(overpass_url, params={'data': query}) if (response.status_code != 200) : raise Overpass_error(response.status_code) return (response.json()) def run_reverse_geocoding(lat, lon) : url = geo_api_url + "/reverse/" response = requests.get(url, params={'lon' : str(lon), 'lat' : str(lat)}) if (response.status_code != 200) : raise Geo_api_error(response.status_code) return (response.json()) # ---------------------------------------------- def executer_requete_et_exporter_resultats(nom_req, critere, aire_de_recherche, overpass_query_fields) : print ("Nom requête : "+nom_req) overpass_query = """[out:json]; ( """+critere+""" ); out center; """ overpass_query_fields.update(api_adresse_fields) print("Execution requete overpass : \n"+overpass_query) data = run_overpass_query(overpass_query) 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() print("Sauvegarde résultat format JSON/OSM") export_json = {"version": data["version"], "generator" : data["generator"] + " and ETALAB API", "osm3s" : data["osm3s"], "elements": [] } index_line = 0 # on refait un JSON allégé juste avec les données qu'on va afficher sur la carte UMAP for element in data["elements"]: export_json["elements"].append({"type" : element["type"], "id" : element["id"]}) if (element["type"] == "node") : export_json["elements"][index_line]["lat"] = element["lat"] export_json["elements"][index_line]["lon"] = element["lon"] else : export_json["elements"][index_line]["center"] = element["center"] export_json["elements"][index_line]["nodes"] = element["nodes"] #export_json["elements"][index_line]["tags"] = element["tags"] description = "" for tag in overpass_query_fields.keys() : if overpass_query_fields[tag]["export_json"] == "Oui" : if tag in element["tags"] : if overpass_query_fields[tag]["FR"] != "" : description = description + overpass_query_fields[tag]["FR"] + " : " description = description + str(element["tags"][tag]) + "\n" export_json["elements"][index_line]["tags"] = {"description": description} index_line = index_line + 1 # print (json.dumps(export_json)) jsonFile = open(nom_req+".json", "w") jsonFile.write(json.dumps(export_json)) jsonFile.close() # =========================================== print("Sauvegarde résultats format ODS") ODSdataSheet = OrderedDict() ODSdata = [] ODSdata.append(overpass_query_fields.keys()) index_line = 2 for element in data["elements"]: line = [] index_col = 0 # if (element["type"] == "node") : for field in overpass_query_fields.keys() : if (field in element["tags"]) : if field == "capacity": val = element["tags"][field] line.append(int(val) if val.isdigit() else val) else : line.append(element["tags"][field]) else : line.append("") index_col = index_col + 1 ODSdata.append(line) index_line = index_line + 1 ODSdataSheet.update({"resultats": ODSdata}) save_data(nom_req+".ods", ODSdataSheet) for nom_requete in requetes_overpass.keys() : for nb_essai in range(max_retry) : # on tente max_retry fois try : executer_requete_et_exporter_resultats(nom_requete, requetes_overpass[nom_requete]["critere"], aire_de_recherche, requetes_overpass[nom_requete]["fields"]) break except 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")