Olav63
/
outils_OSM
2
1
Fork 1
Code Issues 1 Pull Requests Projects Releases Wiki Activity

fix #1 ajout de logs

This commit is contained in:
SebF 2021-10-23 19:48:52 +02:00
parent b4b395394f
commit 81292d3a32
3 changed files with 31 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
osm_vc63/errors.py
View File

@ -19,17 +19,16 @@
"""Errors module"""
import logging
class ApiError(Exception):
"""Api exception"""
def __init__(self, http_code, message="erreur appel API"):
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}"
super().__init__(message)
logging.error(message)
class OverpassError(ApiError):

50
osm_vc63/utils.py
View File

@ -20,6 +20,7 @@
"""Collections de méthodes utilitaires"""
import json
import logging
from collections import OrderedDict
import requests
from pyexcel_ods3 import save_data
@ -68,7 +69,7 @@ class Utils:
save_data(self.dossier_sauvegarde + nom_req + ".ods", ods_data_sheet)
print("Sauvegarde résultats format ODS")
logging.info("Sauvegarde résultats format ODS")
def save_as_json(self, export_json, nom_req):
"""Enregistrement du JSON"""
@ -77,7 +78,7 @@ class Utils:
json_file.write(json.dumps(export_json))
json_file.close()
print("Sauvegarde résultat format JSON/OSM")
logging.info("Sauvegarde résultat format JSON/OSM")
def nettoyage_json_pour_umap(self, data, overpass_query_fields):
"""Sélection uniquement des champs export_json == oui"""
@ -136,7 +137,6 @@ class Utils:
)
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:
@ -159,7 +159,7 @@ class Utils:
def geocodage(self, data):
"""Renseigne une adresse pour chaque élément de data"""
for element in self.progress_bar(data["elements"], prefix="Géocodage"):
for element in data["elements"]:
if element["type"] == "node":
rev_geocode = self.run_reverse_geocoding(element["lat"], element["lon"])
@ -216,6 +216,8 @@ class Utils:
]
element["tags"]["api_adresse:properties:licence"] = rev_geocode["licence"]
logging.info("Géocodage inversé terminé")
return data
def traduction(self, tag, dictionnaire, data):
@ -226,43 +228,3 @@ class Utils:
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()

26
recup_donnees_osm_overpass.py
View File

@ -29,6 +29,8 @@ Module principal : 
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
@ -80,7 +82,7 @@ def init_argparse() -> argparse.ArgumentParser:
"-z",
"--zone",
type=int,
help="Choisir la zone géographique à inspecter. ",
help="Choisir la zone géographique à inspecter.",
default=7406,
)
@ -100,6 +102,10 @@ def init_argparse() -> argparse.ArgumentParser:
)
parser.set_defaults(geocoding_inverse=True)
parser.add_argument(
"-l", "--log-level", type=str, help="Définir le niveau de log.", default="ERROR"
)
return parser
@ -109,6 +115,14 @@ def main():
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("debug.log"), logging.StreamHandler(sys.stdout)],
)
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)
@ -117,12 +131,12 @@ def main():
try:
utils = Utils(OVERPASS_URL, GEO_API_URL, DOSSIER_SAUVEGARDE)
print(f"{75*'#'}\r\nRequête en cours : {req.nom}")
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"])
print(f"{nb_resultats} résultats")
logging.info(f"{nb_resultats} résultats")
if nb_resultats > 0:
if args.geocoding_inverse:
@ -147,14 +161,14 @@ def main():
except errors.ApiError:
if nb_essai == MAX_RETRY:
print("trop d'erreurs d'API - abandon")
logging.error("Trop d'erreurs d'API - abandon")
exit()
print("erreur API - on retente dans " + str(RETRY_DELAY) + "s")
logging.error(f"Erreur API - on retente dans {RETRY_DELAY}s")
time.sleep(RETRY_DELAY)
print("\r\n ### Terminé ###")
logging.info("##### Terminé #####")
if __name__ == "__main__":