from .database import read, read_all, write def from_postgis_polygon(polygon: str) -> list[tuple[float, float]]: boundary = polygon.lower().removeprefix('polygon((').removesuffix('))').split(',') xys = [] for pt in boundary: xy = pt.split(' ') xys.append((float(xy[0]), float(xy[1]))) return xys def to_postgis_polygon(boundary: list[tuple[float, float]]) -> str: polygon = '' for pt in boundary: polygon += f'{pt[0]} {pt[1]},' return f'polygon(({polygon[:-1]}))' def get_nodes_in_boundary(name: str, boundary: list[tuple[float, float]]) -> list[str]: api = 'get_nodes_in_boundary' write(name, f"delete from temp_region where id = '{api}'") write(name, f"insert into temp_region (id, boundary) values ('{api}', '{to_postgis_polygon(boundary)}')") nodes: list[str] = [] for row in read_all(name, f"select c.node from coordinates as c, temp_region as r where ST_Intersects(c.coord, r.boundary) and r.id = '{api}'"): nodes.append(row['node']) write(name, f"delete from temp_region where id = '{api}'") return nodes def get_nodes_in_region(name: str, id: str) -> list[str]: nodes: list[str] = [] for row in read_all(name, f"select c.node from coordinates as c, region as r where ST_Intersects(c.coord, r.boundary) and r.id = '{id}'"): nodes.append(row['node']) return nodes def calculate_convex_hull(name: str, nodes: list[str]) -> list[tuple[float, float]]: write(name, f'delete from temp_node') for node in nodes: write(name, f"insert into temp_node values ('{node}')") # TODO: check none polygon = read(name, f'select st_astext(st_convexhull(st_collect(array(select coord from coordinates where node in (select * from temp_node))))) as boundary' )['boundary'] write(name, f'delete from temp_node') return from_postgis_polygon(polygon)