voronoi_call_sample.py

_voronoi_call_sample.py

import csv
from contextlib import contextmanager

import psycopg2


@contextmanager
def cursor_context():
    conn = psycopg2.connect(dbname='osm', user='tbicr', password='admin',
                            host='127.0.0.1', port='5432')
    cursor = conn.cursor()
    psycopg2.extensions.register_type(psycopg2.extensions.UNICODE, cursor)
    yield cursor
    cursor.close()
    conn.close()


def main():
    with cursor_context() as cursor:
        spamreader = csv.reader(open("jes_minsk_edit_v14.csv", "r"))
        normalized_data = [[c for c in row] for i, row in enumerate(spamreader) if i > 0]
        data = [(j, so, ",".join(["'{}'".format(ni.strip()) for ni in n.split(",")]))
                for i, r, j, s, sf, so, n, no, nno in normalized_data]
    
        house_sql = "SELECT '{}' jes, p.way FROM osm_polygon p, osm_polygon i WHERE " \
                    "i.osm_id = -59195 AND ST_Contains(i.way, p.way) AND " \
                    "p.tags->'addr:street' = '{}' AND p.tags->'addr:housenumber' IN ({})"
        houses_sql = " UNION ".join([house_sql.format(*args) for args in data])
        houses_by_jes_sql = "SELECT p.jes, ST_Union(p.way) way FROM ({}) p GROUP BY p.jes".format(houses_sql)
    
        points_of_houses_by_jes_sql = "SELECT DISTINCT p.jes, (ST_DumpPoints(p.way)).geom way FROM ({}) p".format(houses_by_jes_sql)
        voronoi_sql = "SELECT ST_VoronoiPolygons(p.way, (SELECT way FROM osm_polygon WHERE osm_id = -59195)) way " \
                      "FROM ({}) p".format(points_of_houses_by_jes_sql)
        # voronoi_sql = "SELECT way FROM py_voronoi('({}) p', 'p.way', 'SELECT way FROM osm_polygon WHERE osm_id = -59195') " \
        #               "AS (id integer, way geometry)".format(points_of_houses_by_jes_sql.replace("'", "''"))
        result_sql = "SELECT p.jes, ST_AsGeoJSON(ST_Intersection(ST_Union(v.way), ST_Union(t.way))) FROM ({}) v, ({}) p, osm_polygon t " \
                     "WHERE ST_Contains(v.way, p.way) AND t.osm_id = -59195 " \
                     "GROUP BY p.jes".format(voronoi_sql, points_of_houses_by_jes_sql)

        cursor.execute(result_sql)

        import json
        json.dump({
            'type': 'FeatureCollection',
            'features': [{
                'type': 'Feature',
                'geometry': json.loads(geom),
                'properties': {
                    'jes': jes,
                },
             } for jes, geom in cursor.fetchall()]
        }, open('vorinoi_dump_new.geojson', 'w'), ensure_ascii=False)


if __name__ == '__main__':
    main()

jes_minsk_edit_v14.csv