query from osm if not available in osmcha

Author: ElJocho

mapswipe_workers/mapswipe_workers/definitions.py

@@ -13,6 +13,7 @@
 LOGGING_FILE_PATH = os.path.join(DATA_PATH, "mapswipe_workers.log")
 
 OHSOME_API_LINK = "https://api.ohsome.org/v1/"
+OSM_API_LINK = "https://www.openstreetmap.org/api/0.6/"
 OSMCHA_API_LINK = "https://osmcha.org/api/v1/"
 OSMCHA_API_KEY = os.environ["OSMCHA_API_KEY"]
 

mapswipe_workers/mapswipe_workers/utils/api_calls.py

@@ -1,9 +1,12 @@
+from xml.etree import ElementTree
+
 import requests
 from requests.adapters import HTTPAdapter
 from requests.packages.urllib3.util.retry import Retry
 
 from mapswipe_workers.definitions import (
     OHSOME_API_LINK,
+    OSM_API_LINK,
     OSMCHA_API_KEY,
     OSMCHA_API_LINK,
     CustomError,
@@ -21,13 +24,16 @@ def remove_troublesome_chars(string: str):
     return string
 
 
-def retry_get(url, retries=3, timeout=4):
+def retry_get(url, retries=3, timeout=4, to_osmcha: bool = False):
     """Retry a query for a variable amount of tries."""
     retry = Retry(total=retries)
     with requests.Session() as session:
         session.mount("https://", HTTPAdapter(max_retries=retry))
-        headers = {"Authorization": OSMCHA_API_KEY}
-        return session.get(url, timeout=timeout, headers=headers)
+        if to_osmcha:
+            headers = {"Authorization": OSMCHA_API_KEY}
+            return session.get(url, timeout=timeout, headers=headers)
+        else:
+            return session.get(url, timeout=timeout)
 
 
 def geojsonToFeatureCollection(geojson: dict) -> dict:
@@ -54,18 +60,14 @@ def query_osmcha(changeset_ids: list, changeset_results):
     id_string = ",".join(map(str, changeset_ids))
 
     url = OSMCHA_API_LINK + f"changesets/?ids={id_string}"
-    logger.info(url)
-    logger.info(len(changeset_ids))
-    response = retry_get(url)
+    response = retry_get(url, to_osmcha=True)
     if response.status_code != 200:
         err = f"osmcha request failed: {response.status_code}"
         logger.warning(f"{err}")
         logger.warning(response.json())
         raise CustomError(err)
     response = response.json()
-    logger.info(response)
     for feature in response["features"]:
-        logger.info(feature)
         changeset_results[int(feature["id"])] = {
             "username": remove_troublesome_chars(feature["properties"]["user"]),
             "userid": feature["properties"]["uid"],
@@ -76,9 +78,45 @@ def query_osmcha(changeset_ids: list, changeset_results):
     return changeset_results
 
 
+def query_osm(changeset_ids: list, changeset_results):
+    """Get data from changesetId."""
+    id_string = ",".join(map(str, changeset_ids))
+
+    url = OSM_API_LINK + f"changesets?changesets={id_string}"
+    response = retry_get(url)
+    if response.status_code != 200:
+        err = f"osm request failed: {response.status_code}"
+        logger.warning(f"{err}")
+        logger.warning(response.json())
+        raise CustomError(err)
+    tree = ElementTree.fromstring(response.content)
+
+    for changeset in tree.iter("changeset"):
+        id = changeset.attrib["id"]
+        username = remove_troublesome_chars(changeset.attrib["user"])
+        userid = changeset.attrib["uid"]
+        comment = created_by = None
+        for tag in changeset.iter("tag"):
+            if tag.attrib["k"] == "comment":
+                comment = tag.attrib["v"]
+            if tag.attrib["k"] == "created_by":
+                created_by = tag.attrib["v"]
+
+        changeset_results[int(id)] = {
+            "username": remove_troublesome_chars(username),
+            "userid": userid,
+            "comment": remove_troublesome_chars(comment),
+            "editor": remove_troublesome_chars(created_by),
+        }
+    return changeset_results
+
+
 def remove_noise_and_add_user_info(json: dict) -> dict:
     """Delete unwanted information from properties."""
     logger.info("starting filtering and adding extra info")
+    batch_size = 100
+
+    # remove noise
     changeset_results = {}
 
     missing_rows = {
@@ -100,20 +138,32 @@ def remove_noise_and_add_user_info(json: dict) -> dict:
         changeset_results[new_properties["changesetId"]] = None
         feature["properties"] = new_properties
 
+    # add info
     len_osm = len(changeset_results.keys())
-    batches = int(len(changeset_results.keys()) / 100) + 1
+    batches = int(len(changeset_results.keys()) / batch_size) + 1
     logger.info(
-        f"""{len_osm} changesets will be queried in roughly {batches} batches"""
+        f"""{len_osm} changesets will be queried in roughly {batches} batches from osmCHA"""  # noqa E501
     )
-    chunk_list = chunks(list(changeset_results.keys()), 50)
+
+    chunk_list = chunks(list(changeset_results.keys()), batch_size)
     for i, subset in enumerate(chunk_list):
         changeset_results = query_osmcha(subset, changeset_results)
         progress = round(100 * ((i + 1) / len(chunk_list)), 1)
         logger.info(f"finished query {i+1}/{len(chunk_list)}, {progress}")
 
+    missing_ids = [i for i, v in changeset_results.items() if v is None]
+    chunk_list = chunks(missing_ids, batch_size)
+    batches = int(len(missing_ids) / batch_size) + 1
+    logger.info(
+        f"""{len(missing_ids)} changesets where missing from osmCHA and are now queried via osmAPI in {batches} batches"""  # noqa E501
+    )
+    for i, subset in enumerate(chunk_list):
+        changeset_results = query_osm(subset, changeset_results)
+        progress = round(100 * ((i + 1) / len(chunk_list)), 1)
+        logger.info(f"finished query {i+1}/{len(chunk_list)}, {progress}")
+
     for feature in json["features"]:
         changeset = changeset_results[int(feature["properties"]["changesetId"])]
-        logger.warn(changeset)
         for attribute_name in ["username", "comment", "editor", "userid"]:
             feature["properties"][attribute_name] = changeset[attribute_name]