Skip to content

Commit

Permalink
Fetch brand & version
Browse files Browse the repository at this point in the history
  • Loading branch information
raphodn committed Dec 3, 2024
1 parent ee0af37 commit c93c460
Showing 1 changed file with 40 additions and 7 deletions.
47 changes: 40 additions & 7 deletions open_prices/common/openstreetmap.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
from OSMPythonTools.api import Api
from OSMPythonTools.nominatim import Nominatim

OSM_FIELDS = ["name", "display_name", "lat", "lon"]
OSM_FIELDS_FROM_NOMINATIM = ["name", "display_name", "lat", "lon"]
OSM_FIELDS_FROM_OPENSTREETMAP = ["brand", "version"]
OSM_TAG_FIELDS_MAPPING = {"class": "tag_key", "type": "tag_value"}
OSM_ADDRESS_FIELDS = [
"postcode",
Expand All @@ -11,20 +13,37 @@
OSM_ADDRESS_PLACE_FIELDS = ["village", "town", "city", "municipality"]


def get_location(osm_id: int, osm_type: str) -> list:
def get_location_from_nominatim(osm_id: int, osm_type: str) -> list:
client = Nominatim()
search_query = f"{osm_type}/{osm_id}"
search_query = f"{osm_type.lower()}/{osm_id}"
return client.query(search_query, lookup=True).toJSON()


def get_location_from_openstreetmap(osm_id: int, osm_type: str) -> dict:
api = Api()
response = api.query(f"{osm_type.lower()}/{osm_id}")
return {
"name": response.tag("name"),
# "tag_key": "",
# "tag_value": "",
"brand": response.tag("brand"),
"brand_wikidata": response.tag("brand:wikidata"),
"brand_wikipedia": response.tag("brand:wikipedia"),
"lat": response.lat(),
"lon": response.lon(),
"version": response.version(),
}


def get_location_dict(location):
location_dict = dict()
# fetch data from Nominatim
try:
response = get_location(
response = get_location_from_nominatim(
osm_id=location.osm_id, osm_type=location.osm_type.lower()
)
if len(response):
for osm_field in OSM_FIELDS:
for osm_field in OSM_FIELDS_FROM_NOMINATIM:
if osm_field in response[0]:
key = f"osm_{osm_field}"
value = response[0][osm_field]
Expand All @@ -50,7 +69,21 @@ def get_location_dict(location):
key = "osm_address_city"
value = response[0]["address"][osm_address_place_field]
location_dict[key] = value
return location_dict
except Exception:
# logger.exception("Error returned from OpenStreetMap")
return None
pass
# fetch extra data from OpenStreetMap
try:
response = get_location_from_openstreetmap(
osm_id=location.osm_id, osm_type=location.osm_type.lower()
)
if response:
for osm_field in OSM_FIELDS_FROM_OPENSTREETMAP:
if osm_field in response:
key = f"osm_{osm_field}"
value = response[osm_field]
location_dict[key] = value
except Exception:
pass
# return
return location_dict

0 comments on commit c93c460

Please sign in to comment.