--- /dev/null
+#! /usr/bin/env python
+#-----------------------------------------------------------------------------
+# nominatim - [description]
+#-----------------------------------------------------------------------------
+#
+# Indexing tool for the Nominatim database.
+#
+# Based on C version by Brian Quinion
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#-----------------------------------------------------------------------------
+
+from argparse import ArgumentParser, RawDescriptionHelpFormatter, ArgumentTypeError
+import logging
+import sys
+import re
+import getpass
+from datetime import datetime
+import psycopg2
+from psycopg2.extras import wait_select
+import select
+
+log = logging.getLogger()
+
+def make_connection(options, asynchronous=False):
+ return psycopg2.connect(dbname=options.dbname, user=options.user,
+ password=options.password, host=options.host,
+ port=options.port, async_=asynchronous)
+
+class IndexingThread(object):
+
+ def __init__(self, thread_num, options):
+ log.debug("Creating thread {}".format(thread_num))
+ self.thread_num = thread_num
+ self.conn = make_connection(options, asynchronous=True)
+ self.wait()
+
+ self.cursor = self.conn.cursor()
+ self.perform("SET lc_messages TO 'C'")
+ self.wait()
+
+ self.current_query = None
+
+ def wait(self):
+ wait_select(self.conn)
+ self.current_query = None
+
+ def perform(self, sql, args=None):
+ self.current_query = sql
+ self.cursor.execute(sql, args)
+
+ def is_done(self):
+ if self.current_query is None:
+ return True
+
+ try:
+ if self.conn.poll() == psycopg2.extensions.POLL_OK:
+ self.current_query = None
+ return True
+ except psycopg2.extensions.TransactionRollbackError as e:
+ if e.pgcode is None:
+ raise RuntimeError("Postgres exception has no error code")
+ if e.pgcode == '40P01':
+ log.info("Deadlock detected, retry.")
+ self.cursor.execute(sql)
+ else:
+ raise
+
+
+
+class Indexer(object):
+
+ def __init__(self, options):
+ self.options = options
+ self.conn = make_connection(options)
+
+ self.threads = []
+ self.poll = select.poll()
+ for i in range(options.threads):
+ t = IndexingThread(i, options)
+ self.threads.append(t)
+ self.poll.register(t.conn.fileno(), select.EPOLLIN)
+ self.next_thread = 0
+
+ def run(self):
+ log.info("Starting indexing rank ({} to {}) using {} threads".format(
+ self.options.minrank, self.options.maxrank,
+ self.options.threads))
+
+ for rank in range(self.options.minrank, 30):
+ self.index(RankRunner(rank))
+
+ if self.options.maxrank >= 30:
+ self.index(InterpolationRunner())
+ self.index(RankRunner(30))
+
+ def index(self, obj):
+ log.info("Starting {}".format(obj.name()))
+
+ cur = self.conn.cursor(name="main")
+ cur.execute(obj.sql_index_sectors())
+
+ total_tuples = 0
+ for r in cur:
+ total_tuples += r[1]
+ log.debug("Total number of rows; {}".format(total_tuples))
+
+ cur.scroll(0, mode='absolute')
+
+ done_tuples = 0
+ rank_start_time = datetime.now()
+ for r in cur:
+ sector = r[0]
+
+ # Should we do the remaining ones together?
+ do_all = total_tuples - done_tuples < len(self.threads) * 1000
+
+ pcur = self.conn.cursor(name='places')
+
+ if do_all:
+ pcur.execute(obj.sql_nosector_places())
+ else:
+ pcur.execute(obj.sql_sector_places(), (sector, ))
+
+ for place in pcur:
+ place_id = place[0]
+ log.debug("Processing place {}".format(place_id))
+ thread = self.find_free_thread()
+
+ thread.perform(obj.sql_index_place(), (place_id,))
+ done_tuples += 1
+
+ pcur.close()
+
+ if do_all:
+ break
+
+ cur.close()
+
+ for t in self.threads:
+ t.wait()
+
+ rank_end_time = datetime.now()
+ diff_seconds = (rank_end_time-rank_start_time).total_seconds()
+
+ log.info("Done {} in {} @ {} per second - FINISHED {}\n".format(
+ done_tuples, int(diff_seconds),
+ done_tuples/diff_seconds, obj.name()))
+
+ def find_free_thread(self):
+ while True:
+ for t in self.threads:
+ if t.is_done():
+ return t
+
+ self.poll.poll()
+
+ assert(False, "Unreachable code")
+
+class RankRunner(object):
+
+ def __init__(self, rank):
+ self.rank = rank
+
+ def name(self):
+ return "rank {}".format(self.rank)
+
+ def sql_index_sectors(self):
+ return """SELECT geometry_sector, count(*) FROM placex
+ WHERE rank_search = {} and indexed_status > 0
+ GROUP BY geometry_sector
+ ORDER BY geometry_sector""".format(self.rank)
+
+ def sql_nosector_places(self):
+ return """SELECT place_id FROM placex
+ WHERE indexed_status > 0 and rank_search = {}
+ ORDER BY geometry_sector""".format(self.rank)
+
+ def sql_sector_places(self):
+ return """SELECT place_id FROM placex
+ WHERE indexed_status > 0 and geometry_sector = %s
+ ORDER BY geometry_sector"""
+
+ def sql_index_place(self):
+ return "UPDATE placex SET indexed_status = 0 WHERE place_id = %s"
+
+
+class InterpolationRunner(object):
+
+ def name(self):
+ return "interpolation lines (location_property_osmline)"
+
+ def sql_index_sectors(self):
+ return """SELECT geometry_sector, count(*) FROM location_property_osmline
+ WHERE indexed_status > 0
+ GROUP BY geometry_sector
+ ORDER BY geometry_sector"""
+
+ def sql_nosector_places(self):
+ return """SELECT place_id FROM location_property_osmline
+ WHERE indexed_status > 0
+ ORDER BY geometry_sector"""
+
+ def sql_sector_places(self):
+ return """SELECT place_id FROM location_property_osmline
+ WHERE indexed_status > 0 and geometry_sector = %s
+ ORDER BY geometry_sector"""
+
+ def sql_index_place(self):
+ return """UPDATE location_property_osmline
+ SET indexed_status = 0 WHERE place_id = %s"""
+
+
+def nominatim_arg_parser():
+ """ Setup the command-line parser for the tool.
+ """
+ def h(s):
+ return re.sub("\s\s+" , " ", s)
+
+ p = ArgumentParser(description=__doc__,
+ formatter_class=RawDescriptionHelpFormatter)
+
+ p.add_argument('-d', '--database',
+ dest='dbname', action='store', default='nominatim',
+ help='Name of the PostgreSQL database to connect to.')
+ p.add_argument('-U', '--username',
+ dest='user', action='store',
+ help='PostgreSQL user name.')
+ p.add_argument('-W', '--password',
+ dest='password_prompt', action='store_true',
+ help='Force password prompt.')
+ p.add_argument('-H', '--host',
+ dest='host', action='store',
+ help='PostgreSQL server hostname or socket location.')
+ p.add_argument('-P', '--port',
+ dest='port', action='store',
+ help='PostgreSQL server port')
+ p.add_argument('-r', '--minrank',
+ dest='minrank', type=int, metavar='RANK', default=0,
+ help='Minimum/starting rank.')
+ p.add_argument('-R', '--maxrank',
+ dest='maxrank', type=int, metavar='RANK', default=30,
+ help='Maximum/finishing rank.')
+ p.add_argument('-t', '--threads',
+ dest='threads', type=int, metavar='NUM', default=1,
+ help='Number of threads to create for indexing.')
+ p.add_argument('-v', '--verbose',
+ dest='loglevel', action='count', default=0,
+ help='Increase verbosity')
+
+ return p
+
+if __name__ == '__main__':
+ logging.basicConfig(stream=sys.stderr, format='%(levelname)s: %(message)s')
+
+ options = nominatim_arg_parser().parse_args(sys.argv[1:])
+
+ log.setLevel(max(3 - options.loglevel, 0) * 10)
+
+ options.password = None
+ if options.password_prompt:
+ password = getpass.getpass("Database password: ")
+ options.password = password
+
+ Indexer(options).run()
projects / nominatim.git / commitdiff