Support junctions api

api/s2_junctions.py

@@ -0,0 +1,158 @@
+from psycopg.rows import dict_row, Row
+from connection import g_conn_dict as conn
+from s0_base import *
+from operation import *
+
+def add_junction(name: str, id: str, x: float, y: float, elevation: float) -> None:
+    if is_node(name, id):
+        return
+
+    with conn[name].cursor() as cur:
+        sql = f"insert into _node (id, type) values ('{id}', 'JUNCTION');"
+        sql += f" insert into junctions (id, elevation) values ('{id}', {elevation});"
+        sql += f" insert into coordinates (node, coord) values ('{id}', '({x}, {y})');"
+        cur.execute(sql)
+
+        redo = sql.replace("'", '"')
+        undo = f'delete from coordinates where node = "{id}";'
+        undo += f' delete from junctions where id = "{id}";'
+        undo += f' delete from _node where id = "{id}";'
+        add_operation(name, redo, undo)
+
+def delete_junction(name: str, id: str) -> None:
+    if not is_junction(name, id):
+        return
+
+    with conn[name].cursor(row_factory=dict_row) as cur:
+        cur.execute(f"select * from junctions where id = '{id}'")
+        row = cur.fetchone()
+        if row == None:
+            return
+
+        elevation = row['elevation']
+        demand = 'NULL' if row['demand'] == None else row['demand']
+        pattern = 'NULL' if row['pattern'] == None else row['pattern']
+        pattern = f'"{pattern}"' if pattern != 'NULL' else pattern
+
+        cur.execute(f"select * from coordinates where id = '{id}'")
+        row = cur.fetchone()
+        if row == None:
+            return
+
+        coord = row['coord']
+
+        sql = f"delete from coordinates where node = '{id}';"
+        sql += f" delete from junctions where id = '{id}';"
+        sql += f" delete from _node where id = '{id}';"
+        cur.execute(sql)
+
+        redo = sql.replace("'", '"')
+        undo = f'insert into _node (id, type) values ("{id}", "{type}");'
+        undo += f" insert into junctions (id, elevation, demand, pattern) values ('{id}', {elevation}, {demand}, {pattern});"
+        undo += f"  insert into coordinates (node, coord) values ('{id}', '{coord}');"
+        add_operation(name, redo, undo)
+
+def _get_junction(name: str, id: str) -> Row | None:
+    with conn[name].cursor(row_factory=dict_row) as cur:
+        cur.execute(f"select elevation, demand, pattern from junctions where id = '{id}'")
+        return cur.fetchone()
+
+def get_junction_elevation(name: str, id: str) -> float | None:
+    row = _get_junction(name, id)
+    return float(row['elevation']) if row != None else None
+
+def get_junction_demand(name: str, id: str) -> float | str | None:
+    row = _get_junction(name, id)
+    if row != None:
+        return float(row['demand']) if row['demand'] != None else 'NULL'
+    else:
+        return None
+
+def get_junction_pattern(name: str, id: str) -> str | None:
+    row = _get_junction(name, id)
+    if row != None:
+        return row['pattern'] if row['pattern'] != None else 'NULL'
+    else:
+        return None
+
+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_junction_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_junction_elevation(name: str, id: str, elevation: float) -> None:
+    if not is_junction(name, id):
+        return
+
+    old = get_junction_elevation(name, id)
+    if old == None:
+        return
+
+    with conn[name].cursor() as cur:
+        sql = f"update junctions set elevation = {elevation} where id = '{id}'"
+        cur.execute(sql)
+        redo = sql.replace("'", '"')
+        undo = f'update junctions set elevation = {old} where id = "{id}"'
+        add_operation(name, redo, undo)
+
+def set_junction_demand(name: str, id: str, demand: float) -> None:
+    if not is_junction(name, id):
+        return
+
+    old = get_junction_demand(name, id)
+    if old == None:
+        return
+
+    with conn[name].cursor() as cur:
+        sql = f"update junctions set demand = {demand} where id = '{id}'"
+        cur.execute(sql)
+        redo = sql.replace("'", '"')
+        undo = f'update junctions set demand = {old} where id = "{id}"'
+        add_operation(name, redo, undo)
+
+def set_junction_pattern(name: str, id: str, pattern: str) -> None:
+    if not is_junction(name, id):
+        return
+    if not is_pattern(name, id):
+        return
+
+    old = get_junction_pattern(name, id)
+    if old == None:
+        return
+
+    old = f'"{old}"' if old != 'NULL' else old
+
+    with conn[name].cursor() as cur:
+        sql = f"update junctions set pattern = '{pattern}' where id = '{id}'"
+        cur.execute(sql)
+        redo = sql.replace("'", '"')
+        undo = f'update junctions set pattern = {old} where id = "{id}"'
+        add_operation(name, redo, undo)
+
+def set_junction_coord(name: str, id: str, x: float, y: float) -> None:
+    if not is_junction(name, id):
+        return
+
+    old = get_junction_coord(name, id)
+    if old == None:
+        return
+    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)