SebF/RDOO
1
0
Fork 0
forked from Olav63/outils_OSM
Code Issues Pull Requests Projects Releases Wiki Activity

Merge branch '#8_Optimisation_geocodage_inverse'

Browse Source
This commit is contained in:
SebF 2021-11-06 18:56:26 +01:00
commit 7b4d4cf711
2 changed files with 62 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

108
osm_vc63/utils.py
View File

@ -25,6 +25,7 @@ import os
import datetime import datetime
import shutil import shutil
import pathlib import pathlib
import csv
from collections import OrderedDict from collections import OrderedDict
import requests import requests
from pyexcel_ods3 import save_data from pyexcel_ods3 import save_data
@ -148,77 +149,74 @@ class Utils:
return response.json() return response.json()
def run_reverse_geocoding(self, lat, lon): def geocodage_csv(self, data):
"""Retourne une adresse JSON à partir d'une position GPS.""" """
Renseigne une adresse pour chaque élément de data
en une fois via csv
"""
url = self.geo_api_url + "/reverse/" url = self.geo_api_url + "/reverse/csv/"
response = requests.get(url, params={"lon": str(lon), "lat": str(lat)}) # création du fichier à envoyer à l'API
with open("tmp_geocodage.csv", "w", newline="") as tmp_csv_file:
csv_writer = csv.writer(tmp_csv_file)
csv_writer.writerow(["lat", "lon"])
for element in data["elements"]:
if element["type"] == "node":
csv_writer.writerow([element["lat"], element["lon"]])
else:
csv_writer.writerow(
[element["center"]["lat"], element["center"]["lon"]]
)
# préparation et envoi de la requête
payload = dict(
[("data", ("tmp_geocodage.csv", open("tmp_geocodage.csv", "rb").read()))]
)
response = requests.post(url, files=payload)
# nettoyage
os.remove("tmp_geocodage.csv")
if response.status_code != 200: if response.status_code != 200:
raise errors.GeoApiError(response.status_code) raise errors.GeoApiError(response.status_code)
return response.json() # affectation des addresses
def geocodage(self, data):
"""Renseigne une adresse pour chaque élément de data"""
for element in data["elements"]: for element in data["elements"]:
for row in csv.DictReader(response.text.splitlines()):
if element["type"] == "node": if element["type"] == "node":
rev_geocode = self.run_reverse_geocoding(element["lat"], element["lon"]) lat_ok = row["lat"] == str(element["lat"])
lon_ok = row["lon"] == str(element["lon"])
else: else:
rev_geocode = self.run_reverse_geocoding( lat_ok = row["lat"] == str(element["center"]["lat"])
element["center"]["lat"], element["center"]["lon"] lon_ok = row["lon"] == str(element["center"]["lon"])
)
api_adresse = rev_geocode["features"][0] if lat_ok and lon_ok:
element["tags"]["api_adresse:geometry:coordinates:lon"] = row[
element["tags"]["api_adresse:geometry:coordinates:lon"] = api_adresse[ "result_longitude"
"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"][ element["tags"]["api_adresse:geometry:coordinates:lat"] = row[
"score" "result_latitude"
] ]
element["tags"]["api_adresse:properties:label"] = row[
if "housenumber" in api_adresse["properties"]: "result_label"
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:housenumber"] = row[
element["tags"]["api_adresse:properties:name"] = api_adresse["properties"][ "result_housenumber"
"name"
] ]
element["tags"]["api_adresse:properties:postcode"] = api_adresse[ element["tags"]["api_adresse:properties:type"] = row["result_type"]
"properties" element["tags"]["api_adresse:properties:name"] = row["result_name"]
]["postcode"] element["tags"]["api_adresse:properties:postcode"] = row[
element["tags"]["api_adresse:properties:citycode"] = api_adresse[ "result_postcode"
"properties"
]["citycode"]
element["tags"]["api_adresse:properties:city"] = api_adresse["properties"][
"city"
] ]
element["tags"]["api_adresse:properties:citycode"] = row[
if "street" in api_adresse["properties"]: "result_citycode"
element["tags"]["api_adresse:properties:street"] = api_adresse[ ]
"properties" element["tags"]["api_adresse:properties:city"] = row["result_city"]
]["street"] element["tags"]["api_adresse:properties:street"] = row[
"result_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é") logging.info("Géocodage inversé terminé")

2
rdoo.py
View File

@ -169,7 +169,7 @@ def main():
if nb_resultats > 0: if nb_resultats > 0:
if args.geocoding_inverse: if args.geocoding_inverse:
# géocodage inverse # géocodage inverse
data = utils.geocodage(data) data = utils.geocodage_csv(data)
# traduction # traduction
data = utils.traduction( data = utils.traduction(