nominatim/indexer/runners.py

   1 """
   2 Mix-ins that provide the actual commands for the indexer for various indexing
   3 tasks.
   4 """
   5 import functools
   6
   7 import psycopg2.extras
   8
   9 # pylint: disable=C0111
  10
  11 class AbstractPlacexRunner:
  12     """ Returns SQL commands for indexing of the placex table.
  13     """
  14     SELECT_SQL = 'SELECT place_id FROM placex'
  15
  16     def __init__(self, rank, analyzer):
  17         self.rank = rank
  18         self.analyzer = analyzer
  19
  20
  21     @staticmethod
  22     @functools.lru_cache(maxsize=1)
  23     def _index_sql(num_places):
  24         return """ UPDATE placex
  25                    SET indexed_status = 0, address = v.addr, token_info = v.ti
  26                    FROM (VALUES {}) as v(id, addr, ti)
  27                    WHERE place_id = v.id
  28                """.format(','.join(["(%s, %s::hstore, %s::jsonb)"]  * num_places))
  29
  30
  31     def get_place_details(self, worker, ids):
  32         worker.perform("""SELECT place_id, (placex_prepare_update(placex)).*
  33                           FROM placex WHERE place_id IN %s""",
  34                        (tuple((p[0] for p in ids)), ))
  35
  36
  37     def index_places(self, worker, places):
  38         values = []
  39         for place in places:
  40             values.extend((place[x] for x in ('place_id', 'address')))
  41             values.append(psycopg2.extras.Json(self.analyzer.process_place(place)))
  42
  43         worker.perform(self._index_sql(len(places)), values)
  44
  45
  46 class RankRunner(AbstractPlacexRunner):
  47     """ Returns SQL commands for indexing one rank within the placex table.
  48     """
  49
  50     def name(self):
  51         return "rank {}".format(self.rank)
  52
  53     def sql_count_objects(self):
  54         return """SELECT count(*) FROM placex
  55                   WHERE rank_address = {} and indexed_status > 0
  56                """.format(self.rank)
  57
  58     def sql_get_objects(self):
  59         return """{} WHERE indexed_status > 0 and rank_address = {}
  60                      ORDER BY geometry_sector
  61                """.format(self.SELECT_SQL, self.rank)
  62
  63
  64 class BoundaryRunner(AbstractPlacexRunner):
  65     """ Returns SQL commands for indexing the administrative boundaries
  66         of a certain rank.
  67     """
  68
  69     def name(self):
  70         return "boundaries rank {}".format(self.rank)
  71
  72     def sql_count_objects(self):
  73         return """SELECT count(*) FROM placex
  74                   WHERE indexed_status > 0
  75                     AND rank_search = {}
  76                     AND class = 'boundary' and type = 'administrative'
  77                """.format(self.rank)
  78
  79     def sql_get_objects(self):
  80         return """{} WHERE indexed_status > 0 and rank_search = {}
  81                            and class = 'boundary' and type = 'administrative'
  82                      ORDER BY partition, admin_level
  83                """.format(self.SELECT_SQL, self.rank)
  84
  85
  86 class InterpolationRunner:
  87     """ Returns SQL commands for indexing the address interpolation table
  88         location_property_osmline.
  89     """
  90
  91     def __init__(self, analyzer):
  92         self.analyzer = analyzer
  93
  94
  95     @staticmethod
  96     def name():
  97         return "interpolation lines (location_property_osmline)"
  98
  99     @staticmethod
 100     def sql_count_objects():
 101         return """SELECT count(*) FROM location_property_osmline
 102                   WHERE indexed_status > 0"""
 103
 104     @staticmethod
 105     def sql_get_objects():
 106         return """SELECT place_id, get_interpolation_address(address, osm_id) as address
 107                   FROM location_property_osmline
 108                   WHERE indexed_status > 0
 109                   ORDER BY geometry_sector"""
 110
 111
 112     @staticmethod
 113     @functools.lru_cache(maxsize=1)
 114     def _index_sql(num_places):
 115         return """ UPDATE location_property_osmline
 116                    SET indexed_status = 0, address = v.addr, token_info = v.ti
 117                    FROM (VALUES {}) as v(id, addr, ti)
 118                    WHERE place_id = v.id
 119                """.format(','.join(["(%s, %s::hstore, %s::jsonb)"]  * num_places))
 120
 121
 122     def index_places(self, worker, places):
 123         values = []
 124         for place in places:
 125             values.extend((place[x] for x in ('place_id', 'address')))
 126             values.append(psycopg2.extras.Json(self.analyzer.process_place(place)))
 127
 128         worker.perform(self._index_sql(len(places)), values)
 129
 130
 131
 132 class PostcodeRunner:
 133     """ Provides the SQL commands for indexing the location_postcode table.
 134     """
 135
 136     @staticmethod
 137     def name():
 138         return "postcodes (location_postcode)"
 139
 140     @staticmethod
 141     def sql_count_objects():
 142         return 'SELECT count(*) FROM location_postcode WHERE indexed_status > 0'
 143
 144     @staticmethod
 145     def sql_get_objects():
 146         return """SELECT place_id FROM location_postcode
 147                   WHERE indexed_status > 0
 148                   ORDER BY country_code, postcode"""
 149
 150     @staticmethod
 151     def index_places(worker, ids):
 152         worker.perform(""" UPDATE location_postcode SET indexed_status = 0
 153                            WHERE place_id IN ({})
 154                        """.format(','.join((str(i[0]) for i in ids))))