44 lines
1.3 KiB
Python
44 lines
1.3 KiB
Python
from psycopg.rows import dict_row, Row
|
|
from .connection import g_conn_dict as conn
|
|
from .s0_base import *
|
|
from .operation import *
|
|
from .change_set import ChangeSet
|
|
|
|
|
|
def _to_point(coord: str) -> dict[str, float]:
|
|
coord = coord.removeprefix('(')
|
|
coord = coord.removesuffix(')')
|
|
coord = coord.split(',')
|
|
return { 'x': float(coord[0]), 'y': float(coord[1]) }
|
|
|
|
|
|
def get_node_coord(name: str, id: str) -> dict[str, float] | None:
|
|
with conn[name].cursor(row_factory=dict_row) as cur:
|
|
cur.execute(f"select * from coordinates where node = '{id}'")
|
|
row = cur.fetchone()
|
|
if row == None:
|
|
return None
|
|
|
|
coord = str(row['coord'])
|
|
return _to_point(coord)
|
|
|
|
|
|
def set_node_coord(name: str, node_type: str, id: str, x: float, y: float) -> ChangeSet:
|
|
old = get_node_coord(name, id)
|
|
if old == None:
|
|
return ChangeSet()
|
|
|
|
old_x, old_y = old['x'], old['y']
|
|
|
|
with conn[name].cursor() as cur:
|
|
sql = f"update coordinates set coord = '({x},{y})' where node = '{id}'"
|
|
cur.execute(sql)
|
|
|
|
redo = sql.replace("'", '"')
|
|
undo = f'update coordinates set coord = "({old_x},{old_y})" where node = "{id}"'
|
|
add_operation(name, redo, undo)
|
|
|
|
change = ChangeSet()
|
|
change.update(node_type, id, 'coord', 'point', str({'x': x, 'y': y}))
|
|
return change
|