-from check_functions import Almost
-
-class PlaceColumn:
-
- def __init__(self, context, force_name):
- self.columns = { 'admin_level' : 15}
- self.force_name = force_name
- self.context = context
- self.geometry = None
-
- def add(self, key, value):
- if hasattr(self, 'set_key_' + key):
- getattr(self, 'set_key_' + key)(value)
- elif key.startswith('name+'):
- self.add_hstore('name', key[5:], value)
- elif key.startswith('extra+'):
- self.add_hstore('extratags', key[6:], value)
- elif key.startswith('addr+'):
- self.add_hstore('address', key[5:], value)
- elif key in ('name', 'address', 'extratags'):
- self.columns[key] = eval('{' + value + '}')
- else:
- assert key in ('class', 'type')
- self.columns[key] = None if value == '' else value
-
- def set_key_name(self, value):
- self.add_hstore('name', 'name', value)
-
- def set_key_osm(self, value):
- assert value[0] in 'NRW'
- assert value[1:].isdigit()
-
- self.columns['osm_type'] = value[0]
- self.columns['osm_id'] = int(value[1:])
-
- def set_key_admin(self, value):
- self.columns['admin_level'] = int(value)
-
- def set_key_housenr(self, value):
- if value:
- self.add_hstore('address', 'housenumber', value)
-
- def set_key_postcode(self, value):
- if value:
- self.add_hstore('address', 'postcode', value)
-
- def set_key_street(self, value):
- if value:
- self.add_hstore('address', 'street', value)
-
- def set_key_addr_place(self, value):
- if value:
- self.add_hstore('address', 'place', value)
-
- def set_key_country(self, value):
- if value:
- self.add_hstore('address', 'country', value)
-
- def set_key_geometry(self, value):
- self.geometry = self.context.osm.parse_geometry(value, self.context.scene)
- assert self.geometry is not None
-
- def add_hstore(self, column, key, value):
- if column in self.columns:
- self.columns[column][key] = value
- else:
- self.columns[column] = { key : value }
-
- def db_insert(self, cursor):
- assert 'osm_type' in self.columns
- if self.force_name and 'name' not in self.columns:
- self.add_hstore('name', 'name', ''.join(random.choice(string.printable)
- for _ in range(int(random.random()*30))))
-
- if self.columns['osm_type'] == 'N' and self.geometry is None:
- pt = self.context.osm.grid_node(self.columns['osm_id'])
- if pt is None:
- pt = (random.random()*360 - 180, random.random()*180 - 90)
-
- self.geometry = "ST_SetSRID(ST_Point(%f, %f), 4326)" % pt
- else:
- assert self.geometry is not None, "Geometry missing"
- query = 'INSERT INTO place (%s, geometry) values(%s, %s)' % (
- ','.join(self.columns.keys()),
- ','.join(['%s' for x in range(len(self.columns))]),
- self.geometry)
- cursor.execute(query, list(self.columns.values()))
-
-class LazyFmt(object):
-
- def __init__(self, fmtstr, *args):
- self.fmt = fmtstr
- self.args = args
-
- def __str__(self):
- return self.fmt % self.args
-
-class PlaceObjName(object):
-
- def __init__(self, placeid, conn):
- self.pid = placeid
- self.conn = conn
-
- def __str__(self):
- if self.pid is None:
- return "<null>"
-
- if self.pid == 0:
- return "place ID 0"
-
- cur = self.conn.cursor()
- cur.execute("""SELECT osm_type, osm_id, class
- FROM placex WHERE place_id = %s""",
- (self.pid, ))
- assert cur.rowcount == 1, "No entry found for place id %s" % self.pid
-
- return "%s%s:%s" % cur.fetchone()
-
-def compare_place_id(expected, result, column, context):
- if expected == '0':
- assert result == 0, \
- LazyFmt("Bad place id in column %s. Expected: 0, got: %s.",
- column, PlaceObjName(result, context.db))
- elif expected == '-':
- assert result is None, \
- LazyFmt("bad place id in column %s: %s.",
- column, PlaceObjName(result, context.db))
- else:
- assert NominatimID(expected).get_place_id(context.db.cursor()) == result, \
- LazyFmt("Bad place id in column %s. Expected: %s, got: %s.",
- column, expected, PlaceObjName(result, context.db))