diff --git a/osm_vc63/errors.py b/osm_vc63/errors.py new file mode 100644 index 0000000..07f40f7 --- /dev/null +++ b/osm_vc63/errors.py @@ -0,0 +1,26 @@ +#!/usr/bin/env python3 +"""Errors module""" + + +class ApiError(Exception): + """Api 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 OverpassError(ApiError): + """Overpass exception""" + + pass + + +class GeoApiError(ApiError): + """GeoApi exception""" + + pass diff --git a/osm_vc63/requetes.py b/osm_vc63/requetes.py new file mode 100644 index 0000000..667cf8d --- /dev/null +++ b/osm_vc63/requetes.py @@ -0,0 +1,147 @@ +#!/usr/bin/env python3 +"""Module des requêtes""" + + +class Requete: + """Objet requête""" + + nom: str + critere: str + champs: dict + + def __init__(self, nom, critere, champs): + self.nom = nom + self.critere = critere + self.champs = champs + + +REQS = [] +CHAMPS_STATIONNEMENT = { + "amenity": {"export_json": "Non", "FR": "aménagement"}, + "capacity": {"export_json": "Oui", "FR": "nombre d'emplacements"}, + "access": {"export_json": "Oui", "FR": "accès"}, + "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"}, +} +CHAMPS_POI = { + "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"}, +} +CHAMPS_ADRESSE = { + "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"}, +} + +REQS.append( + Requete( + "stationnements_velos_publics", + # pylint: disable=C0301 + r'nwr["amenity"="bicycle_parking"](area:aire_de_recherche); - nwr["amenity"="bicycle_parking"]["access"~"(no|permit|private|customers)"](area:aire_de_recherche);', + dict(CHAMPS_STATIONNEMENT, **CHAMPS_ADRESSE), + ) +) + +REQS.append( + Requete( + "stationnements_velos_non_publics", + # pylint: disable=C0301 + r'nwr["amenity"="bicycle_parking"]["access"~"(no|permit|private|customers)"](area:aire_de_recherche);', + dict(CHAMPS_STATIONNEMENT, **CHAMPS_ADRESSE), + ) +) + +CHAMP_LOCAL = {"service:bicycle:diy": {"export_json": "Non", "FR": ""}} +REQS.append( + Requete( + "ateliers_autoreparation", + r'nwr["service:bicycle:diy"="yes"](area:aire_de_recherche);', + dict(CHAMP_LOCAL, **CHAMPS_POI, **CHAMPS_ADRESSE), + ) +) + +CHAMP_LOCAL = {"association": {"export_json": "Non", "FR": ""}} +REQS.append( + Requete( + "associations_velo", + r'nwr["association"="bicycle"](area:aire_de_recherche);', + dict(CHAMP_LOCAL, **CHAMPS_POI, **CHAMPS_ADRESSE), + ) +) + +CHAMP_LOCAL = {"craft": {"export_json": "Non", "FR": ""}} +REQS.append( + Requete( + "fabriquants_velo", + r'nwr["craft"="bicycle"](area:aire_de_recherche);', + dict(CHAMP_LOCAL, **CHAMPS_POI, **CHAMPS_ADRESSE), + ) +) + +CHAMP_LOCAL = {"shop": {"export_json": "Non", "FR": ""}} +REQS.append( + Requete( + "vendeurs_velo", + # pylint: disable=C0301 + r'nwr["shop"="bicycle"](area:aire_de_recherche); nwr["service:bicycle:retail"="yes"](area:aire_de_recherche);', + dict(CHAMP_LOCAL, **CHAMPS_POI, **CHAMPS_ADRESSE), + ) +) + +CHAMP_LOCAL = {"amenity": {"export_json": "Non", "FR": ""}} +REQS.append( + Requete( + "velos_libre_service", + r'nwr["amenity"="bicycle_rental"](area:aire_de_recherche);', + dict(CHAMP_LOCAL, **CHAMPS_POI, **CHAMPS_ADRESSE), + ) +) + +CHAMP_LOCAL = {"service:bicycle:rental": {"export_json": "Non", "FR": ""}} +REQS.append( + Requete( + "location_velo", + r'nwr["service:bicycle:rental"="yes"](area:aire_de_recherche);', + dict(CHAMP_LOCAL, **CHAMPS_POI, **CHAMPS_ADRESSE), + ) +) diff --git a/osm_vc63/utils.py b/osm_vc63/utils.py new file mode 100644 index 0000000..e767aef --- /dev/null +++ b/osm_vc63/utils.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +"""Collections de méthodes utilitaires""" + +import json +from collections import OrderedDict +import requests +from pyexcel_ods3 import save_data +from osm_vc63 import errors + + +class Utils: + """Classe de méthodes utilitaires""" + + overpass_url: str + geo_api_url: str + dossier_sauvegarde: str + + def __init__(self, overpass_url, geo_api_url, dossier_sauvegarde): + self.overpass_url = overpass_url + self.geo_api_url = geo_api_url + self.dossier_sauvegarde = dossier_sauvegarde + + def save_as_ods(self, fields, data, nom_req): + """Sauvegarde de data dans un classeur ods""" + + ods_data_sheet = OrderedDict() + ods_data = [] + ods_data.append(fields.keys()) + index_line = 2 + + for element in data["elements"]: + line = [] + index_col = 0 + + for field in 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 + + ods_data.append(line) + index_line = index_line + 1 + + ods_data_sheet.update({"resultats": ods_data}) + + save_data(self.dossier_sauvegarde + nom_req + ".ods", ods_data_sheet) + + print("Sauvegarde résultats format ODS") + + def save_as_json(self, export_json, nom_req): + """Enregistrement du JSON""" + + json_file = open(self.dossier_sauvegarde + nom_req + ".json", "w") + json_file.write(json.dumps(export_json)) + json_file.close() + + print("Sauvegarde résultat format JSON/OSM") + + def nettoyage_json_pour_umap(self, data, overpass_query_fields): + """Sélection uniquement des champs export_json == oui""" + + export_json = { + "version": data["version"], + "generator": data["generator"] + " and ETALAB API", + "osm3s": data["osm3s"], + "elements": [], + } + + index_line = 0 + + for element in data["elements"]: + export_json["elements"].append( + {"type": element["type"], "id": element["id"]} + ) + + # positionnement des éléments + if element["type"] == "node": # noeuds + export_json["elements"][index_line]["lat"] = element["lat"] + export_json["elements"][index_line]["lon"] = element["lon"] + else: # ways et relations + export_json["elements"][index_line]["center"] = element["center"] + export_json["elements"][index_line]["nodes"] = element["nodes"] + + # filtrage des 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 + + return export_json + + def run_overpass_query(self, 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(self.overpass_url, params={"data": overpass_query}) + + if response.status_code != 200: + raise errors.OverpassError(response.status_code) + + return response.json() + + def run_reverse_geocoding(self, lat, lon): + """Retourne une adresse JSON à partir d'une position GPS.""" + + url = self.geo_api_url + "/reverse/" + + response = requests.get(url, params={"lon": str(lon), "lat": str(lat)}) + + if response.status_code != 200: + raise errors.GeoApiError(response.status_code) + + return response.json() + + # TODO : optimiser en faisant un appel au service /reverse/csv/ plutot que le service unitaire /reverse/ + def geocodage(self, data): + """Renseigne une adresse pour chaque élément de data""" + + for element in self.progress_bar(data["elements"], prefix="Géocodage"): + + if element["type"] == "node": + rev_geocode = self.run_reverse_geocoding(element["lat"], element["lon"]) + else: + rev_geocode = self.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"] + + return data + + def traduction(self, tag, dictionnaire, data): + """Traduit le champ tag des éléments de data avec dict""" + + for element in data["elements"]: + if tag in element["tags"]: + element["tags"][tag] = dictionnaire[element["tags"][tag]] + + return data + + def progress_bar( + # pylint:disable=C0330 + self, + iterable, + decimals=1, + length=50, + prefix="", + fill="█", + print_end="\r", + ): + """ + Call in a loop to create terminal progress bar + @params: + iterable - Required : iterable object (Iterable) + decimals - Optional : positive number of decimals in percent complete (Int) + length - Optional : character length of bar (Int) + prefix - Optional : prefix string (Str) + fill - Optional : bar fill character (Str) + print_end - Optional : end character (e.g. "\r", "\r\n") (Str) + """ + total = len(iterable) + + if total == 0: + return + + # Initial Call + print(f"\r{prefix} |{'-' * length}| {0}%", end=print_end) + # Update Progress Bar + for i, item in enumerate(iterable): + yield item + percent = ("{0:." + str(decimals) + "f}").format( + 100 * ((i + 1) / float(total)) + ) + filled = int(length * (i + 1) // total) + progress = fill * filled + "-" * (length - filled) + print(f"\r{prefix} |{progress}| {percent}%", end=print_end) + + # Print New Line on Complete + print() diff --git a/recup_donnees_osm_overpass.py b/recup_donnees_osm_overpass.py new file mode 100644 index 0000000..acb5ab6 --- /dev/null +++ b/recup_donnees_osm_overpass.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +""" +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 +""" + +# 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 time +import os +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 + + +# 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(3_600_000_000 + 110_866) + +# 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 main(): + """Routine principale""" + + 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: + # 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()