test/bdd/steps/place_inserter.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2025 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Helper classes for filling the place table.
   9 """
  10 import random
  11 import string
  12
  13
  14 class PlaceColumn:
  15     """ Helper class to collect contents from a behave table row and
  16         insert it into the place table.
  17     """
  18     def __init__(self, context):
  19         self.columns = {'admin_level': 15}
  20         self.context = context
  21         self.geometry = None
  22
  23     def add_row(self, row, force_name):
  24         """ Parse the content from the given behave row as place column data.
  25         """
  26         for name, value in zip(row.headings, row.cells):
  27             self._add(name, value)
  28
  29         assert 'osm_type' in self.columns, "osm column missing"
  30
  31         if force_name and 'name' not in self.columns:
  32             self._add_hstore(
  33                 'name',
  34                 'name',
  35                 ''.join(random.choices(string.printable, k=random.randrange(30))),
  36             )
  37
  38         return self
  39
  40     def _add(self, key, value):
  41         if hasattr(self, '_set_key_' + key):
  42             getattr(self, '_set_key_' + key)(value)
  43         elif key.startswith('name+'):
  44             self._add_hstore('name', key[5:], value)
  45         elif key.startswith('extra+'):
  46             self._add_hstore('extratags', key[6:], value)
  47         elif key.startswith('addr+'):
  48             self._add_hstore('address', key[5:], value)
  49         elif key in ('name', 'address', 'extratags'):
  50             self.columns[key] = eval('{' + value + '}')
  51         else:
  52             assert key in ('class', 'type'), "Unknown column '{}'.".format(key)
  53             self.columns[key] = None if value == '' else value
  54
  55     def _set_key_name(self, value):
  56         self._add_hstore('name', 'name', value)
  57
  58     def _set_key_osm(self, value):
  59         assert value[0] in 'NRW' and value[1:].isdigit(), \
  60                "OSM id needs to be of format <NRW><id>."
  61
  62         self.columns['osm_type'] = value[0]
  63         self.columns['osm_id'] = int(value[1:])
  64
  65     def _set_key_admin(self, value):
  66         self.columns['admin_level'] = int(value)
  67
  68     def _set_key_housenr(self, value):
  69         if value:
  70             self._add_hstore('address', 'housenumber', value)
  71
  72     def _set_key_postcode(self, value):
  73         if value:
  74             self._add_hstore('address', 'postcode', value)
  75
  76     def _set_key_street(self, value):
  77         if value:
  78             self._add_hstore('address', 'street', value)
  79
  80     def _set_key_addr_place(self, value):
  81         if value:
  82             self._add_hstore('address', 'place', value)
  83
  84     def _set_key_country(self, value):
  85         if value:
  86             self._add_hstore('address', 'country', value)
  87
  88     def _set_key_geometry(self, value):
  89         self.geometry = self.context.osm.parse_geometry(value)
  90         assert self.geometry is not None, "Bad geometry: {}".format(value)
  91
  92     def _add_hstore(self, column, key, value):
  93         if column in self.columns:
  94             self.columns[column][key] = value
  95         else:
  96             self.columns[column] = {key: value}
  97
  98     def db_delete(self, cursor):
  99         """ Issue a delete for the given OSM object.
 100         """
 101         cursor.execute('DELETE FROM place WHERE osm_type = %s and osm_id = %s',
 102                        (self.columns['osm_type'], self.columns['osm_id']))
 103
 104     def db_insert(self, cursor):
 105         """ Insert the collected data into the database.
 106         """
 107         if self.columns['osm_type'] == 'N' and self.geometry is None:
 108             pt = self.context.osm.grid_node(self.columns['osm_id'])
 109             if pt is None:
 110                 pt = (random.uniform(-180, 180), random.uniform(-90, 90))
 111
 112             self.geometry = "ST_SetSRID(ST_Point(%f, %f), 4326)" % pt
 113         else:
 114             assert self.geometry is not None, "Geometry missing"
 115
 116         query = 'INSERT INTO place ({}, geometry) values({}, {})'.format(
 117             ','.join(self.columns.keys()),
 118             ','.join(['%s' for x in range(len(self.columns))]),
 119             self.geometry)
 120         cursor.execute(query, list(self.columns.values()))