nominatim/nominatim.py

   1 #! /usr/bin/env python3
   2 #-----------------------------------------------------------------------------
   3 # nominatim - [description]
   4 #-----------------------------------------------------------------------------
   5 #
   6 # Indexing tool for the Nominatim database.
   7 #
   8 # Based on C version by Brian Quinion
   9 #
  10 # This program is free software; you can redistribute it and/or
  11 # modify it under the terms of the GNU General Public License
  12 # as published by the Free Software Foundation; either version 2
  13 # of the License, or (at your option) any later version.
  14 #
  15 # This program is distributed in the hope that it will be useful,
  16 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  17 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18 # GNU General Public License for more details.
  19 #
  20 # You should have received a copy of the GNU General Public License
  21 # along with this program; if not, write to the Free Software
  22 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  23 #-----------------------------------------------------------------------------
  24 # pylint: disable=C0111
  25 from argparse import ArgumentParser, RawDescriptionHelpFormatter
  26 import logging
  27 import sys
  28 import getpass
  29 import select
  30
  31 from indexer.progress import ProgressLogger # pylint: disable=E0401
  32 from indexer.db import DBConnection, make_connection # pylint: disable=E0401
  33
  34 LOG = logging.getLogger()
  35
  36 class RankRunner:
  37     """ Returns SQL commands for indexing one rank within the placex table.
  38     """
  39
  40     def __init__(self, rank):
  41         self.rank = rank
  42
  43     def name(self):
  44         return "rank {}".format(self.rank)
  45
  46     def sql_count_objects(self):
  47         return """SELECT count(*) FROM placex
  48                   WHERE rank_address = {} and indexed_status > 0
  49                """.format(self.rank)
  50
  51     def sql_get_objects(self):
  52         return """SELECT place_id FROM placex
  53                   WHERE indexed_status > 0 and rank_address = {}
  54                   ORDER BY geometry_sector""".format(self.rank)
  55
  56     @staticmethod
  57     def sql_index_place(ids):
  58         return "UPDATE placex SET indexed_status = 0 WHERE place_id IN ({})"\
  59                .format(','.join((str(i) for i in ids)))
  60
  61
  62 class InterpolationRunner:
  63     """ Returns SQL commands for indexing the address interpolation table
  64         location_property_osmline.
  65     """
  66
  67     @staticmethod
  68     def name():
  69         return "interpolation lines (location_property_osmline)"
  70
  71     @staticmethod
  72     def sql_count_objects():
  73         return """SELECT count(*) FROM location_property_osmline
  74                   WHERE indexed_status > 0"""
  75
  76     @staticmethod
  77     def sql_get_objects():
  78         return """SELECT place_id FROM location_property_osmline
  79                   WHERE indexed_status > 0
  80                   ORDER BY geometry_sector"""
  81
  82     @staticmethod
  83     def sql_index_place(ids):
  84         return """UPDATE location_property_osmline
  85                   SET indexed_status = 0 WHERE place_id IN ({})"""\
  86                .format(','.join((str(i) for i in ids)))
  87
  88 class BoundaryRunner:
  89     """ Returns SQL commands for indexing the administrative boundaries
  90         of a certain rank.
  91     """
  92
  93     def __init__(self, rank):
  94         self.rank = rank
  95
  96     def name(self):
  97         return "boundaries rank {}".format(self.rank)
  98
  99     def sql_count_objects(self):
 100         return """SELECT count(*) FROM placex
 101                   WHERE indexed_status > 0
 102                     AND rank_search = {}
 103                     AND class = 'boundary' and type = 'administrative'""".format(self.rank)
 104
 105     def sql_get_objects(self):
 106         return """SELECT place_id FROM placex
 107                   WHERE indexed_status > 0 and rank_search = {}
 108                         and class = 'boundary' and type = 'administrative'
 109                   ORDER BY partition, admin_level""".format(self.rank)
 110
 111     @staticmethod
 112     def sql_index_place(ids):
 113         return "UPDATE placex SET indexed_status = 0 WHERE place_id IN ({})"\
 114                .format(','.join((str(i) for i in ids)))
 115
 116 class Indexer:
 117     """ Main indexing routine.
 118     """
 119
 120     def __init__(self, opts):
 121         self.minrank = max(1, opts.minrank)
 122         self.maxrank = min(30, opts.maxrank)
 123         self.conn = make_connection(opts)
 124         self.threads = [DBConnection(opts) for _ in range(opts.threads)]
 125
 126     def index_boundaries(self):
 127         LOG.warning("Starting indexing boundaries using %s threads",
 128                     len(self.threads))
 129
 130         for rank in range(max(self.minrank, 5), min(self.maxrank, 26)):
 131             self.index(BoundaryRunner(rank))
 132
 133     def index_by_rank(self):
 134         """ Run classic indexing by rank.
 135         """
 136         LOG.warning("Starting indexing rank (%i to %i) using %i threads",
 137                     self.minrank, self.maxrank, len(self.threads))
 138
 139         for rank in range(max(1, self.minrank), self.maxrank):
 140             self.index(RankRunner(rank))
 141
 142         if self.maxrank == 30:
 143             self.index(RankRunner(0))
 144             self.index(InterpolationRunner(), 20)
 145             self.index(RankRunner(self.maxrank), 20)
 146         else:
 147             self.index(RankRunner(self.maxrank))
 148
 149     def index(self, obj, batch=1):
 150         """ Index a single rank or table. `obj` describes the SQL to use
 151             for indexing. `batch` describes the number of objects that
 152             should be processed with a single SQL statement
 153         """
 154         LOG.warning("Starting %s (using batch size %s)", obj.name(), batch)
 155
 156         cur = self.conn.cursor()
 157         cur.execute(obj.sql_count_objects())
 158
 159         total_tuples = cur.fetchone()[0]
 160         LOG.debug("Total number of rows: %i", total_tuples)
 161
 162         cur.close()
 163
 164         progress = ProgressLogger(obj.name(), total_tuples)
 165
 166         if total_tuples > 0:
 167             cur = self.conn.cursor(name='places')
 168             cur.execute(obj.sql_get_objects())
 169
 170             next_thread = self.find_free_thread()
 171             while True:
 172                 places = [p[0] for p in cur.fetchmany(batch)]
 173                 if not places:
 174                     break
 175
 176                 LOG.debug("Processing places: %s", str(places))
 177                 thread = next(next_thread)
 178
 179                 thread.perform(obj.sql_index_place(places))
 180                 progress.add(len(places))
 181
 182             cur.close()
 183
 184             for thread in self.threads:
 185                 thread.wait()
 186
 187         progress.done()
 188
 189     def find_free_thread(self):
 190         """ Generator that returns the next connection that is free for
 191             sending a query.
 192         """
 193         ready = self.threads
 194         command_stat = 0
 195
 196         while True:
 197             for thread in ready:
 198                 if thread.is_done():
 199                     command_stat += 1
 200                     yield thread
 201
 202             # refresh the connections occasionaly to avoid potential
 203             # memory leaks in Postgresql.
 204             if command_stat > 100000:
 205                 for thread in self.threads:
 206                     while not thread.is_done():
 207                         thread.wait()
 208                     thread.connect()
 209                 command_stat = 0
 210                 ready = self.threads
 211             else:
 212                 ready, _, _ = select.select(self.threads, [], [])
 213
 214         assert False, "Unreachable code"
 215
 216
 217 def nominatim_arg_parser():
 218     """ Setup the command-line parser for the tool.
 219     """
 220     parser = ArgumentParser(description="Indexing tool for Nominatim.",
 221                             formatter_class=RawDescriptionHelpFormatter)
 222
 223     parser.add_argument('-d', '--database',
 224                         dest='dbname', action='store', default='nominatim',
 225                         help='Name of the PostgreSQL database to connect to.')
 226     parser.add_argument('-U', '--username',
 227                         dest='user', action='store',
 228                         help='PostgreSQL user name.')
 229     parser.add_argument('-W', '--password',
 230                         dest='password_prompt', action='store_true',
 231                         help='Force password prompt.')
 232     parser.add_argument('-H', '--host',
 233                         dest='host', action='store',
 234                         help='PostgreSQL server hostname or socket location.')
 235     parser.add_argument('-P', '--port',
 236                         dest='port', action='store',
 237                         help='PostgreSQL server port')
 238     parser.add_argument('-b', '--boundary-only',
 239                         dest='boundary_only', action='store_true',
 240                         help='Only index administrative boundaries (ignores min/maxrank).')
 241     parser.add_argument('-r', '--minrank',
 242                         dest='minrank', type=int, metavar='RANK', default=0,
 243                         help='Minimum/starting rank.')
 244     parser.add_argument('-R', '--maxrank',
 245                         dest='maxrank', type=int, metavar='RANK', default=30,
 246                         help='Maximum/finishing rank.')
 247     parser.add_argument('-t', '--threads',
 248                         dest='threads', type=int, metavar='NUM', default=1,
 249                         help='Number of threads to create for indexing.')
 250     parser.add_argument('-v', '--verbose',
 251                         dest='loglevel', action='count', default=0,
 252                         help='Increase verbosity')
 253
 254     return parser
 255
 256 if __name__ == '__main__':
 257     logging.basicConfig(stream=sys.stderr, format='%(levelname)s: %(message)s')
 258
 259     OPTIONS = nominatim_arg_parser().parse_args(sys.argv[1:])
 260
 261     LOG.setLevel(max(3 - OPTIONS.loglevel, 0) * 10)
 262
 263     OPTIONS.password = None
 264     if OPTIONS.password_prompt:
 265         PASSWORD = getpass.getpass("Database password: ")
 266         OPTIONS.password = PASSWORD
 267
 268     if OPTIONS.boundary_only:
 269         Indexer(OPTIONS).index_boundaries()
 270     else:
 271         Indexer(OPTIONS).index_by_rank()