#!/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/>. """Collections de méthodes utilitaires""" import json import logging import os import datetime import shutil import pathlib 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_resultats: str def __init__(self, overpass_url, geo_api_url, dossier_resultats): self.overpass_url = overpass_url self.geo_api_url = geo_api_url self.dossier_resultats = dossier_resultats def save_as_ods(self, fields, data, nom_req, ods_data_sheet=OrderedDict()): """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({f"{nom_req}": ods_data}) save_data(self.dossier_resultats + "resultats.ods", ods_data_sheet) logging.info("Sauvegarde résultats format ODS") def save_as_json(self, export_json, nom_req): """Enregistrement du JSON""" json_file = open(self.dossier_resultats + nom_req + ".json", "w") json_file.write(json.dumps(export_json)) json_file.close() 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""" 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) 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() def geocodage(self, data): """Renseigne une adresse pour chaque élément de data""" for element in data["elements"]: 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"] logging.info("Géocodage inversé terminé") 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 archivage(self, dossier_archive): """Archivage des données précédentes""" fichier = pathlib.Path(self.dossier_resultats + "resultats.ods") if not fichier.exists(): return date_fichier = datetime.date.fromtimestamp(fichier.stat().st_ctime) os.makedirs(dossier_archive + str(date_fichier), exist_ok=True) # pylint: disable=W0106 [ shutil.move( self.dossier_resultats + file, dossier_archive + str(date_fichier) ) for file in os.listdir(self.dossier_resultats) if not os.path.isdir(self.dossier_resultats + file) ]