Merge pull request #149 from GIScience/feat/import-multipass

feat: import using multiple passes

Author: takb

.github/workflows/run_tests.yml

@@ -10,8 +10,12 @@ jobs:
   run_tests:
     runs-on: ubuntu-22.04
     steps:
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: 3.11
     - name: Check out source
-      uses: actions/checkout@v2
+      uses: actions/checkout@v4
       with:
         fetch-depth: 0
     - name: Install postgis
@@ -21,9 +25,9 @@ jobs:
         postgresql user: 'gis_admin'
         postgresql password: 'admin'
     - name: Install packages
-      run: sudo apt-get update && sudo apt-get install -y build-essential protobuf-compiler libprotobuf-dev python3 #proj-bin libproj-dev
+      run: sudo apt-get update && sudo apt-get install -y build-essential protobuf-compiler libprotobuf-dev
     - name: Activate cache for python packages
-      uses: actions/cache@v2
+      uses: actions/cache@v4
       with:
         path: ~/.cache/pip
         key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.9-slim
+FROM python:3.11-slim
 LABEL org.opencontainers.image.authors="Timothy Ellersiek <timothy@openrouteservice.org>"
 
 # protobuf is required to parse osm files.

docker-compose-standalone.yml

@@ -1,4 +1,3 @@
-version: '2.2'
 services:
  api:
   container_name: ops-api

docker-compose.yml

@@ -1,5 +1,3 @@
-version: "2.2"
-
 volumes:
  postgis-data:
 
@@ -67,8 +65,6 @@ services:
   mem_limit: 28g
   networks:
    - poi_network
-  profiles:
-   - init
 
  update:
   container_name: ops-update
@@ -86,8 +82,6 @@ services:
   mem_limit: 28g
   networks:
    - poi_network
-  profiles:
-   - update
 
  test:
   container_name: ops-test

manage.py

@@ -12,7 +12,7 @@
 
 logging.basicConfig(
     level=logging.DEBUG if os.environ.get('OPS_DEBUG', False) else logging.INFO,
-    format='%(levelname)-8s %(message)s',
+    format='%(asctime)s %(levelname)-8s %(message)s',
 )
 logger = logging.getLogger(__name__)
 
@@ -68,7 +68,7 @@ def import_data():
             continue
         logger.info(f"Found directory: {dir_name}")
         for filename in file_list:
-            if filename.endswith(".osm.pbf") or filename.endswith(".osm"):
+            if filename.endswith(".pbf") or filename.endswith(".osm"):
                 osm_files.append(os.path.join(dir_name, filename))
     osm_files.sort()
 
@@ -82,18 +82,22 @@ def import_data():
             finally:
                 f.close()
 
-    # we have found previous data in the database, check if file list has changed which would require a full rebuild
-    if len(import_log) and set(import_log.keys()) != set(osm_files):
-        logger.error(f"File set has changed since last import, full rebuild required. Exiting.")
-        return
+    # check if file list has changed which would require a full rebuild by deleting the import log
+    if len(import_log):
+        if set(import_log.keys()) != set(osm_files):
+            logger.error(f"File set has changed since last import, full rebuild required. Exiting.")
+            return
+    else:
+        import_log = {key: 0 for key in osm_files}
+        with open(logfile, "w") as f:
+            json.dump(import_log, f, indent=4, sort_keys=True)
+            f.close()
 
     logger.info(f"Starting to import OSM data ({len(osm_files)} files in batch)")
     logger.debug(f"Files in import batch: {osm_files}")
-    parser.run_import(osm_files, import_log, db)
+    # parser.run_import(osm_files, import_log, db)
+    parser.run_import_new(osm_files, import_log, logfile, db)
 
-    with open(logfile, "w") as f:
-        json.dump(import_log, f, indent=4, sort_keys=True)
-        f.close()
 
 
 if __name__ == '__main__':

openpoiservice/server/db_import/node_store.py

