test/bdd/steps/place_inserter.py

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