93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
from .database import *
|
|
from .s0_base import get_link_nodes
|
|
|
|
def sql_update_coord(node: str, x: float, y: float) -> str:
|
|
coord = f"st_geomfromtext('point({x} {y})')"
|
|
return str(f"update coordinates set coord = {coord} where node = '{node}';")
|
|
|
|
|
|
def sql_insert_coord(node: str, x: float, y: float) -> str:
|
|
coord = f"st_geomfromtext('point({x} {y})')"
|
|
return str(f"insert into coordinates (node, coord) values ('{node}', {coord});")
|
|
|
|
|
|
def sql_delete_coord(node: str) -> str:
|
|
return str(f"delete from coordinates where node = '{node}';")
|
|
|
|
|
|
def from_postgis_point(coord: str) -> dict[str, float]:
|
|
xy = coord.lower().removeprefix('point(').removesuffix(')').split(' ')
|
|
return { 'x': float(xy[0]), 'y': float(xy[1]) }
|
|
|
|
|
|
def get_node_coord(name: str, node: str) -> dict[str, float]:
|
|
row = try_read(name, f"select st_astext(coord) as coord_geom from coordinates where node = '{node}'")
|
|
if row == None:
|
|
write(name, sql_insert_coord(node, 0.0, 0.0))
|
|
return {'x': 0.0, 'y': 0.0}
|
|
return from_postgis_point(row['coord_geom'])
|
|
|
|
# DingZQ 2025-01-03, get nodes in extent
|
|
# return node id list
|
|
# node_id:junction:x:y
|
|
def get_nodes_in_extent(name: str, x1: float, y1: float, x2: float, y2: float) -> list[str]:
|
|
nodes = []
|
|
objs = read_all(name, 'select node, st_astext(coord) as coord_geom from coordinates')
|
|
for obj in objs:
|
|
node_id = obj['node']
|
|
coord = from_postgis_point(obj['coord_geom'])
|
|
x = coord['x']
|
|
y = coord['y']
|
|
if x1 <= x <= x2 and y1 <= y <= y2:
|
|
nodes.append(f"{node_id}:{x}:{y}")
|
|
return nodes
|
|
|
|
# DingZQ 2025-01-03, get links in extent
|
|
# return link id list
|
|
# link_id:pipe:node_id1:node_id2
|
|
def get_links_in_extent(name: str, x1: float, y1: float, x2: float, y2: float) -> list[str]:
|
|
node_ids = set([s.split(':')[0] for s in get_nodes_in_extent(name, x1, y1, x2, y2)])
|
|
|
|
all_link_ids = []
|
|
with conn[name].cursor(row_factory=dict_row) as cur:
|
|
cur.execute(f"select id from pipes")
|
|
for record in cur:
|
|
all_link_ids.append(record['id'])
|
|
|
|
links = []
|
|
for link_id in all_link_ids:
|
|
nodes = get_link_nodes(name, link_id)
|
|
if nodes[0] in node_ids and nodes[1] in node_ids:
|
|
links.append(f"{link_id}:pipe:{nodes[0]}:{nodes[1]}")
|
|
return links
|
|
|
|
|
|
def node_has_coord(name: str, node: str) -> bool:
|
|
return try_read(name, f"select node from coordinates where node = '{node}'") != None
|
|
|
|
|
|
#--------------------------------------------------------------
|
|
# [EPA2][EPA3][IN][OUT]
|
|
# id x y
|
|
#--------------------------------------------------------------
|
|
# exception ! need merge to node change set !
|
|
|
|
|
|
def inp_in_coord(line: str) -> str:
|
|
tokens = line.split()
|
|
node = tokens[0]
|
|
coord = f"st_geomfromtext('point({tokens[1]} {tokens[2]})')"
|
|
return str(f"insert into coordinates (node, coord) values ('{node}', {coord});")
|
|
|
|
|
|
def inp_out_coord(name: str) -> list[str]:
|
|
lines = []
|
|
objs = read_all(name, 'select node, st_astext(coord) as coord_geom from coordinates')
|
|
for obj in objs:
|
|
node = obj['node']
|
|
coord = from_postgis_point(obj['coord_geom'])
|
|
x = coord['x']
|
|
y = coord['y']
|
|
lines.append(f'{node} {x} {y}')
|
|
return lines
|