@@ -0,0 +1,68 @@
+import numpy as np
+from cykhash import Int64toInt64Map
+
+class NodeStore:
+    def __init__(self, initial_capacity=1024):
+        self.size = 0
+        self.capacity = initial_capacity
+        self.index_map = Int64toInt64Map()
+        self.data = np.empty((self.capacity, 2), dtype=np.float64)
+
+    def _resize(self):
+        # Double the capacity
+        self.capacity *= 2
+        new_data = np.empty((self.capacity, 2), dtype=np.float64)
+        new_data[:self.size] = self.data  # Copy old data
+        self.data = new_data
+
+    def append(self, key, value_tuple):
+        """
+        Insert a new key-value pair into the store.
+
+        Args:
+            key (int): The key to insert.
+            value_tuple (tuple of floats): The associated value tuple.
+        """
+        if key in self.index_map:
+            return self.set(key, value_tuple)
+
+        self.index_map[key] = self.size
+        if self.size >= self.capacity:
+            self._resize()
+        self.data[self.size] = value_tuple
+        self.size += 1
+
+    def set(self, key, value_tuple):
+        """
+        Set the value of the key in the store.
+
+        Args:
+            key (int): The key to modify.
+            value_tuple (tuple of floats): The new value tuple.
+        """
+        if not key in self.index_map:
+            raise KeyError(f"Key does not exist.")
+        idx = self.index_map.get(key)
+        self.data[idx] = value_tuple
+
+    def compact(self):
+        """
+        Compact the internal storage NumPy array for memory efficiency.
+        """
+        self.data = self.data[:self.size]
+        self.capacity = self.size
+
+    def get(self, key):
+        """Retrieve the value tuple for a given key."""
+        idx = self.index_map.get(key)
+        if idx is not None:
+            return self.data[idx]
+        return None
+
+    def __contains__(self, key):
+        """Check if a key exists in the store."""
+        return key in self.index_map
+
+    def __len__(self):
+        """Return the number of stored keys."""
+        return len(self.index_map)

openpoiservice/server/db_import/parse_osm.py

@@ -1,15 +1,22 @@
 # openpoiservice/server/parse_osm.py
-from openpoiservice.server import db
-from openpoiservice.server import categories_tools, ops_settings
-from openpoiservice.server.db_import.models import POIs, Tags, Categories
-from openpoiservice.server.db_import.objects import PoiObject, TagsObject
+
 import logging
 from bisect import bisect_left
 from collections import deque
 
+import numpy as np
+from cykhash.khashsets import Int64Set
+
+from openpoiservice.server import categories_tools, ops_settings
+from openpoiservice.server import db
+from openpoiservice.server.db_import.models import POIs, Tags, Categories
+from openpoiservice.server.db_import.node_store import NodeStore
+from openpoiservice.server.db_import.objects import PoiObject, TagsObject
+
 logger = logging.getLogger(__name__)
 
 
+# deprecated, remove later
 class WayObject(object):
     """ Class that creates a way object. """
 
@@ -60,20 +67,22 @@ def __init__(self, osm_file_index, update_mode=False):
         self.tags_cnt = 0
         self.categories_cnt = 0
         self.relation_ways = {}
-        self.nodes = {}
-        self.process_ways = []
+        self.nodes_store = NodeStore()
+        self.process_ways_set = Int64Set()
         self.poi_objects = []
         self.tags_objects = []
         self.categories_objects = []
-        self.ways_temp = []
-        self.ways_obj = None
-        self.tags_object = None
-        self.poi_object = None
-        self.process_ways_length = None
         self.update_mode = update_mode
         self.osm_file_index = osm_file_index
         self.failed = False
 
+        # deprecated, remove later
+        self.nodes = {}
+        self.process_ways = []
+        self.ways_obj = None
+        self.ways_temp = []
+
+
     def parse_nodes(self, osm_nodes):
         """
         Callback function called by imposm while nodes are parsed.
@@ -89,6 +98,7 @@ def parse_nodes(self, osm_nodes):
                 self.failed = True
                 return
 
+
     def parse_relations(self, relations):
         """
         Callback function called by imposm while relations are parsed. The idea is to extract polygons which may
@@ -116,6 +126,8 @@ def parse_relations(self, relations):
                         self.relation_ways[osmid_rel_member].update({"relation_id": osmid})
                         self.relations_cnt += 1
 
+
+    # deprecated, remove later
     def parse_ways(self, ways):
         """
         Callback function called by imposm while ways are parsed. If a category can't be found it may likely
@@ -149,6 +161,8 @@ def parse_ways(self, ways):
                 self.ways_obj = WayObject(osmid, osm_type, tags, refs, categories, len(refs))
                 self.process_ways.append(self.ways_obj)
 
+
+    # deprecated, remove later
     def parse_coords_for_ways(self, coords):
         """
         Callback function called by imposm while coordinates are parsed. Due due ordering we can use coords
