nominatim/tools/tiger_data.py

   1 """
   2 Functions for importing tiger data and handling tarbar and directory files
   3 """
   4 import csv
   5 import io
   6 import logging
   7 import os
   8 import tarfile
   9
  10 import psycopg2.extras
  11
  12 from nominatim.db.connection import connect
  13 from nominatim.db.async_connection import WorkerPool
  14 from nominatim.db.sql_preprocessor import SQLPreprocessor
  15 from nominatim.errors import UsageError
  16
  17 LOG = logging.getLogger()
  18
  19
  20 def handle_tarfile_or_directory(data_dir):
  21     """ Handles tarfile or directory for importing tiger data
  22     """
  23
  24     tar = None
  25     if data_dir.endswith('.tar.gz'):
  26         try:
  27             tar = tarfile.open(data_dir)
  28         except tarfile.ReadError as err:
  29             LOG.fatal("Cannot open '%s'. Is this a tar file?", data_dir)
  30             raise UsageError("Cannot open Tiger data file.") from err
  31
  32         csv_files = [i for i in tar.getmembers() if i.name.endswith('.csv')]
  33         LOG.warning("Found %d CSV files in tarfile with path %s", len(csv_files), data_dir)
  34         if not csv_files:
  35             LOG.warning("Tiger data import selected but no files in tarfile's path %s", data_dir)
  36             return None, None
  37     else:
  38         files = os.listdir(data_dir)
  39         csv_files = [os.path.join(data_dir, i) for i in files if i.endswith('.csv')]
  40         LOG.warning("Found %d CSV files in path %s", len(csv_files), data_dir)
  41         if not csv_files:
  42             LOG.warning("Tiger data import selected but no files found in path %s", data_dir)
  43             return None, None
  44
  45     return csv_files, tar
  46
  47
  48 def handle_threaded_sql_statements(pool, fd, analyzer):
  49     """ Handles sql statement with multiplexing
  50     """
  51     lines = 0
  52     # Using pool of database connections to execute sql statements
  53
  54     sql = "SELECT tiger_line_import(%s, %s, %s, %s, %s, %s)"
  55
  56     for row in csv.DictReader(fd, delimiter=';'):
  57         try:
  58             address = dict(street=row['street'], postcode=row['postcode'])
  59             args = ('SRID=4326;' + row['geometry'],
  60                     int(row['from']), int(row['to']), row['interpolation'],
  61                     psycopg2.extras.Json(analyzer.process_place(dict(address=address))),
  62                     analyzer.normalize_postcode(row['postcode']))
  63         except ValueError:
  64             continue
  65         pool.next_free_worker().perform(sql, args=args)
  66
  67         lines += 1
  68         if lines == 1000:
  69             print('.', end='', flush=True)
  70             lines = 0
  71
  72
  73 def add_tiger_data(data_dir, config, threads, tokenizer):
  74     """ Import tiger data from directory or tar file `data dir`.
  75     """
  76     dsn = config.get_libpq_dsn()
  77     files, tar = handle_tarfile_or_directory(data_dir)
  78
  79     if not files:
  80         return
  81
  82     with connect(dsn) as conn:
  83         sql = SQLPreprocessor(conn, config)
  84         sql.run_sql_file(conn, 'tiger_import_start.sql')
  85
  86     # Reading files and then for each file line handling
  87     # sql_query in <threads - 1> chunks.
  88     place_threads = max(1, threads - 1)
  89
  90     with WorkerPool(dsn, place_threads, ignore_sql_errors=True) as pool:
  91         with tokenizer.name_analyzer() as analyzer:
  92             for fname in files:
  93                 if not tar:
  94                     fd = open(fname)
  95                 else:
  96                     fd = io.TextIOWrapper(tar.extractfile(fname))
  97
  98                 handle_threaded_sql_statements(pool, fd, analyzer)
  99
 100                 fd.close()
 101
 102     if tar:
 103         tar.close()
 104     print('\n')
 105     LOG.warning("Creating indexes on Tiger data")
 106     with connect(dsn) as conn:
 107         sql = SQLPreprocessor(conn, config)
 108         sql.run_sql_file(conn, 'tiger_import_finish.sql')