1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
import os
import psycopg
import yaml
class IllegalCategoryException(Exception):
pass
with open(os.path.join('configs', 'database.yml'), 'r') as file:
db_con_params = yaml.safe_load(file.read())
with psycopg.connect(**db_con_params) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT subclass
FROM poi
WHERE subclass NOT LIKE '%;%'
GROUP BY class, subclass
HAVING COUNT(*) > 1
;
""")
categories = cur.fetchall()
categories = [category[0] for category in categories]
def get_chains() -> (list[dict]):
with psycopg.connect(**db_con_params, row_factory=psycopg.rows.dict_row) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT name, color FROM chain
""")
chains = cur.fetchall()
return chains
def get_all(category: str) -> (list[dict]):
if category not in categories:
raise IllegalCategoryException()
else:
with psycopg.connect(**db_con_params, row_factory=psycopg.rows.dict_row) as conn:
with conn.cursor() as cur:
cur.execute("""
WITH filtered AS (
SELECT osm_id, name, brand, geom, class, subclass
FROM poi
WHERE subclass = %(subclass)s
)
SELECT
filtered.osm_id,
filtered.name,
filtered.brand,
ST_Y(ST_Transform(filtered.geom, 4326)) AS lat,
ST_X(ST_Transform(filtered.geom, 4326)) AS long,
ST_AsGeoJSON(ST_Transform(polygon.geom, 4326)) as polygon
FROM filtered
JOIN
(
SELECT (ST_DUMP(ST_VoronoiPolygons(ST_Collect(geom)))).geom as geom
FROM filtered
) polygon ON ST_Contains(polygon.geom, filtered.geom)
;
""", {'subclass': category})
all = cur.fetchall()
return all
|