@@ -225,6 +239,93 @@ def parse_coords_for_ways(self, coords):
 
             self.ways_temp = []
 
+
+    def parse_ways_first(self, ways):
+        """
+        Callback function called by imposm while ways are parsed. If a category can't be found it may likely
+        be that the osmid of this way can be found in self.relation_ways which will contain additional tags
+        and therefore eventually a category. A way object is added to a list process_ways which at this point
+        is lacking coordinates -> next step.
+
+        :param ways: osm way objects
+        :type ways: list of osm ways
+        """
+        for osmid, tags, refs in ways:
+            if len(refs) >= 1000:
+                continue
+
+            categories = categories_tools.get_category(tags)
+
+            if len(categories) == 0 and osmid in self.relation_ways:
+                tags = self.relation_ways[osmid]
+                categories = categories_tools.get_category(tags)
+
+            if len(categories) > 0:
+                self.ways_cnt += 1
+                for ref in refs:
+                    self.nodes_store.append(ref, (None, None))
+                    self.process_ways_set.add(osmid)
+
+
+    def parse_coords_and_store(self, coords):
+        for osmid, lat, lng in coords:
+            if osmid in self.nodes_store:
+                self.nodes_store.set(osmid, (lat, lng))
+
+
+    def parse_ways_second(self, ways):
+        """
+        Callback function called by imposm while ways are parsed. If a category can't be found it may likely
+        be that the osmid of this way can be found in self.relation_ways which will contain additional tags
+        and therefore eventually a category. A way object is added to a list process_ways which at this point
+        is lacking coordinates -> next step.
+
+        :param ways: osm way objects
+        :type ways: list of osm ways
+        """
+        for osmid, tags, refs in ways:
+            if osmid not in self.process_ways_set:
+                continue
+            categories = categories_tools.get_category(tags)
+            # from way
+            osm_type = 2
+
+            if len(categories) == 0 and osmid in self.relation_ways:
+                # current way is the outer ring of a relation which was marked as having a category
+                tags = self.relation_ways[osmid]
+                categories = categories_tools.get_category(tags)
+                # from relation
+                osm_type = 3
+
+            # Calculate centroid of way
+            refs = set(refs)
+            sum_lat = 0
+            sum_lng = 0
+            way_valid = True
+            for ref in refs:
+                if ref not in self.nodes_store: # should never ha
+                    way_valid = False
+                    break
+                lat, lng = self.nodes_store.get(ref)
+                if lat is None or lng is None or np.isnan(lat) or np.isnan(lng):
+                    way_valid = False
+                    break
+                sum_lat += lat
+                sum_lng += lng
+            if not way_valid:
+                continue
+
+            self.process_ways_set.remove(osmid)
+            centroid_lat = sum_lat / len(refs)
+            centroid_lng = sum_lng / len(refs)
+            try:
+                self.create_poi(osm_type, osmid, [centroid_lat, centroid_lng], tags, categories)
+            except Exception as e:
+                logger.debug(e)
+                self.failed = True
+                return
+
+
     def create_poi(self, osm_type, osm_id, lat_lng, tags, categories=None):
         """
         Creates a poi entity if a category is found. Stored afterwards.
@@ -257,6 +358,7 @@ def create_poi(self, osm_type, osm_id, lat_lng, tags, categories=None):
 
             self.store_poi(PoiObject(osm_type, osm_id, lat_lng, categories))
 
+
     def store_poi(self, poi_object):
         """
         Appends poi storage objects to buffer for bulk storage to database.
@@ -273,6 +375,7 @@ def store_poi(self, poi_object):
             logger.debug(f"Pois: {self.pois_count}, tags: {self.tags_cnt}, categories: {self.categories_cnt}")
             self.save_buffer()
 
+
     def store_tags(self, tags_object):
         """
         Appends tags storage objects to buffer for bulk storage to database.
@@ -285,6 +388,7 @@ def store_tags(self, tags_object):
             value=tags_object.value
         ))
 
+
     def store_categories(self, osmtype, osmid, category):
         """
         Appends category storage objects to buffer for bulk storage to database.
@@ -296,6 +400,7 @@ def store_categories(self, osmtype, osmid, category):
             category=category
         ))
 
+
     def save_buffer(self):
         """
         Save POIs, tags and categories to database and clear buffer.